import hashlib
import hmac
import json
import random
import time

from fastapi import HTTPException
from sqlalchemy import func, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.plugins.pos_operations.models import PosStaff, PosStaffRevision, StorePosRules
from app.plugins.pos_operations.pin import hash_pin


async def lane_from_token(db: AsyncSession, authorization: str | None):
    """Bearer 令牌 -> lane，同时做插件开关检查。路由在启动时无条件挂载，
    所以关停插件必须在这里生效，否则「关掉」只是界面上的开关。

    两个 Agent 侧路由文件（router.py、returns_router.py）共用同一份解析逻辑，
    放在这里而不是任一路由文件里，避免 router 互相 import。"""
    from app.core.services.plugin_helper import require_plugin
    from app.plugins.pos_sync.services import authenticate_lane_by_token
    if not authorization or not authorization.lower().startswith("bearer "):
        raise HTTPException(status_code=401, detail="missing POS sync token")
    lane = await authenticate_lane_by_token(db, authorization.split(" ", 1)[1].strip())
    await require_plugin("pos_operations", db, lane.tenant_id)
    return lane


def _canonical_payload_hash(payload: dict) -> str:
    encoded = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def return_approval_content_hash(body) -> str:
    data = body.model_dump() if hasattr(body, "model_dump") else dict(body)
    normalized_items = []
    for item in data.get("items") or []:
        item = item.model_dump() if hasattr(item, "model_dump") else dict(item)
        normalized_items.append({
            "orderItemId": item.get("orderItemId"),
            "productId": item.get("productId"),
            "variantId": item.get("variantId"),
            "name": item.get("name"),
            "quantity": str(item.get("quantity")),
            "unitPriceCents": item.get("unitPriceCents"),
        })
    return _canonical_payload_hash({
        "sourceType": data.get("sourceType", "receipt"),
        "orderNo": data.get("orderNo"),
        "customerId": data.get("customerId"),
        "items": normalized_items,
        "reason": data.get("reason"),
        "overrideReason": data.get("overrideReason"),
        "stockDisposition": data.get("stockDisposition", "resellable"),
    })


def verify_approval_evidence(
    evidence: dict | None, *, sync_token: str, action_type: str = "return",
    tenant_id: int, store_id: int, lane_id: str,
    operator_user_id: int, approver_user_id: int,
    content_hash: str, idempotency_key: str | None = None,
    now_ms: int | None = None,
) -> bool:
    """Verify short-lived Agent evidence without sending an approver PIN upstream."""
    if not isinstance(evidence, dict) or not sync_token:
        return False
    signature = evidence.get("signature")
    if not isinstance(signature, str):
        return False
    claims = {key: value for key, value in evidence.items() if key != "signature"}
    expected = {
        "actionType": action_type,
        "tenantId": tenant_id,
        "storeId": store_id,
        "laneId": lane_id,
        "operatorUserId": operator_user_id,
        "approverUserId": approver_user_id,
        "contentHash": content_hash,
    }
    if idempotency_key is not None:
        expected["idempotencyKey"] = idempotency_key
    if any(claims.get(key) != value for key, value in expected.items()):
        return False
    expires_at = claims.get("expiresAt")
    current_ms = int(time.time() * 1000) if now_ms is None else now_ms
    if not isinstance(expires_at, int) or current_ms > expires_at or expires_at > current_ms + 120_000:
        return False
    digest = _canonical_payload_hash(claims)
    expected_signature = hmac.new(
        sync_token.encode("utf-8"), digest.encode("ascii"), hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(signature, expected_signature)


async def _get_staff(db: AsyncSession, tenant_id: int, user_id: int) -> PosStaff | None:
    return (await db.execute(
        select(PosStaff).where(PosStaff.tenant_id == tenant_id, PosStaff.user_id == user_id)
    )).scalar_one_or_none()


class StaffInvalid(ValueError):
    """员工配置不合法。"""


async def bump_staff_revision(db, tenant_id: int) -> int:
    """员工发生任何变更就推进一次修订号。

    必须单调：Agent 只在 revision 变大时才替换本地快照，一旦倒退（例如删掉持有最高
    credential_version 的员工），快照就永久卡住，被停用的收银员仍能登录。
    """
    row = (await db.execute(
        select(PosStaffRevision).where(PosStaffRevision.tenant_id == tenant_id).with_for_update()
    )).scalar_one_or_none()
    if row is None:
        row = PosStaffRevision(tenant_id=tenant_id, revision=1)
        db.add(row)
        await db.flush()
    else:
        row.revision += 1
    return row.revision


async def current_staff_revision(db, tenant_id: int) -> int:
    row = (await db.execute(
        select(PosStaffRevision).where(PosStaffRevision.tenant_id == tenant_id)
    )).scalar_one_or_none()
    return row.revision if row else 0


def _generate_pin() -> str:
    return str(random.randint(100000, 999999))


async def upsert_staff(db, *, tenant_id, user_id, pos_enabled, store_ids, pin) -> tuple[PosStaff, str | None]:
    from app.core.models.user import User

    # 用户必须真实存在于本租户，否则会建出一个指向空账号的收银员。
    user = (await db.execute(
        select(User).where(
            User.id == user_id, User.tenant_id == tenant_id, User.is_active == 1,
        )
    )).scalar_one_or_none()
    if user is None:
        raise StaffInvalid("用户不存在或不属于本租户")
    if getattr(user, "is_active", 1) in (0, False):
        raise StaffInvalid("该用户已停用，不能启用为 POS 收银员")

    staff = await _get_staff(db, tenant_id, user_id)
    if staff is None:
        staff = PosStaff(tenant_id=tenant_id, user_id=user_id, credential_version=1)
        db.add(staff)
    else:
        staff.credential_version += 1
    staff.pos_enabled = 1 if pos_enabled else 0
    staff.store_ids = list(store_ids or [])
    generated_pin = None
    if pin is not None:
        staff.pin_hash = hash_pin(pin)
    elif pos_enabled and not staff.pin_hash:
        generated_pin = _generate_pin()
        staff.pin_hash = hash_pin(generated_pin)

    # 启用却没有 PIN 的收银员永远登录不了，只会在门店造成一次排查。
    if staff.pos_enabled and not staff.pin_hash:
        raise StaffInvalid("启用 POS 前必须设置 PIN")

    await bump_staff_revision(db, tenant_id)
    await db.commit()
    await db.refresh(staff)
    return staff, generated_pin


async def reset_pin(db, *, tenant_id, user_id, pin) -> tuple[PosStaff, str]:
    staff = await _get_staff(db, tenant_id, user_id)
    if staff is None:
        raise ValueError("staff not found")
    pin_value = pin if pin is not None else _generate_pin()
    staff.pin_hash = hash_pin(pin_value)
    staff.credential_version += 1
    staff.failed_attempts = 0
    staff.locked_until = None
    await bump_staff_revision(db, tenant_id)
    await db.commit()
    await db.refresh(staff)
    return staff, pin_value


async def list_staff(db, tenant_id: int) -> list[PosStaff]:
    return list((await db.execute(
        select(PosStaff).where(PosStaff.tenant_id == tenant_id).order_by(PosStaff.id.asc())
    )).scalars().all())


class RulesInvalid(ValueError):
    pass


_ALLOWED_REFUND_METHODS = {"store_credit"}


async def get_rules(db, *, tenant_id: int, store_id: int) -> StorePosRules:
    existing = (await db.execute(
        select(StorePosRules).where(
            StorePosRules.tenant_id == tenant_id, StorePosRules.store_id == store_id)
    )).scalar_one_or_none()
    if existing is not None:
        return existing
    # Unsaved instance: SQLAlchemy column defaults only apply on flush/insert,
    # so spell them out explicitly here rather than relying on the model.
    return StorePosRules(
        tenant_id=tenant_id, store_id=store_id,
        minimum_age_default=18, photo_id_required=1, age_second_approval_required=1,
        returns_online_only=1, return_window_days=30, receiptless_returns_enabled=1,
        receiptless_refund_method="store_credit", all_returns_require_second_approval=1,
        refund_original_tender=1,
        price_override_approval_required=1,
        receipt_print_enabled=0, receipt_auto_print=0, receipt_print_eftpos_slip=0,
        cash_drawer_kick_on_print=0, receipt_paper_width="80mm",
        rules_version=1,
    )


# 白名单：字段名 -> 校验器。不在表里的键一律拒绝，而不是静默忽略——
# 静默忽略会让管理员以为自己改了某项规则，实际上没有。
_BOOL_RULES = (
    "photo_id_required", "age_second_approval_required", "returns_online_only",
    "receiptless_returns_enabled", "all_returns_require_second_approval",
    "refund_original_tender",
    "price_override_approval_required",
    # 小票与钱柜开关：可自由开关，不属于 mandatory-true 安全开关。
    "receipt_print_enabled", "receipt_auto_print", "receipt_print_eftpos_slip",
    "cash_drawer_kick_on_print",
)
_ALLOWED_PAPER_WIDTHS = {"58mm", "80mm"}
_MANDATORY_TRUE_RULES = {
    "returns_online_only",
    "age_second_approval_required", "refund_original_tender",
}


def _as_bool_flag(field: str, value) -> int:
    # 只接受真正的布尔/0/1。字符串 "no"/"false" 是真值，会把安全开关误开成启用。
    if isinstance(value, bool):
        return 1 if value else 0
    if isinstance(value, int) and value in (0, 1):
        return value
    raise RulesInvalid(f"{field} 必须是布尔值")


def validate_rules_payload(data: dict) -> dict:
    """把任意 dict 收敛成一组已校验的字段。这些开关是所有其它防护的总闸。"""
    if not isinstance(data, dict):
        raise RulesInvalid("规则数据格式错误")

    cleaned: dict = {}
    for field, value in data.items():
        if field in ("id", "tenant_id", "store_id", "rules_version",
                     "created_at", "updated_at"):
            continue                       # 由服务端管理，忽略而不报错
        if field in _BOOL_RULES:
            cleaned[field] = _as_bool_flag(field, value)
            if field in _MANDATORY_TRUE_RULES and cleaned[field] == 0:
                raise RulesInvalid(f"{field} cannot be disabled")
        elif field == "return_window_days":
            if not isinstance(value, int) or isinstance(value, bool) or value < 1 or value > 3650:
                raise RulesInvalid("return_window_days 必须是 1..3650 的整数")
            cleaned[field] = value
        elif field == "minimum_age_default":
            if not isinstance(value, int) or isinstance(value, bool) or not 0 <= value <= 120:
                raise RulesInvalid("minimum_age_default 必须是 0..120 的整数")
            cleaned[field] = value
        elif field == "receiptless_refund_method":
            if value not in _ALLOWED_REFUND_METHODS:
                raise RulesInvalid("unsupported receiptless_refund_method")
            cleaned[field] = value
        elif field == "receipt_paper_width":
            if value not in _ALLOWED_PAPER_WIDTHS:
                raise RulesInvalid("receipt_paper_width 必须是 58mm 或 80mm")
            cleaned[field] = value
        else:
            raise RulesInvalid(f"未知的规则字段：{field}")
    return cleaned


async def upsert_rules(db, *, tenant_id: int, store_id: int, data: dict,
                       actor_id: int | None = None) -> StorePosRules:
    cleaned = validate_rules_payload(data)

    rules = (await db.execute(
        select(StorePosRules).where(
            StorePosRules.tenant_id == tenant_id, StorePosRules.store_id == store_id)
    )).scalar_one_or_none()
    if rules is None:
        rules = StorePosRules(tenant_id=tenant_id, store_id=store_id, rules_version=1)
        db.add(rules)
    else:
        rules.rules_version += 1

    # 关停安全开关是高风险动作，必须留痕，否则事后无法追溯是谁在什么时候放开的。
    relaxed = {f: cleaned[f] for f in _BOOL_RULES if f in cleaned and cleaned[f] == 0}
    before = {f: getattr(rules, f, None) for f in cleaned}

    for field, value in cleaned.items():
        setattr(rules, field, value)

    if relaxed:
        from app.services.audit import log_audit
        await log_audit(
            db, tenant_id=tenant_id, action="pos.rules.relaxed",
            actor_type="admin", actor_id=actor_id,
            target_type="store_pos_rules", target_id=store_id,
            changes={"relaxed": relaxed, "before": before},
        )

    await db.commit()
    await db.refresh(rules)
    return rules


async def list_stores(db, tenant_id: int) -> list[dict]:
    """本租户的门店列表。

    系统里没有独立的 stores 表——门店身份只存在于 pos_lanes 上（superadmin 手工录入）。
    因此这里按 pos_lanes 聚合去重。副作用：还没建过收银机的门店不会出现。
    """
    from app.plugins.pos_sync.models import PosLane

    rows = (await db.execute(
        select(PosLane.store_id, PosLane.store_code, func.count(PosLane.id))
        .where(PosLane.tenant_id == tenant_id)
        .group_by(PosLane.store_id, PosLane.store_code)
        .order_by(PosLane.store_id.asc())
    )).all()
    return [
        {"storeId": store_id, "storeCode": store_code or str(store_id), "laneCount": lane_count}
        for store_id, store_code, lane_count in rows
    ]


async def list_assignable_users(db, tenant_id: int,
                                resolve_permissions=None) -> list[dict]:
    """可设为 POS 收银员的后台用户。

    收银员必须先是本租户的后台用户；权限来自该用户的角色，不在收银员表单里设置。
    因此这里同时返回 hasPosPermission，让管理员在下拉里就能看出「配了也用不了」的账号
    ——否则要等收银员在收银机上被 403 才发现。

    ponytail: 逐个用户解析权限是 N+1；后台用户通常只有几十个，先这样。
    真的慢了再改成一次性批量取角色权限。
    """
    from app.api.routers.admin.permissions import _name_from_profile
    from app.core.models.user import User

    resolve_permissions = resolve_permissions or resolve_pos_permissions
    users = list((await db.execute(
        select(User).where(User.tenant_id == tenant_id).order_by(User.id.asc())
    )).scalars().all())

    staff_ids = {
        row.user_id for row in (await db.execute(
            select(PosStaff).where(PosStaff.tenant_id == tenant_id)
        )).scalars().all()
    }

    out = []
    for user in users:
        if not getattr(user, "is_active", 1):
            continue                      # 停用账号不能设为收银员
        perms = await resolve_permissions(db, tenant_id, user.id)
        out.append({
            "id": user.id,
            # User 表没有 name 列——姓名在 profile JSON 里，缺失时回落到邮箱前缀。
            # 复用后台用户列表已有的解析逻辑，它还兼容 profile 被存成字符串的情况。
            "name": _name_from_profile(user),
            "email": getattr(user, "email", None),
            "hasPosPermission": bool(perms),
            "alreadyStaff": user.id in staff_ids,
        })
    return out


async def resolve_pos_permissions(db, tenant_id: int, user_id: int) -> list[str]:
    """The staff member's effective ``pos.*`` permission keys. The Agent needs these to
    check approval authority (e.g. pos.age_approve) offline."""
    from app.core.models.user import User
    from app.core.services.permission_service import PermissionService

    user = (await db.execute(
        select(User).where(User.id == user_id, User.tenant_id == tenant_id)
    )).scalar_one_or_none()
    if user is None or not user.is_active:
        return []
    effective = await PermissionService(db).get_effective_permissions(user)
    return sorted(k for k in effective.permissions if k.startswith("pos."))


async def verify_pos_actor(
    db, *, tenant_id: int, store_id: int, user_id: int | None, permission: str,
    pin: str | None = None, resolve_permissions=resolve_pos_permissions,
) -> list[str]:
    """服务端校验操作人/审批人。

    Agent 侧已经用 PIN 验过一次，但那是客户端；云端不能只相信请求里传来的两个数字
    ID 不相等，否则任何持有 Lane Token 的人都能伪造「第二人已批准」。
    """
    if not user_id:
        raise PosActorInvalid("缺少操作人")
    staff = await _get_staff(db, tenant_id, user_id)
    if staff is None or not staff.pos_enabled:
        raise PosActorInvalid(f"用户 {user_id} 不是启用状态的 POS 员工")
    if not staff.pin_hash:
        raise PosActorInvalid(f"用户 {user_id} 未设置 PIN")
    scope = staff.store_ids or []
    if scope and store_id not in scope:
        raise PosActorInvalid(f"用户 {user_id} 无此门店权限")
    # 审批人必须当场证明自己输入过 PIN。只核对「这个 ID 有权限」的话，
    # 任何持有 Lane Token 的人只要知道两个员工 ID 就能伪造一次双人批准。
    if pin is not None:
        from app.plugins.pos_operations.pin import verify_pin
        if not verify_pin(pin, staff.pin_hash):
            raise PosActorInvalid(f"用户 {user_id} 的 PIN 不正确")

    perms = await resolve_permissions(db, tenant_id, user_id)
    if permission not in perms:
        raise PosActorInvalid(f"用户 {user_id} 缺少权限 {permission}")
    return perms


class PosActorInvalid(ValueError):
    """操作人/审批人校验失败。"""


async def _pos_user_active(db, tenant_id: int, user_id: int) -> bool:
    from app.core.models.user import User
    return (await db.execute(
        select(User.id).where(
            User.id == user_id, User.tenant_id == tenant_id, User.is_active == 1,
        )
    )).scalar_one_or_none() is not None


async def build_receipt_template_snapshot(db, *, tenant_id: int, store_id: int) -> dict:
    """受信小票模板 + 版本 + 纸宽，供 Agent 缓存。

    template_version = 内容 sha256（截断），管理员改模板即变。无模板时 content=None，
    Agent 用内置 fallback。
    """
    from app.core.models.print_template import PrintTemplate

    tpl = (await db.execute(
        select(PrintTemplate).where(
            PrintTemplate.tenant_id == tenant_id, PrintTemplate.template_key == "receipt")
    )).scalar_one_or_none()
    rules = await get_rules(db, tenant_id=tenant_id, store_id=store_id)
    row = (await db.execute(text(
        "SELECT is_active, config FROM plugin_configs "
        "WHERE tenant_id = :tenant_id AND plugin_name = 'i18n_multilang' LIMIT 1"
    ), {"tenant_id": tenant_id})).first()
    i18n_enabled, default_locale = True, 'zh'
    if row is not None:
        raw_config = row[1] if isinstance(row[1], dict) else json.loads(row[1] or '{}')
        i18n_enabled = bool(row[0])
        default_locale = 'en' if raw_config.get('default_locale') == 'en' else 'zh'
    content = tpl.content if tpl else None
    version = hashlib.sha256(content.encode("utf-8")).hexdigest()[:16] if content else ""
    return {
        "templateKey": "receipt" if content else "fallback",
        "content": content,
        "templateVersion": version,
        "paperWidth": rules.receipt_paper_width,
        # 小票行为开关随模板同通道下发，Agent 缓存后离线也能决定打不打、弹不弹柜。
        "printEnabled": bool(rules.receipt_print_enabled),
        "autoPrint": bool(rules.receipt_auto_print),
        "printEftposSlip": bool(rules.receipt_print_eftpos_slip),
        "kickDrawerOnPrint": bool(rules.cash_drawer_kick_on_print),
        "rulesVersion": rules.rules_version,
        "multilangEnabled": i18n_enabled,
        "defaultLocale": default_locale,
        "priceOverrideApprovalRequired": bool(rules.price_override_approval_required),
        "allReturnsRequireSecondApproval": bool(rules.all_returns_require_second_approval),
    }


async def build_staff_snapshot(
    db, *, tenant_id: int, resolve_permissions=resolve_pos_permissions,
    resolve_user_active=_pos_user_active,
) -> dict:
    rows = list((await db.execute(
        select(PosStaff).where(PosStaff.tenant_id == tenant_id).order_by(PosStaff.id.asc())
    )).scalars().all())
    staff = []
    for r in rows:
        active = await resolve_user_active(db, tenant_id, r.user_id)
        staff.append({
            "userId": r.user_id,
            "posEnabled": bool(r.pos_enabled) and active,
            "storeIds": r.store_ids or [],
            "pinHash": r.pin_hash,
            "credentialVersion": r.credential_version,
            "permissions": await resolve_permissions(db, tenant_id, r.user_id) if active else [],
        })
    # 单调修订号，与员工数量/凭据版本无关，删除员工也不会让它倒退。
    revision = await current_staff_revision(db, tenant_id)
    return {"staff": staff, "revision": revision}