"""斗鱼活动任务路由。"""

from __future__ import annotations

import asyncio
import threading
from datetime import UTC, datetime
from typing import Any, cast

from fastapi import (
    APIRouter,
    Depends,
    HTTPException,
    Query,
    Request,
    WebSocket,
    WebSocketDisconnect,
)
from loguru import logger
from sqlalchemy import case, func, or_, select
from sqlalchemy.orm import Session, joinedload

from core.douyu import (
    FishFinRechargeClient,
    FishFinRechargeConfig,
    FishFinRechargeConfigError,
)

from ..database import SessionLocal, get_db
from ..deps import authenticate_websocket, get_current_user, require_permission
from ..models import (
    Account,
    AccountLease,
    DouyuConfig,
    DouyuEsportsGoodsSnapshot,
    DouyuGoodsSnapshot,
    DouyuTask,
    DouyuWorkbench,
    DouyuWorkbenchAccount,
    DouyuXpdGoodsSnapshot,
    User,
)
from ..permissions import user_has_permission
from ..schemas import (
    DouyuConfigOut,
    DouyuConfigUpdate,
    DouyuGoodsOut,
    DouyuTaskAccountOut,
    DouyuTaskBatchRequest,
    DouyuTaskOut,
    DouyuWorkbenchAccountsUpdate,
    DouyuXpdGoodsOut,
)
from ..services.account_binding import bind_payload
from ..services.audit_service import record_audit
from ..services.business_events import record_recharge
from ..services.douyu_runner import DouyuBatchRunner, douyu_batch_registry
from ..services.douyu_service import (
    DOUYU_CONFIG_FIELDS,
    DOUYU_HANDBOOK_TASK_TYPES,
    SUPPORTED_DOUYU_TASK_TYPES,
    apply_douyu_config_defaults,
    cleanup_orphan_douyu_tasks,
    cookie_holding_filter,
    create_douyu_planned_tasks,
    douyu_config_value,
    douyu_task_payload,
    ensure_douyu_config,
)

router = APIRouter(prefix="/api/douyu", tags=["斗鱼活动"])


def _supplier_response_value(payload: dict, *keys: str):
    """兼容供应商把订单字段放在根节点、data 或 result 中。"""
    data_raw = payload.get("data")
    data = cast(dict[str, Any], data_raw) if isinstance(data_raw, dict) else {}
    result_raw = payload.get("result")
    result = cast(dict[str, Any], result_raw) if isinstance(result_raw, dict) else {}
    for source in (payload, data, result):
        for key in keys:
            if source.get(key) is not None:
                return source[key]
    return None


@router.post("/supplier-recharge/callback")
async def supplier_recharge_callback(request: Request, db: Session = Depends(get_db)):
    """接收供应商直充终态通知，验签后按 out_order_id 幂等更新任务。"""
    try:
        payload = await request.json()
    except ValueError as exc:
        raise HTTPException(status_code=400, detail="供应商回调不是 JSON") from exc
    if not isinstance(payload, dict):
        raise HTTPException(status_code=400, detail="供应商回调格式无效")
    try:
        client = FishFinRechargeClient(FishFinRechargeConfig.from_env())
    except FishFinRechargeConfigError as exc:
        logger.error("[douyu] 供应商直充回调配置无效: {}", exc)
        raise HTTPException(status_code=503, detail="供应商直充回调未配置") from exc
    if not client.verify_response_sign(payload, "POST"):
        logger.warning("[douyu] 供应商直充回调验签失败: keys={}", sorted(payload))
        raise HTTPException(status_code=401, detail="供应商回调签名无效")

    out_order_id = str(_supplier_response_value(payload, "out_order_id") or "").strip()
    if not out_order_id:
        raise HTTPException(status_code=400, detail="供应商回调缺少 out_order_id")
    task = (
        db.query(DouyuTask)
        .filter(
            DouyuTask.supplier_out_order_id == out_order_id,
            DouyuTask.task_type == "create_gold_qr",
        )
        .first()
    )
    if not task:
        raise HTTPException(status_code=404, detail="供应商回调订单不存在")

    status = DouyuBatchRunner._to_int(
        _supplier_response_value(payload, "order_status", "orderStatus")
    )
    result = dict(task.result) if isinstance(task.result, dict) else {}
    result.update(
        {
            "out_order_id": out_order_id,
            "order_id": _supplier_response_value(payload, "order_id", "orderId")
            or result.get("order_id"),
            "supplier_order_status": status,
            "supplier_order": DouyuBatchRunner._supplier_result(payload),
            "supplier_callback_received": True,
        }
    )
    # 这笔充值到没到账，用**回调自己说的终态**判断，而不是「任务状态有没有被改」。
    # 否则会漏掉一个真实黑洞：runner 轮询超时先判了 failed（gold.py 末尾），
    # 回调随后才到达成功 —— 守卫会跳过状态更新，这笔真金白银永远不进时间线。
    recharge_confirmed = status == 2
    if task.status not in {"success", "failed", "stopped"}:
        if status == 2:
            task.status = "success"
            task.message = "供应商直充成功（异步通知）"
            task.finished_at = datetime.now(UTC)
            if task.account:
                task.account.updated_at = datetime.now(UTC)
        elif status in {3, 4}:
            reason = str(
                _supplier_response_value(payload, "fail_reason", "message", "msg")
                or "供应商直充失败"
            )
            task.status = "failed"
            task.message = reason[:512]
            task.finished_at = datetime.now(UTC)
        else:
            task.status = "running"
            task.message = (
                f"供应商直充订单处理中（状态 {status if status is not None else '-'}）"
            )
    task.result = result
    callback_success = status not in {3, 4}
    # 回调说成功了就补一条充值事件。幂等键与 runner 那条**完全一致**
    # （business_events.task_ref），所以「runner 也观测到了」时不会重复记账；
    # 而「runner 已判超时/失败、回调才到」这个黑洞则能被补上。
    if recharge_confirmed and task.account is not None:
        record_recharge(
            db,
            task.account,
            task_id=task.id,
            amount=_to_int_or_none(result.get("buy_num")),
            order_id=str(result.get("out_order_id") or ""),
            supplier_order_id=str(result.get("order_id") or ""),
            channel="supplier_api",
            douyu_uid=str(result.get("douyu_uid") or ""),
            balance_after=task.account.gold_balance,
            confirmed=False,
        )
    record_audit(
        db,
        None,
        action="recharge:douyu:callback",
        target=f"supplier_order:{out_order_id}",
        detail={
            "out_order_id": out_order_id,
            "task_id": task.id,
            "order_id": result.get("order_id"),
            "supplier_order_status": status,
            "task_status": task.status,
            "message": task.message,
        },
        success=callback_success,
    )
    db.commit()
    logger.info(
        "[douyu] 供应商直充回调已处理: out_order_id={} status={}", out_order_id, status
    )
    acknowledgement = {"code": 200, "message": "success"}
    acknowledgement["sign"] = client.sign(acknowledgement, "POST")
    return acknowledgement


def _can_view_all(user: User) -> bool:
    return user_has_permission(user, "account:view_all")


def _visible_task_accounts_query(db: Session, current: User):
    """返回当前用户可用于斗鱼任务的账号查询。

    只要账号还持有 Cookie 就可见：排队中的普通登录（pending/running）和
    重登失败（relogin_failed）都会保留旧 CK，不能让账号从工作台消失。
    """
    cookie_ids = select(Account.id).where(cookie_holding_filter())
    query = (
        db.query(Account)
        .options(joinedload(Account.assigned_user))
        .filter(Account.id.in_(cookie_ids))
    )
    if _can_view_all(current):
        return query
    if user_has_permission(current, "account:view_assigned"):
        return query.filter(Account.assigned_to == current.id)
    raise HTTPException(status_code=403, detail="无权查看斗鱼账号")


def _visible_tasks_query(db: Session, current: User):
    """返回当前用户可查看的斗鱼任务查询。"""
    # 任务列表只展示账号识别信息，避免 joinedload 把密码、邮箱等加密大字段带出。
    account_loader = joinedload(DouyuTask.account).load_only(
        Account.id,
        Account.username,
        Account.uid,
        Account.nickname,
        Account.assigned_to,
    )
    query = db.query(DouyuTask).options(account_loader)
    if _can_view_all(current):
        return query
    if user_has_permission(current, "account:view_assigned"):
        # 彻底消除全表 filesort：直接通过 douyu_tasks 上的 account_assigned_to 复合索引倒序定位
        return query.filter(DouyuTask.account_assigned_to == current.id)
    raise HTTPException(status_code=403, detail="无权查看斗鱼任务")


def _require_task_account_access(
    db: Session, current: User, account_ids: list[int]
) -> None:
    """确保任务只会提交到当前用户可操作的账号。

    可操作 = **我名下** ∪ **我当前持有有效租约**。
    后半条是账号池加的（归属是管理员/运营的维护责任，客服只按需租用）——
    旧路径「名下账号」一字未改，这里只做**并集**。
    """
    requested_ids = set(account_ids)
    query = db.query(Account.id).filter(Account.id.in_(requested_ids))
    if not _can_view_all(current):
        if not user_has_permission(current, "account:view_assigned"):
            raise HTTPException(status_code=403, detail="无权操作斗鱼账号")
        my_leases = select(AccountLease.account_id).where(
            AccountLease.holder_user_id == current.id,
            AccountLease.status == "active",
        )
        query = query.filter(
            or_(Account.assigned_to == current.id, Account.id.in_(my_leases))
        )
    allowed_ids = {account_id for (account_id,) in query.all()}
    if allowed_ids != requested_ids:
        raise HTTPException(status_code=403, detail="包含无权操作的斗鱼账号")


def _require_batch_owner(db: Session, current: User, batch_id: str) -> None:
    """客服只能停止或订阅自己创建的任务批次。"""
    if _can_view_all(current):
        return
    exists = (
        db.query(DouyuTask.id)
        .filter(DouyuTask.batch_id == batch_id, DouyuTask.created_by == current.id)
        .first()
    )
    if not exists:
        raise HTTPException(status_code=403, detail="无权操作该斗鱼任务批次")


def start_douyu_batch(
    *,
    batch_id: str,
    task_type: str,
    payload: dict,
    concurrency: int,
) -> None:
    """在线程里启动一个批次的执行器（**必须在有事件循环的 async 路由里调用**）。

    抽出来是因为账号池也要起批次：两条路径的执行方式必须完全一致 ——
    复制一份的话，将来改了注册表或线程模型就会出现"老路径好了、池路径没好"。

    ⚠️ `DouyuBatchRunner` 按 `batch_id` 取**全部 planned 任务、不看 `task_type`**，
    再按每条任务自己的类型分发。所以**一个批次里可以放不同类型的任务**；
    但它们会被并发执行（`concurrency` 个 worker），顺序没有保证 ——
    需要严格先后（充值 → 赠令 → 兑换）时只能把 `concurrency` 设成 1。
    """
    log_queue = asyncio.Queue()
    loop = asyncio.get_running_loop()
    thread_db = SessionLocal()
    runner = DouyuBatchRunner(
        db=thread_db,
        batch_id=batch_id,
        task_type=task_type,
        payload=payload,
        log_queue=log_queue,
        loop=loop,
        concurrency=concurrency,
    )
    douyu_batch_registry.register(batch_id, log_queue, loop, runner)
    thread = threading.Thread(target=runner.run, daemon=True)
    thread.start()


def _parse_account_names(raw_names: str) -> list[str]:
    """解析前端粘贴的 Excel 账号名，每行一个并去重。"""
    names: list[str] = []
    seen: set[str] = set()
    for value in (
        (raw_names or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
    ):
        name = value.strip()
        if name and name not in seen:
            names.append(name)
            seen.add(name)
    return names


def _apply_task_account_filters(
    query,
    *,
    search: str = "",
    tag: str = "",
    account_names: str = "",
    ids: str = "",
):
    """按工作台导入弹窗的搜索、标签、多行账号名筛选任务账号。"""
    id_list = [int(item) for item in ids.split(",") if item.strip().isdigit()]
    if id_list:
        query = query.filter(Account.id.in_(id_list))
    tag_text = (tag or "").strip()
    if tag_text:
        query = query.filter(Account.tag == tag_text)
    selected_names = _parse_account_names(account_names)
    if selected_names:
        query = query.filter(Account.username.in_(selected_names))
    search_text = (search or "").strip()
    if search_text:
        pattern = f"%{search_text}%"
        query = query.filter(
            or_(
                Account.username.ilike(pattern),
                Account.uid.ilike(pattern),
                Account.nickname.ilike(pattern),
                Account.tag.ilike(pattern),
                Account.game_name.ilike(pattern),
                Account.esports_game_name.ilike(pattern),
                Account.xpd_game_name.ilike(pattern),
            )
        )
    return query, selected_names


def _order_task_accounts(query, selected_names: list[str]):
    """自定义账号名按粘贴顺序排序，未指定时保持最新账号优先。"""
    if selected_names:
        input_order = case(
            {name: index for index, name in enumerate(selected_names)},
            value=Account.username,
            else_=len(selected_names),
        )
        return query.order_by(input_order, Account.id.desc())
    return query.order_by(Account.id.desc())


def _account_out(account: Account) -> DouyuTaskAccountOut:
    return DouyuTaskAccountOut(
        id=account.id,
        username=account.username,
        uid=account.uid or "",
        nickname=account.nickname or "",
        tag=account.tag or "",
        points=account.points,
        game_name=account.game_name or "",
        game_channel=account.game_channel or "",
        gold_balance=account.gold_balance,
        exchange_balance=account.exchange_balance,
        **bind_payload(account),
        change_role_wait_time=account.change_role_wait_time,
        elite_opened=account.elite_opened,
        elite_expire_time=account.elite_expire_time,
        esports_points=account.esports_points,
        esports_game_name=account.esports_game_name or "",
        esports_game_channel=account.esports_game_channel or "",
        esports_change_role_wait_time=account.esports_change_role_wait_time,
        esports_can_change_time=account.esports_can_change_time,
        xpd_game_name=account.xpd_game_name or "",
        xpd_openid=account.xpd_openid or "",
        xpd_role_id=account.xpd_role_id or "",
        xpd_plat_id=account.xpd_plat_id,
        xpd_area_id=account.xpd_area_id,
        xpd_balance=account.xpd_balance,
        xpd_fragments=account.xpd_fragments,
        assigned_to=account.assigned_to,
        assigned_username=account.assigned_user.username
        if account.assigned_user
        else None,
    )


def _task_out(task: DouyuTask, *, include_detail: bool = False) -> DouyuTaskOut:
    return DouyuTaskOut(**douyu_task_payload(task, include_detail=include_detail))


def _config_out(config: DouyuConfig) -> DouyuConfigOut:
    text = lambda field: str(douyu_config_value(field, getattr(config, field)))
    number = lambda field: int(douyu_config_value(field, getattr(config, field)))
    boolean = lambda field: bool(douyu_config_value(field, getattr(config, field)))
    return DouyuConfigOut(
        manual_id=text("manual_id"),
        rid=text("rid"),
        bind_act_alias=text("bind_act_alias"),
        confirm_act_alias=text("confirm_act_alias"),
        legacy_act_alias=text("legacy_act_alias"),
        room_id=text("room_id"),
        elite_amount=number("elite_amount"),
        esports_manual_id=text("esports_manual_id"),
        esports_act_alias=text("esports_act_alias"),
        esports_amount=number("esports_amount"),
        esports_chicken_gift_id=text("esports_chicken_gift_id"),
        esports_chicken_skin_id=text("esports_chicken_skin_id"),
        esports_firework_gift_id=text("esports_firework_gift_id"),
        esports_firework_skin_id=text("esports_firework_skin_id"),
        peace_enabled=boolean("peace_enabled"),
        xpd_act_alias=text("xpd_act_alias"),
        xpd_act_id=text("xpd_act_id"),
        xpd_rid=text("xpd_rid"),
        gold_pay_type=number("gold_pay_type"),
        gold_recharge_channel=text("gold_recharge_channel"),
        gold_api_product_id=text("gold_api_product_id"),
        gold_api_account_template_name=text("gold_api_account_template_name"),
        gift_id=text("gift_id"),
        skin_id=text("skin_id"),
        updated_at=config.updated_at,
    )


@router.get("/task-types")
def task_types(current: User = Depends(require_permission("douyu:task"))):
    """返回斗鱼任务类型。"""
    return SUPPORTED_DOUYU_TASK_TYPES


@router.get("/accounts")
def list_task_accounts(
    search: str = Query(""),
    tag: str = Query(""),
    ids: str = Query(""),
    account_names: str = Query(""),
    page: int | None = Query(None, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """查看可执行斗鱼任务的账号（必须有成功 Cookie）。

    tag 按账号标签精确筛选；ids 为逗号分隔的账号 ID，用于工作台按已导入账号过滤；
    account_names 为多行斗鱼账户名（每行一个），按粘贴顺序精确匹配；为空时返回全部。
    """
    query, selected_names = _apply_task_account_filters(
        _visible_task_accounts_query(db, current),
        search=search,
        tag=tag,
        account_names=account_names,
        ids=ids,
    )
    total = None
    if page is not None:
        total = query.order_by(None).count()
    query = _order_task_accounts(query, selected_names)
    if page is not None:
        query = query.offset((page - 1) * page_size).limit(page_size)
    else:
        query = query.limit(500)
    result = [_account_out(account) for account in query.all()]
    if page is not None:
        return {
            "items": result,
            "total": total or 0,
            "page": page,
            "page_size": page_size,
        }
    return result


@router.get("/accounts/ids")
def list_task_account_ids(
    search: str = Query(""),
    tag: str = Query(""),
    account_names: str = Query(""),
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """返回当前用户可导入工作台的全部账号 ID，不受列表分页限制。"""
    query, selected_names = _apply_task_account_filters(
        _visible_task_accounts_query(db, current),
        search=search,
        tag=tag,
        account_names=account_names,
    )
    account_ids = [
        account_id
        for (account_id,) in _order_task_accounts(
            query.enable_eagerloads(False).with_entities(Account.id),
            selected_names,
        ).all()
    ]
    return {"account_ids": account_ids, "total": len(account_ids)}


@router.get("/workbench-accounts")
def list_workbench_accounts(
    handbook_scope: str = Query(..., pattern="^(elite|esports|peace)$"),
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """返回当前用户在指定工作台启用的账号，供不同浏览器同步。"""
    rows = (
        db.query(DouyuWorkbenchAccount.account_id)
        .join(Account, Account.id == DouyuWorkbenchAccount.account_id)
        .filter(
            DouyuWorkbenchAccount.user_id == current.id,
            DouyuWorkbenchAccount.handbook_scope == handbook_scope,
        )
    )
    if not _can_view_all(current):
        rows = rows.filter(Account.assigned_to == current.id)
    account_ids = [
        account_id
        for (account_id,) in rows.order_by(DouyuWorkbenchAccount.id.asc()).all()
    ]
    configured = (
        db.query(DouyuWorkbench.id)
        .filter(
            DouyuWorkbench.user_id == current.id,
            DouyuWorkbench.handbook_scope == handbook_scope,
        )
        .first()
        is not None
    )
    return {"account_ids": account_ids, "configured": configured}


@router.put("/workbench-accounts")
def update_workbench_accounts(
    req: DouyuWorkbenchAccountsUpdate,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """用当前完整账号集合覆盖一个工作台，作为跨浏览器的同步状态。"""
    account_ids = sorted(set(req.account_ids))
    if any(account_id < 1 for account_id in account_ids):
        raise HTTPException(status_code=400, detail="无效的账号ID")
    if account_ids:
        _require_task_account_access(db, current, account_ids)
    workbench = (
        db.query(DouyuWorkbench)
        .filter(
            DouyuWorkbench.user_id == current.id,
            DouyuWorkbench.handbook_scope == req.handbook_scope,
        )
        .first()
    )
    if workbench is None:
        db.add(DouyuWorkbench(user_id=current.id, handbook_scope=req.handbook_scope))
    else:
        workbench.updated_at = datetime.now(UTC)
    db.query(DouyuWorkbenchAccount).filter(
        DouyuWorkbenchAccount.user_id == current.id,
        DouyuWorkbenchAccount.handbook_scope == req.handbook_scope,
    ).delete(synchronize_session=False)
    db.add_all(
        [
            DouyuWorkbenchAccount(
                user_id=current.id,
                handbook_scope=req.handbook_scope,
                account_id=account_id,
            )
            for account_id in account_ids
        ]
    )
    db.commit()
    return {"account_ids": account_ids, "success": True}


@router.get("/config", response_model=DouyuConfigOut)
def get_config(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:config")),
):
    """获取斗鱼活动配置。"""
    return _config_out(ensure_douyu_config(db))


@router.get("/recharge-channel")
def get_recharge_channel(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """返回当前鱼翅充值渠道，不暴露供应商商品等配置细节。"""
    config = ensure_douyu_config(db)
    return {
        "gold_recharge_channel": douyu_config_value(
            "gold_recharge_channel",
            config.gold_recharge_channel,
        ),
    }


@router.put("/config", response_model=DouyuConfigOut)
def update_config(
    req: DouyuConfigUpdate,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:config")),
):
    """更新斗鱼活动配置。"""
    config = ensure_douyu_config(db)
    changed_fields = []
    for field in DOUYU_CONFIG_FIELDS:
        value = getattr(req, field, None)
        if value is None:
            continue
        setattr(config, field, value.strip() if isinstance(value, str) else value)
        changed_fields.append(field)
    apply_douyu_config_defaults(config)
    config.updated_at = datetime.now(UTC)
    recharge_fields = [field for field in changed_fields if field.startswith("gold_")]
    if recharge_fields:
        record_audit(
            db,
            current,
            action="recharge:douyu:config",
            target="douyu_config",
            detail={"changed_fields": recharge_fields},
        )
    db.commit()
    db.refresh(config)
    return _config_out(config)


@router.get("/goods", response_model=list[DouyuGoodsOut])
def list_goods(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """查看已缓存的斗鱼商品快照。"""
    rows = db.query(DouyuGoodsSnapshot).order_by(DouyuGoodsSnapshot.id.asc()).all()
    return rows


@router.get("/esports-goods", response_model=list[DouyuGoodsOut])
def list_esports_goods(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """查看已缓存的电竞手册皮肤商城商品。"""
    rows = (
        db.query(DouyuEsportsGoodsSnapshot)
        .order_by(DouyuEsportsGoodsSnapshot.id.asc())
        .all()
    )
    return rows


@router.get("/xpd-goods", response_model=list[DouyuXpdGoodsOut])
def list_xpd_goods(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """查看已缓存的和平小店商品。"""
    rows = (
        db.query(DouyuXpdGoodsSnapshot).order_by(DouyuXpdGoodsSnapshot.id.asc()).all()
    )
    return rows


@router.post("/tasks/batch")
async def create_task_batch(
    req: DouyuTaskBatchRequest,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """创建斗鱼任务记录并启动后台执行器。"""
    if not req.account_ids:
        raise HTTPException(status_code=400, detail="请选择斗鱼账号")
    _require_task_account_access(db, current, req.account_ids)

    cleanup_orphan_douyu_tasks(
        db,
        active_batch_ids=douyu_batch_registry.active_ids(),
        statuses=("pending", "running"),
        message="任务已中断（无执行器接管）",
    )

    try:
        batch_id, count = create_douyu_planned_tasks(
            db,
            req.account_ids,
            req.task_type,
            req.handbook_scope,
            current.id,
            req.payload,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    if count == 0:
        raise HTTPException(
            status_code=400, detail="没有可执行的斗鱼账号，请先登录获取 Cookie"
        )

    if req.task_type == "create_gold_qr":
        config = ensure_douyu_config(db)
        recharge_channel = str(
            douyu_config_value("gold_recharge_channel", config.gold_recharge_channel)
        ).strip()
        payment_method = (
            "API 直充支付" if recharge_channel == "supplier_api" else "微信扫码支付"
        )
        recharge_accounts = (
            db.query(DouyuTask, Account)
            .join(Account, Account.id == DouyuTask.account_id)
            .filter(
                DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr"
            )
            .order_by(DouyuTask.id.asc())
            .limit(100)
            .all()
        )
        record_audit(
            db,
            current,
            action="recharge:douyu:create",
            target=f"douyu_batch:{batch_id}",
            detail={
                "batch_id": batch_id,
                "task_type": req.task_type,
                "count": count,
                "account_count": len(req.account_ids),
                "handbook_scope": req.handbook_scope,
                "recharge_channel": recharge_channel,
                "payment_method": payment_method,
                # 仅记录充值身份，禁止在审计中写入 Cookie、密码等凭据。
                "recharge_accounts": [
                    {
                        "task_id": task.id,
                        "username": account.username,
                        "douyu_uid": account.uid or "",
                        "douyu_nickname": account.nickname or "",
                    }
                    for task, account in recharge_accounts
                ],
                "recharge_accounts_truncated": count > len(recharge_accounts),
            },
        )
        db.commit()

    start_douyu_batch(
        batch_id=batch_id,
        task_type=req.task_type,
        payload=req.payload,
        concurrency=req.concurrency,
    )

    return {"batch_id": batch_id, "count": count, "success": True}


@router.get("/tasks")
def list_tasks(
    batch_id: str | None = None,
    handbook_scope: str | None = Query(None, pattern="^(elite|esports|peace)$"),
    include_detail: bool = Query(
        False, description="是否返回完整任务结果（默认否，轮询请保持 false）"
    ),
    page: int | None = Query(None, ge=1),
    page_size: int = Query(100, ge=1, le=500),
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """查看斗鱼任务记录，默认只返回最近 100 条。"""
    query = _visible_tasks_query(db, current)
    if batch_id:
        query = query.filter(DouyuTask.batch_id == batch_id)
    if handbook_scope:
        # 旧任务没有归属字段，按历史任务类型继续展示，但绝不会触发自动二维码弹窗。
        query = query.filter(
            or_(
                DouyuTask.handbook_scope == handbook_scope,
                (DouyuTask.handbook_scope == "legacy")
                & DouyuTask.task_type.in_(DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]),
            )
        )
    total = None
    if page is not None:
        total = (
            query.enable_eagerloads(False)
            .order_by(None)
            .with_entities(func.count(DouyuTask.id))
            .scalar()
            or 0
        )
    query = query.order_by(DouyuTask.id.desc())
    if page is not None:
        query = query.offset((page - 1) * page_size).limit(page_size)
    else:
        query = query.limit(page_size)
    result = [_task_out(task, include_detail=include_detail) for task in query.all()]
    if page is not None:
        return {
            "items": result,
            "total": total or 0,
            "page": page,
            "page_size": page_size,
        }
    return result


@router.post("/tasks/cleanup-orphans")
def cleanup_orphan_tasks(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """手动清理没有内存执行器接管的斗鱼任务。"""
    cleaned = cleanup_orphan_douyu_tasks(
        db,
        active_batch_ids=douyu_batch_registry.active_ids(),
        statuses=("pending", "running"),
        message="任务已中断（无执行器接管）",
    )
    return {
        "message": f"已清理 {cleaned} 个残留任务",
        "cleaned": cleaned,
        "success": True,
    }


@router.get("/tasks/{task_id}", response_model=DouyuTaskOut)
def get_task(
    task_id: int,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """获取单条斗鱼任务详情。"""
    task = _visible_tasks_query(db, current).filter(DouyuTask.id == task_id).first()
    if not task:
        raise HTTPException(status_code=404, detail="任务不存在")
    return _task_out(task, include_detail=True)


@router.post("/stop/{batch_id}")
def stop_batch(
    batch_id: str,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("douyu:task")),
):
    """停止正在运行的斗鱼批次。"""
    _require_batch_owner(db, current, batch_id)
    is_recharge_batch = (
        db.query(DouyuTask.id)
        .filter(DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr")
        .first()
        is not None
    )
    batch = douyu_batch_registry.get(batch_id)
    if batch:
        if batch.get("finished"):
            douyu_batch_registry.pop(batch_id)
            cleaned = cleanup_orphan_douyu_tasks(
                db, batch_id=batch_id, message="批次已结束"
            )
            if cleaned:
                return {
                    "message": f"批次已结束，已清理 {cleaned} 个残留任务",
                    "success": True,
                }
            raise HTTPException(status_code=404, detail="批次已结束")
        batch["runner"].stop()
        if is_recharge_batch:
            record_audit(
                db,
                current,
                action="recharge:douyu:stop",
                target=f"douyu_batch:{batch_id}",
                detail={"batch_id": batch_id, "mode": "running"},
            )
            db.commit()
        return {"message": "已发送停止信号", "success": True}

    cleaned = cleanup_orphan_douyu_tasks(
        db, batch_id=batch_id, message="任务已停止（批次不存在）"
    )
    if cleaned:
        if is_recharge_batch:
            record_audit(
                db,
                current,
                action="recharge:douyu:stop",
                target=f"douyu_batch:{batch_id}",
                detail={
                    "batch_id": batch_id,
                    "mode": "orphan_cleanup",
                    "cleaned": cleaned,
                },
            )
            db.commit()
        return {"message": f"已清理 {cleaned} 个残留任务", "success": True}
    raise HTTPException(status_code=404, detail="批次不存在或已结束")


@router.websocket("/ws/{batch_id}")
async def ws_douyu_logs(websocket: WebSocket, batch_id: str):
    """斗鱼实时日志推送通道。"""
    user = authenticate_websocket(websocket)
    if not user:
        await websocket.close(code=1008, reason="未授权")
        return
    if not user_has_permission(user, "douyu:task"):
        await websocket.close(code=1008, reason="无权限")
        return
    db = SessionLocal()
    try:
        _require_batch_owner(db, user, batch_id)
    except HTTPException:
        await websocket.close(code=1008, reason="无权访问该任务批次")
        return
    finally:
        db.close()
    await websocket.accept()

    batch = douyu_batch_registry.get(batch_id)
    if not batch:
        await websocket.send_json({"level": "error", "message": "批次不存在或已结束"})
        await websocket.close()
        return

    log_queue: asyncio.Queue = batch["log_queue"]
    try:
        while True:
            try:
                msg = await asyncio.wait_for(log_queue.get(), timeout=30)
                await websocket.send_json(msg)
                if msg.get("level") == "result":
                    await asyncio.sleep(0.1)
                    break
            except TimeoutError:
                await websocket.send_json({"level": "heartbeat", "message": ""})
    except WebSocketDisconnect:
        pass
    finally:
        latest = douyu_batch_registry.get(batch_id)
        if latest and latest.get("finished"):
            douyu_batch_registry.pop(batch_id)


def _to_int_or_none(value: object) -> int | None:
    """回调里的 `buy_num` 可能是字符串/Decimal，安全转 int。"""
    try:
        return int(float(str(value)))
    except (TypeError, ValueError):
        return None