"""账号池：候选筛选、两梯队排序、供给预估、租约领取与释放。

配套文档：`docs/设计-账号自动分配与回收.md`（§3.3 候选与排序、§3.4 补给链、
§3.5 原子领取、§3.9 自动收回）。

## 为什么要拆「纯函数」与「SQL」

排序与充值量计算是**业务规则的集中地**（先花快过期的、先用小余额、
缺多少补多少），也是最容易被改错、必须能被反向验证的部分 ——
所以它们是不碰数据库的纯函数，用例可以脱离 MySQL 直接喂数据。

SQL 只负责**筛选**（谁的 CK 有效、谁已开通、谁 365 天内没兑过、谁没被租走）。

## 两个必须记住的约束

1. **`elite_opened = 1` 是硬门槛**：开通精英宝典要人工扫码
   （`create_elite_qr` 生成支付码），系统不能自动开通 —— 未开通的账号进池也白进。
2. **并发抢号靠数据库**：`account_leases.active_account_id` 上的唯一索引裁决胜负，
   撞键就换下一个候选（SAVEPOINT 包住，不影响已经领到的那些）。
   不要退回「先查再插」——那在并发下必然双发。
"""

from __future__ import annotations

from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import and_, func, not_, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from ..models import Account, AccountEvent, AccountLease, DouyuTask, User
from .account_binding import SCOPE_ELITE, SCOPE_ESPORTS, SCOPE_PEACE

#: 租约默认有效期（分钟）。到点由 `account_lease_reaper` 兜底回收。
DEFAULT_LEASE_TTL_MINUTES = 60

#: 同一件皮肤多久内不允许在同一账号再兑（滚动天数）。
EXCHANGE_COOLDOWN_DAYS = 365

#: 1 个精英令 = 10 精英积分 = 1 鱼翅 = 1 元（2026-10-07 生产实测）。
POINTS_PER_GIFT = 10

#: 单次鱼翅充值上限（元）。见设计 §3.4.2。
MAX_RECHARGE_AMOUNT = 800

#: 池成员 = 未分配 ∪ 管理员/运营名下；客服名下账号不进池（不跨归属）。
_POOL_OWNER_ROLES = ("super_admin", "operation")

#: 作用域 → （积分列名，开通标记列名）。和平小店没有「开通」概念。
_SCOPE_COLUMNS: dict[str, tuple[str, str | None]] = {
    SCOPE_ELITE: ("points", "elite_opened"),
    SCOPE_ESPORTS: ("esports_points", "esports_opened"),
    SCOPE_PEACE: ("xpd_balance", None),
}

SUPPORTED_POOL_SCOPES = tuple(_SCOPE_COLUMNS)


def _now(now: datetime | None = None) -> datetime:
    return now or datetime.now(UTC)


@dataclass(frozen=True)
class Candidate:
    """一个可分配账号在**本次兑换**下的画像。"""

    account_id: int
    username: str
    points: int
    score: int
    tier: str  # "A" = 积分已够（免充值）；"B" = 需要充值
    topup_gifts: int
    points_expire_at: datetime | None = None
    role_change_available_at: datetime | None = None

    @property
    def topup_yuan(self) -> int:
        """需要充多少元（1 元 = 1 鱼翅 = 1 精英令）。"""
        return self.topup_gifts

    @property
    def needs_topup(self) -> bool:
        return self.topup_gifts > 0


def topup_gifts_for(points: int | None, score: int) -> int:
    """凑够 `score` 还差几个精英令（向下取整的缺口向上取整到整数个令）。

    >>> topup_gifts_for(300, 1280)
    98
    >>> topup_gifts_for(1280, 1280)
    0
    >>> topup_gifts_for(None, 20)
    2
    """
    current = int(points or 0)
    missing = max(0, int(score) - current)
    return (missing + POINTS_PER_GIFT - 1) // POINTS_PER_GIFT


def _sort_key(candidate: Candidate):
    """两梯队排序键（文档 §3.3）。

    A 梯队（免充值）：先花**快过期**的 → 再用**小余额**（把大额账号留给贵商品）。
    B 梯队（需充值）：先选**充值最少**的 → 再用**大余额**（留存量的先用掉）。

    `points_expire_at` 为空排最后；最后统一用 `account_id` 兜稳定顺序，
    保证同样的输入永远得到同样的顺序（否则用例无法断言）。
    """
    if candidate.tier == "A":
        expire_missing = candidate.points_expire_at is None
        expire_key = candidate.points_expire_at or datetime.max.replace(tzinfo=UTC)
        return (0, expire_missing, expire_key, candidate.points, candidate.account_id)
    return (
        1,
        candidate.topup_gifts,
        -candidate.points,
        candidate.account_id,
    )


def make_candidate(
    *,
    account_id: int,
    username: str,
    points: int | None,
    score: int,
    points_expire_at: datetime | None = None,
    role_change_available_at: datetime | None = None,
) -> Candidate:
    """由账号行构造候选（纯函数，便于用例直接喂数据）。"""
    current = int(points or 0)
    gifts = topup_gifts_for(current, score)
    return Candidate(
        account_id=account_id,
        username=username,
        points=current,
        score=int(score),
        tier="A" if gifts == 0 else "B",
        topup_gifts=gifts,
        points_expire_at=points_expire_at,
        role_change_available_at=role_change_available_at,
    )


def rank_candidates(candidates: Iterable[Candidate]) -> list[Candidate]:
    """按两梯队规则排序（纯函数）。"""
    return sorted(candidates, key=_sort_key)


def latest_exchange_exists_clause(commodity_id: str, since: datetime):
    """「该账号在 `since` 之后兑过这件商品」的子查询条件。"""
    return (
        select(AccountEvent.id)
        .where(
            AccountEvent.account_id == Account.id,
            AccountEvent.event_type == "exchange",
            # 只按 subject_key 判：D1 回填后我方任务与官方补录都是商品 ID。
            # 不额外约束 subject_type —— 部分回填的历史行 subject_type 仍是 goods，
            # 按它过滤会漏判（这正是 D1 要修的东西）。
            AccountEvent.subject_key == commodity_id,
            AccountEvent.occurred_at >= since,
        )
        .exists()
    )


def active_lease_exists_clause():
    """「该账号存在有效租约」的子查询条件。"""
    return (
        select(AccountLease.id)
        .where(AccountLease.active_account_id == Account.id)
        .exists()
    )


def pool_membership_clause():
    """池成员：未分配 ∪ 管理员/运营名下（客服名下账号不进池）。"""
    owner_ids = select(User.id).where(User.role.in_(_POOL_OWNER_ROLES))
    return or_(Account.assigned_to.is_(None), Account.assigned_to.in_(owner_ids))


def candidate_query(
    *,
    scope: str,
    score: int,
    commodity_id: str,
    now: datetime | None = None,
):
    """候选账号的基础查询（**未排序、未限量**，供计数与取行共用）。"""
    if scope not in _SCOPE_COLUMNS:
        raise ValueError(f"暂不支持的作用域: {scope}")
    points_column_name, opened_column_name = _SCOPE_COLUMNS[scope]
    points_column = getattr(Account, points_column_name)
    since = _now(now) - timedelta(days=EXCHANGE_COOLDOWN_DAYS)

    conditions: list[Any] = [
        Account.status == "active",
        Account.ck_check_status == "valid",
        Account.cookie.isnot(None),
        Account.cookie != "",
        Account.auto_managed.is_(True),
        pool_membership_clause(),
        not_(active_lease_exists_clause()),
        not_(latest_exchange_exists_clause(commodity_id, since)),
    ]
    if opened_column_name:
        # 未开通的账号系统开不了（要人工扫码），直接排除。
        # `NULL` = 没查过，同样排除 —— 不能把「不知道」当「已开通」放行。
        conditions.append(getattr(Account, opened_column_name).is_(True))
    # 积分未知（没查过）时按 0 处理会把充值量算错，先排除。
    conditions.append(points_column.isnot(None))
    return select(Account).where(and_(*conditions)), points_column


def eligible_accounts(
    db: Session,
    *,
    scope: str = SCOPE_ELITE,
    score: int,
    commodity_id: str,
    now: datetime | None = None,
    limit: int | None = None,
) -> list[Candidate]:
    """筛出可分配账号并按两梯队排序。

    `limit` 是为了「要 5 个」时不用把整个池拉出来 —— 但**排序键里有 Python 侧
    的表达式**（到期时间、充值量），所以 SQL 只做粗筛 + 有界取行，再在 Python 里排。
    取值上限给足余量（默认 `limit*5`，至少 200），否则「积分够的账号排在后面」
    会被截断掉。
    """
    query, points_column = candidate_query(
        scope=scope, score=score, commodity_id=commodity_id, now=now
    )
    query = query.order_by(points_column.desc(), Account.id.asc())
    if limit is not None:
        query = query.limit(max(200, int(limit) * 5))
    rows = db.execute(query).scalars().all()
    candidates = [
        make_candidate(
            account_id=row.id,
            username=row.username,
            points=getattr(row, _SCOPE_COLUMNS[scope][0]),
            score=score,
            points_expire_at=row.points_expire_at,
            role_change_available_at=row.role_change_available_at,
        )
        for row in rows
    ]
    return rank_candidates(candidates)


def estimate_supply(
    db: Session,
    *,
    scope: str = SCOPE_ELITE,
    score: int,
    commodity_id: str,
    now: datetime | None = None,
) -> dict[str, Any]:
    """不做任何锁定，预估「现在能分出几个、要花多少钱」（文档 §3.8）。

    返回的四个数在写死的口径下**互斥且可加**：
    `zero_topup + (after_topup - zero_topup) + blocked 三项 = 池外无关账号` 不计。
    """
    if scope not in _SCOPE_COLUMNS:
        raise ValueError(f"暂不支持的作用域: {scope}")
    points_column_name, opened_column_name = _SCOPE_COLUMNS[scope]
    points_column = getattr(Account, points_column_name)
    opened_column = getattr(Account, opened_column_name) if opened_column_name else None
    since = _now(now) - timedelta(days=EXCHANGE_COOLDOWN_DAYS)

    valid_base = [
        Account.status == "active",
        Account.ck_check_status == "valid",
        Account.cookie.isnot(None),
        Account.cookie != "",
        Account.auto_managed.is_(True),
        pool_membership_clause(),
    ]
    not_redeemed = not_(latest_exchange_exists_clause(commodity_id, since))

    def count(*extra) -> int:
        return int(
            db.execute(
                select(func.count(Account.id)).where(and_(*valid_base, *extra))
            ).scalar()
            or 0
        )

    opened_ok = (
        [opened_column.is_(True), points_column.isnot(None)]
        if opened_column is not None
        else [points_column.isnot(None)]
    )
    eligible = count(*opened_ok, not_redeemed, not_(active_lease_exists_clause()))
    zero_topup = count(
        *opened_ok,
        not_redeemed,
        not_(active_lease_exists_clause()),
        points_column >= score,
    )
    # 已兑过（含仍在冷却期）的账号里，只看「手册已开通」的那部分 ——
    # 未开通的账号无论如何都用不了，混进这个数会让人误以为是"冷却挡住了"。
    redeemed = count(*opened_ok, latest_exchange_exists_clause(commodity_id, since))
    not_opened = (
        count(points_column.isnot(None), not_(opened_column.is_(True)))
        if opened_column is not None
        else 0
    )
    leased = count(*opened_ok, not_redeemed, active_lease_exists_clause())

    gap_expr = func.greatest(0, score - func.coalesce(points_column, 0))
    total_gap = int(
        db.execute(
            select(func.coalesce(func.sum(gap_expr), 0)).where(
                and_(
                    *valid_base,
                    *opened_ok,
                    not_redeemed,
                    not_(active_lease_exists_clause()),
                )
            )
        ).scalar()
        or 0
    )
    est_gifts = (total_gap + POINTS_PER_GIFT - 1) // POINTS_PER_GIFT

    return {
        "scope": scope,
        "commodity_id": commodity_id,
        "score": int(score),
        "zero_topup": zero_topup,
        "after_topup": eligible,
        "leased": leased,
        "blocked": {
            f"{scope}_not_opened": not_opened,
            "redeemed_within_cooldown": redeemed,
        },
        "cost": {"est_gifts": est_gifts, "est_yuan": est_gifts},
        "cooldown_days": EXCHANGE_COOLDOWN_DAYS,
    }


def active_lease_for_account(db: Session, account_id: int) -> AccountLease | None:
    return (
        db.query(AccountLease)
        .filter(
            AccountLease.account_id == account_id,
            AccountLease.status == "active",
        )
        .first()
    )


def leased_account_ids(
    db: Session, *, holder_user_id: int, include_expired: bool = False
) -> list[int]:
    """当前用户持有的有效租约对应的账号 ID。"""
    query = db.query(AccountLease.account_id).filter(
        AccountLease.holder_user_id == holder_user_id,
        AccountLease.status == "active",
    )
    if not include_expired:
        query = query.filter(AccountLease.expires_at > _now())
    return [int(row[0]) for row in query.all()]


def lease_accounts(
    db: Session,
    *,
    holder_user_id: int,
    candidates: Sequence[Candidate],
    scope: str = SCOPE_ELITE,
    commodity_id: str = "",
    batch_id: str = "",
    quantity: int | None = None,
    ttl_minutes: int = DEFAULT_LEASE_TTL_MINUTES,
    now: datetime | None = None,
) -> list[AccountLease]:
    """按顺序领取租约，返回**真正领到**的那些。

    并发抢号由 `uq_lease_active_account` 裁决：每个候选用一个 SAVEPOINT 包住，
    撞唯一键就跳过它继续下一个 —— 已领到的不受影响（`rollback()` 会连坐，
    所以必须用 `begin_nested()`）。

    **本函数会 commit**：租约必须先落库，后续建任务才有依据。
    """
    moment = _now(now)
    expires_at = moment + timedelta(minutes=max(1, int(ttl_minutes)))
    leased: list[AccountLease] = []
    for candidate in candidates:
        if quantity is not None and len(leased) >= int(quantity):
            break
        lease = AccountLease(
            account_id=candidate.account_id,
            holder_user_id=holder_user_id,
            scope=scope,
            purpose="exchange",
            commodity_id=commodity_id,
            batch_id=batch_id,
            status="active",
            leased_at=moment,
            expires_at=expires_at,
            release_reason="",
            created_at=moment,
        )
        try:
            with db.begin_nested():
                db.add(lease)
                db.flush()
        except IntegrityError:
            # 撞唯一键 = 别人刚抢走这个账号。这是**预期内**的并发结果，不是故障。
            # SAVEPOINT 已经把这一条回滚掉了，直接换下一个候选。
            if lease in db:
                db.expunge(lease)
            continue
        leased.append(lease)
    db.commit()
    return leased


def bind_leases_to_batch(
    db: Session, leases: Sequence[AccountLease], batch_id: str
) -> None:
    """把租约与任务批次关联起来（任务创建后才拿得到 batch_id）。"""
    if not leases or not batch_id:
        return
    for lease in leases:
        lease.batch_id = batch_id
    db.commit()


def release_lease(
    db: Session,
    lease: AccountLease,
    *,
    reason: str,
    now: datetime | None = None,
) -> bool:
    """释放单个租约；已经是终态则返回 False（幂等）。"""
    if lease.status != "active":
        return False
    lease.status = "released"
    lease.released_at = _now(now)
    lease.release_reason = str(reason or "")[:64]
    db.commit()
    return True


def release_for_task(
    db: Session,
    task: DouyuTask,
    *,
    reason: str,
    now: datetime | None = None,
) -> bool:
    """任务终态时释放它对应账号的租约（文档 §3.9 的即时释放）。

    按 `(batch_id, account_id)` 定位：一个批次里每个账号各有一条租约，
    不能因为某一条任务结束就把整批的租约都释放掉。
    """
    if not task.batch_id:
        return False
    lease = (
        db.query(AccountLease)
        .filter(
            AccountLease.batch_id == task.batch_id,
            AccountLease.account_id == task.account_id,
            AccountLease.status == "active",
        )
        .first()
    )
    if lease is None:
        return False
    return release_lease(db, lease, reason=reason, now=now)


def release_many(
    db: Session,
    leases: Sequence[AccountLease],
    *,
    reason: str,
    now: datetime | None = None,
    status: str = "released",
) -> int:
    """批量释放（补位失败、客服放弃、超管强制释放）。

    `status` 只允许 `released`（人/流程主动放）与 `expired`（TTL 或孤儿批次兜底）——
    两者都会让生成列变 NULL、从而释放唯一索引，区别只在于事后怎么看这件事。
    """
    moment = _now(now)
    count = 0
    for lease in leases:
        if lease.status != "active":
            continue
        lease.status = status
        lease.released_at = moment
        lease.release_reason = str(reason or "")[:64]
        count += 1
    if count:
        db.commit()
    return count


def release_batch(
    db: Session, batch_id: str, *, reason: str, now: datetime | None = None
) -> int:
    """释放一个任务批次下的全部有效租约。"""
    leases = (
        db.query(AccountLease)
        .filter(AccountLease.batch_id == batch_id, AccountLease.status == "active")
        .all()
    )
    return release_many(db, leases, reason=reason, now=now)


def reap_expired(
    db: Session, *, now: datetime | None = None, batch_size: int = 500
) -> int:
    """把过期租约置为 expired（兜底作业调用）。

    只动**已过期**的：正在跑的任务即使超了 TTL 也先留着，
    避免把还在执行中的账号放给第二个人 —— TTL 到期的第一批由客服/运营处理，
    真正兜底的是「批次已不在内存」那条路径（见 reaper 作业）。
    """
    moment = _now(now)
    leases = (
        db.query(AccountLease)
        .filter(AccountLease.status == "active", AccountLease.expires_at <= moment)
        .order_by(AccountLease.id.asc())
        .limit(max(1, int(batch_size)))
        .all()
    )
    return release_many(db, leases, reason="ttl_expired", now=moment, status="expired")


def orphan_leases(db: Session, *, active_batch_ids: set[str], limit: int = 500):
    """批次已不在内存注册表（进程重启）的有效租约 —— 交给 reaper 释放。"""
    query = db.query(AccountLease).filter(AccountLease.status == "active")
    if active_batch_ids:
        query = query.filter(AccountLease.batch_id.notin_(list(active_batch_ids)))
    return query.order_by(AccountLease.id.asc()).limit(max(1, int(limit))).all()


#: 任务还没跑完的状态（与 `douyu_service.DOUYU_ACTIVE_TASK_STATUSES` 同义）。
#: 这里写一份而不是 import，是为了不让纯函数模块反向依赖整条 runner 依赖链。
_ACTIVE_TASK_STATUSES = ("planned", "pending", "running")


def reap_missed_releases(db: Session, *, limit: int = 500) -> int:
    """回收「任务已经终态、租约却还挂着」的租约。

    `_mark_task` 的即时释放是主路径，但进程被杀、任务行被保留策略删掉、
    释放时抛异常，都会留下悬空租约。判据是**该批次已经没有未完成任务**：

    - 批次里还有 `planned/pending/running` → 说明还在跑，不动它；
    - `batch_id` 为空 → 租约刚建、任务还没建出来，不动（由 TTL 兜底）。

    只动「没有在跑的任务」的租约，所以绝不会把正在执行的账号交给第二个人。
    """
    active_batches = select(DouyuTask.batch_id).where(
        DouyuTask.status.in_(_ACTIVE_TASK_STATUSES)
    )
    leases = (
        db.query(AccountLease)
        .filter(
            AccountLease.status == "active",
            AccountLease.batch_id != "",
            AccountLease.batch_id.notin_(active_batches),
        )
        .order_by(AccountLease.id.asc())
        .limit(max(1, int(limit)))
        .all()
    )
    return release_many(db, leases, reason="task_finished_missed")


def recharge_amounts_for(gold_needed: int) -> list[int]:
    """把一笔充值需求拆成每笔不超过 800 元的订单（文档 §3.4.2）。

    返回**空列表**表示不需要充值。单账号一件 1280 分的套装只需要 128 元，
    正常不会拆单；一次补很多分才会走到这里。

    >>> recharge_amounts_for(0)
    []
    >>> recharge_amounts_for(98)
    [98]
    >>> recharge_amounts_for(1700)
    [800, 800, 100]
    """
    remaining = max(0, int(gold_needed))
    orders: list[int] = []
    while remaining > 0:
        chunk = min(remaining, MAX_RECHARGE_AMOUNT)
        orders.append(chunk)
        remaining -= chunk
    return orders


@dataclass(frozen=True)
class SupplyStep:
    """补给链里的一步：任务类型 + 该任务自己的载荷。"""

    task_type: str
    payload: dict


def build_supply_chain(
    candidate: Candidate,
    *,
    commodity_id: str,
    exchange_task_type: str = "exchange_goods",
) -> list[SupplyStep]:
    """给一个 B 梯队账号排出「充值 → 赠令 → 兑换」的执行链（纯函数）。

    设计要点（都是被现实逼出来的）：

    1. **先查余额再决定要不要花钱**。链首固定放 `query_gold_balance` /`query_points`：
       充值步骤带 `_skip_when` 闸门，依据的是**执行那一刻**账号上的鱼翅/积分；
       不先刷新一次，闸门读的是旧值 —— 旧值偏低就会**多充真金白银**。
    2. **缺多少充多少**，超过 800 元自动拆单（服务端另有硬校验）。
    3. **赠令带 `points >= score` 闸门**：积分已经够了就别再送礼（送出去是钱）。
    4. A 梯队（积分已够）**不生成补给步骤**，链上只有兑换一步。

    执行顺序由「同一批次 + 并发 1」保证（`DouyuBatchRunner` 按任务 id 顺序取行）。
    """
    if not candidate.needs_topup:
        return [
            SupplyStep(
                exchange_task_type,
                {"commodity_id": commodity_id, "num": 1},
            )
        ]

    # 检查手册开通状态与最新积分
    status_step = (
        SupplyStep("query_elite_status", {"require_opened": True})
        if exchange_task_type == "exchange_goods"
        else SupplyStep("query_points", {})
    )
    steps: list[SupplyStep] = [
        status_step,
        SupplyStep("query_gold_balance", {}),
    ]
    # 充值：余额够就跳过（闸门在执行时求值，读的是刚刷新过的账号字段）
    for amount in recharge_amounts_for(candidate.topup_gifts):
        steps.append(
            SupplyStep(
                "create_gold_qr",
                {
                    "amount": amount,
                    "_skip_when": {
                        "column": "gold_balance",
                        "op": ">=",
                        "value": candidate.topup_gifts,
                    },
                },
            )
        )
    steps.append(
        SupplyStep(
            "donate_elite_gift",
            {
                "gift_count": candidate.topup_gifts,
                "_skip_when": {
                    "column": "points",
                    "op": ">=",
                    "value": candidate.score,
                },
            },
        )
    )
    steps.append(
        SupplyStep(exchange_task_type, {"commodity_id": commodity_id, "num": 1})
    )
    return steps


def validate_recharge_amount(amount: int | None) -> int:
    """校验单次充值金额（元）。

    只做**上限**校验，不做下限 —— 下限由调用方按业务给默认值。
    """
    value = int(amount or 0)
    if value <= 0:
        raise ValueError("充值金额必须大于 0")
    if value > MAX_RECHARGE_AMOUNT:
        raise ValueError(
            f"单次充值不得超过 {MAX_RECHARGE_AMOUNT} 元，本次请求 {value} 元"
        )
    return value


def lease_status_summary(db: Session, *, holder_user_id: int) -> dict[str, int]:
    """给池工作台用的租约概览。"""
    moment = _now()
    base = db.query(AccountLease).filter(
        AccountLease.holder_user_id == holder_user_id,
        AccountLease.status == "active",
    )
    return {
        "active": base.count(),
        "expiring_soon": base.filter(
            AccountLease.expires_at <= moment + timedelta(minutes=10)
        ).count(),
    }


__all__ = [
    "DEFAULT_LEASE_TTL_MINUTES",
    "EXCHANGE_COOLDOWN_DAYS",
    "MAX_RECHARGE_AMOUNT",
    "POINTS_PER_GIFT",
    "SUPPORTED_POOL_SCOPES",
    "Candidate",
    "SupplyStep",
    "active_lease_for_account",
    "bind_leases_to_batch",
    "build_supply_chain",
    "eligible_accounts",
    "estimate_supply",
    "lease_accounts",
    "lease_status_summary",
    "leased_account_ids",
    "make_candidate",
    "orphan_leases",
    "rank_candidates",
    "reap_expired",
    "reap_missed_releases",
    "recharge_amounts_for",
    "release_batch",
    "release_for_task",
    "release_lease",
    "release_many",
    "topup_gifts_for",
    "validate_recharge_amount",
]