live-hub-py/web/backend/services/douyu_service.py

"""斗鱼活动任务服务。"""

from __future__ import annotations

import uuid
from datetime import UTC, datetime

from sqlalchemy import and_, func, select
from sqlalchemy.orm import Session

from core.douyu.activity_client import DouyuActivityClient
from core.douyu.cookie_utils import cookie_value

from ..models import Account, DouyuConfig, DouyuTask, LoginTask

SUPPORTED_DOUYU_TASK_TYPES = {
    "get_bind_qr": "获取绑定二维码",
    "confirm_bind": "确认绑定",
    "create_elite_qr": "开通精英宝典30",
    "query_elite_status": "查询精英宝典状态",
    "prepare_esports_bind": "绑定电竞手册角色",
    "get_esports_bind_qr": "切换电竞角色",
    "query_esports_game_name": "查询最新游戏角色",
    "confirm_esports_bind": "确认电竞手册绑定",
    "create_esports_qr": "开通电竞手册30",
    "query_esports_points": "查询电竞积分",
    "donate_esports_chicken_gift": "赠送冠军鸡腿",
    "donate_esports_firework_gift": "赠送冠军烟花",
    "refresh_esports_goods": "刷新电竞皮肤商城",
    "exchange_esports_goods": "兑换电竞皮肤",
    "create_gold_qr": "充值鱼翅",
    "donate_elite_gift": "赠送精英令",
    "query_points": "一键查询积分",
    "lock_goods": "锁定商品",
    "pay_locked_order": "支付锁单",
    "exchange_goods": "兑换商品",
    "query_game_name": "一键获取游戏名",
    "query_change_bind_time": "一键查询换绑时间",
    "query_limited_goods": "一键查询限兑商品",
    # 跨域聚合查询：鱼翅/积分/精英/电竞/小店/兑换记录/高价值套装是否已拥有
    "query_account_overview": "一键查询账号信息",
    "query_gold_balance": "一键查询鱼翅余额",
    "refresh_goods": "刷新商品列表",
    "query_exchange_records": "一键查询兑换记录",
    "prefetch_csrf_token": "一键获取兑换 CSRF Token",
    "get_xpd_bind_qr": "生成小店绑定二维码",
    "query_xpd_bind_info": "查询小店绑定信息",
    "confirm_xpd_bind": "确认小店绑定",
    "query_xpd_role": "查询小店绑定角色",
    "refresh_xpd_goods": "刷新小店商品列表",
    "query_xpd_balance": "查询小店点券余额",
    "query_xpd_fragments": "查询小店扭蛋碎片",
    "query_xpd_purchase_records": "查询小店兑换记录",
    "exchange_xpd_goods": "兑换小店商品",
}

DOUYU_HANDBOOK_SCOPES = {"elite", "esports", "peace"}
DOUYU_HANDBOOK_TASK_TYPES = {
    "elite": {
        "get_bind_qr",
        "confirm_bind",
        "create_elite_qr",
        "query_elite_status",
        "create_gold_qr",
        "donate_elite_gift",
        "query_points",
        "lock_goods",
        "pay_locked_order",
        "exchange_goods",
        "query_game_name",
        "query_change_bind_time",
        "query_limited_goods",
        "query_gold_balance",
        "refresh_goods",
        "query_exchange_records",
        "prefetch_csrf_token",
        "query_account_overview",
    },
    "esports": {
        "prepare_esports_bind",
        "get_esports_bind_qr",
        "query_esports_game_name",
        "confirm_esports_bind",
        "create_esports_qr",
        "query_esports_points",
        "query_gold_balance",
        "query_change_bind_time",
        "query_limited_goods",
        "refresh_esports_goods",
        "exchange_esports_goods",
        "create_gold_qr",
        "donate_esports_chicken_gift",
        "donate_esports_firework_gift",
        "query_account_overview",
    },
    "peace": {
        "get_xpd_bind_qr",
        "query_xpd_bind_info",
        "confirm_xpd_bind",
        "query_xpd_role",
        "refresh_xpd_goods",
        "query_xpd_balance",
        "query_xpd_fragments",
        "query_xpd_purchase_records",
        "exchange_xpd_goods",
        "query_account_overview",
    },
}


DOUYU_CONFIG_DEFAULTS = {
    "manual_id": "G4KA4Qnz4LDp7",
    "rid": "9263298",
    "bind_act_alias": "20260120QYOOB",
    "confirm_act_alias": "20260120QYOOB",
    "legacy_act_alias": "cjm",
    "room_id": "9263298",
    "elite_amount": 3000,
    "esports_manual_id": "1W0yGkOdDbAXb",
    "esports_act_alias": "20260226ZMGDG",
    "esports_amount": 3000,
    "esports_chicken_gift_id": "24766",
    "esports_chicken_skin_id": "0",
    "esports_firework_gift_id": "24767",
    "esports_firework_skin_id": "3850",
    "gold_pay_type": 1,
    "gold_recharge_channel": "wechat_qr",
    "gold_api_product_id": "111570",
    "gold_api_account_template_name": "斗鱼昵称",
    "gift_id": "23643",
    "skin_id": "2942",
    # 和平小店总开关，默认关（小店当前关闭；关着时一键查询完全跳过它的请求）
    "peace_enabled": False,
    "xpd_act_alias": "20260623KDQFH",
    "xpd_act_id": "46195",
    "xpd_rid": "9263298",
}

DOUYU_CONFIG_FIELDS = tuple(DOUYU_CONFIG_DEFAULTS.keys())
DOUYU_ACTIVE_TASK_STATUSES = ("planned", "pending", "running")


def douyu_config_value(field: str, value):
    """读取配置值；空值自动回退到默认值。"""
    default = DOUYU_CONFIG_DEFAULTS[field]
    if isinstance(default, int):
        try:
            return int(value if value is not None else default)
        except (TypeError, ValueError):
            return default
    text = str(value or "").strip()
    return text or str(default)


def apply_douyu_config_defaults(config: DouyuConfig) -> bool:
    """补齐斗鱼配置默认值，返回是否发生变更。"""
    changed = False
    for field in DOUYU_CONFIG_FIELDS:
        if (
            field == "bind_act_alias"
            and str(getattr(config, field, "") or "").strip() == "20250213NQCYX"
        ):
            setattr(config, field, DOUYU_CONFIG_DEFAULTS[field])
            changed = True
            continue
        normalized = douyu_config_value(field, getattr(config, field, None))
        if field == "gold_recharge_channel" and normalized not in {
            "wechat_qr",
            "supplier_api",
        }:
            normalized = DOUYU_CONFIG_DEFAULTS[field]
        if getattr(config, field, None) != normalized:
            setattr(config, field, normalized)
            changed = True
    return changed


def ensure_douyu_config(db: Session) -> DouyuConfig:
    """获取单条斗鱼配置，不存在则创建。"""
    config = db.query(DouyuConfig).first()
    if config:
        if apply_douyu_config_defaults(config):
            db.commit()
            db.refresh(config)
        return config
    config = DouyuConfig(**DOUYU_CONFIG_DEFAULTS)
    db.add(config)
    db.commit()
    db.refresh(config)
    return config


# 登录/重登尝试的在途状态：判断只看 attempt_*，与账号持有的 CK 无关。
LOGIN_ATTEMPT_IN_FLIGHT_STATUSES = ("pending", "running")


def cookie_holding_filter():
    """判定**账号**当前是否持有 CK。

    CK 是账号级凭证（2026-10-04 起从 `login_tasks.cookie` 迁到 `accounts.cookie`）。
    判定只看账号自己，不再看任何登录记录的状态 —— 这正是解耦的目的：
    删历史、批次排队、重登失败都不该让账号凭空「没有 CK」。
    """
    return and_(
        Account.cookie.isnot(None),
        Account.cookie != "",
    )


def cookie_operable_filter():
    """判定**账号**能否基于现有 CK 再做检测或重登。

    在途排除仍然要查 `login_tasks`：「正被批次处理」是**某次尝试**的属性，
    CK 才是账号的属性。所以这里是「账号有 CK」+「该账号没有在途尝试」。

    走 `login_tasks(account_id, attempt_status)` 索引（迁移 0042 新增），
    否则每轮选行都要回表。
    """
    in_flight = (
        select(LoginTask.id)
        .where(LoginTask.account_id == Account.id)
        .where(
            func.coalesce(LoginTask.attempt_status, "").in_(
                LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
            )
        )
        .exists()
    )
    return and_(cookie_holding_filter(), ~in_flight)


def latest_cookie(db: Session, account_id: int) -> str:
    """读取账号当前 CK。"""
    account = db.query(Account.cookie).filter(Account.id == account_id).first()
    return (account[0] if account else "") or ""


def cookie_holding_account_ids_query(db: Session):
    """返回当前仍持有 CK 的斗鱼账号 ID 查询。"""
    return db.query(Account.id).filter(cookie_holding_filter())


def visible_douyu_task_accounts(db: Session, account_ids: list[int]) -> list[Account]:
    """只保留当前仍持有 CK 的斗鱼账号。"""
    if not account_ids:
        return []
    return (
        db.query(Account)
        .filter(Account.id.in_(account_ids), cookie_holding_filter())
        .all()
    )


def update_account_profile_from_cookie(account: Account, cookie: str) -> None:
    """从 Cookie 回填 uid/nickname。"""
    profile = DouyuActivityClient.profile_from_cookie(cookie)
    if profile["uid"]:
        account.uid = profile["uid"]
    if profile["nickname"]:
        account.nickname = profile["nickname"]


def account_uid(account: Account, cookie: str) -> str:
    """优先从账号字段读取 uid，缺失时从 Cookie 中取。"""
    return account.uid or cookie_value(cookie, "acf_uid")


def create_douyu_task(
    db: Session,
    account: Account,
    task_type: str,
    handbook_scope: str,
    created_by: int,
    payload: dict | None = None,
    *,
    batch_id: str | None = None,
    account_assigned_to: int | None = None,
    commit: bool = True,
) -> DouyuTask:
    """建**一条**斗鱼任务（单账号、单类型、载荷可各账号不同）。

    存在的理由：`create_douyu_planned_tasks` 给一批账号**共用一个** `payload`，
    而账号池的补给链每个账号的充值额/赠令数都不一样，共用载荷根本表达不了。

    `batch_id` 传值即并入该批次 —— 执行器按批取任务、按任务自己的类型分发，
    所以**一个批次里可以放不同类型的任务**（这正是补给链的实现方式）。
    顺序由「同一批次 + 并发 1」保证（见 `routers/douyu.py::start_douyu_batch`）。
    """
    if task_type not in SUPPORTED_DOUYU_TASK_TYPES:
        raise ValueError("不支持的任务类型")
    if handbook_scope not in DOUYU_HANDBOOK_SCOPES:
        raise ValueError("无效的工作台")
    if task_type not in DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]:
        raise ValueError("该任务不属于当前工作台")

    task = DouyuTask(
        batch_id=batch_id or uuid.uuid4().hex[:12],
        account_id=account.id,
        account_assigned_to=(
            account_assigned_to
            if account_assigned_to is not None
            else account.assigned_to
        ),
        task_type=task_type,
        handbook_scope=handbook_scope,
        status="planned",
        message="任务已创建，等待执行",
        result={"payload": payload} if payload else None,
        created_by=created_by,
    )
    db.add(task)
    if commit:
        db.commit()
        db.refresh(task)
    else:
        db.flush()
    return task


def create_douyu_planned_tasks(
    db: Session,
    account_ids: list[int],
    task_type: str,
    handbook_scope: str,
    created_by: int,
    payload: dict | None = None,
    *,
    account_assigned_to_override: int | None = None,
) -> tuple[str, int]:
    """创建斗鱼任务记录，等待后台执行器消费。

    `account_assigned_to_override` **只给账号池路径用**：池化后账号归属是
    管理员/运营，若仍按归属写，租用该账号的客服在任务列表里**看不到自己提交的任务**
    —— 列表对客服的过滤正是 `DouyuTask.account_assigned_to == 我`
    （见 `routers/douyu.py` 的 `_visible_tasks_query`）。

    这一列本来就是「账号归属客服冗余字段，配合复合索引消除任务列表全表 filesort」，
    实际作用是「谁能在列表里看到这条任务」。池路径写**租用人**既保住索引命中，
    又让客服看得到自己的任务。**默认不传 = 保持旧行为**，旧路径完全不受影响。
    """
    if task_type not in SUPPORTED_DOUYU_TASK_TYPES:
        raise ValueError("不支持的任务类型")
    if handbook_scope not in DOUYU_HANDBOOK_SCOPES:
        raise ValueError("无效的工作台")
    if task_type not in DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]:
        raise ValueError("该任务不属于当前工作台")

    accounts = visible_douyu_task_accounts(db, account_ids)
    if (
        task_type in {"refresh_goods", "refresh_esports_goods", "refresh_xpd_goods"}
        and accounts
    ):
        # 商品快照是全局数据，一个可用 CK 足够；没有 CK 时前端无法选账号创建任务。
        accounts = accounts[:1]

    batch_id = uuid.uuid4().hex[:12]
    payload = payload or {}
    for account in accounts:
        db.add(
            DouyuTask(
                batch_id=batch_id,
                account_id=account.id,
                account_assigned_to=(
                    account_assigned_to_override
                    if account_assigned_to_override is not None
                    else account.assigned_to
                ),
                task_type=task_type,
                handbook_scope=handbook_scope,
                status="planned",
                message="任务已创建，等待执行",
                result={"payload": payload} if payload else None,
                created_by=created_by,
            )
        )
    db.commit()
    return batch_id, len(accounts)


def cleanup_orphan_douyu_tasks(
    db: Session,
    *,
    active_batch_ids: set[str] | None = None,
    batch_id: str | None = None,
    statuses: tuple[str, ...] = DOUYU_ACTIVE_TASK_STATUSES,
    message: str = "任务已中断（服务重启或批次丢失）",
) -> int:
    """清理没有执行器接管的斗鱼任务。"""
    query = db.query(DouyuTask).filter(DouyuTask.status.in_(statuses))
    if batch_id:
        query = query.filter(DouyuTask.batch_id == batch_id)
    elif active_batch_ids is not None and active_batch_ids:
        query = query.filter(~DouyuTask.batch_id.in_(list(active_batch_ids)))

    tasks = query.all()
    if not tasks:
        return 0
    now = datetime.now(UTC)
    for task in tasks:
        task.status = "stopped"
        task.message = message
        task.finished_at = now
    db.commit()
    return len(tasks)


def slim_douyu_goods(goods: object) -> object:
    """保留兑换图片和列表展示需要的商品小字段。"""
    if not isinstance(goods, dict):
        return None
    return {
        key: value
        for key, value in goods.items()
        if key
        in {
            "commodityId",
            "commodity_id",
            "commodityName",
            "name",
            "webPic",
            "pic",
            "score",
            "status",
        }
    }


def slim_douyu_limited_goods(goods: object) -> list[dict[str, object]]:
    """限兑列表只返回前几个商品名，避免任务列表携带完整原始数组。"""
    if not isinstance(goods, list):
        return []
    result: list[dict[str, object]] = []
    for item in goods[:5]:
        if not isinstance(item, dict):
            continue
        result.append(
            {
                key: value
                for key, value in item.items()
                if key in {"commodityId", "commodity_id", "commodityName", "name"}
            }
        )
    return result


def strip_douyu_raw_snapshots(value: object) -> object:
    """递归移除任务结果中的原始接口快照。"""
    if isinstance(value, dict):
        return {
            key: strip_douyu_raw_snapshots(item)
            for key, item in value.items()
            if key != "raw"
        }
    if isinstance(value, list):
        return [strip_douyu_raw_snapshots(item) for item in value]
    return value


def sanitize_douyu_task_result(
    result: dict | None, task_type: str, *, include_detail: bool = False
) -> dict | None:
    """列表/实时推送接口剥离原始快照与大数组；详情接口保留完整 result。"""
    if not isinstance(result, dict):
        return result
    if include_detail:
        return result

    data = strip_douyu_raw_snapshots(result)
    if not isinstance(data, dict):
        return None
    removable_keys = {
        "bind_info",
        "before_bind_info",
        "bind_candidates",
        "cooldown_bind_info",
        "after_bind_info",
        "activity_bind_snapshot",
        "activity_bind_info",
        "points_query",
        "exchange_balance_query",
        "query_act_aliases",
    }
    if task_type != "query_xpd_purchase_records":
        removable_keys.add("records")
    for key in removable_keys:
        data.pop(key, None)

    # 和平小店任务 result 内嵌 role（含 gameOpenId 等），列表接口剥离其原始快照
    role = data.get("role")
    if isinstance(role, dict):
        data["role"] = strip_douyu_raw_snapshots(role)

    goods = data.get("goods")
    if isinstance(goods, list):
        data.pop("goods", None)
    elif isinstance(goods, dict):
        data["goods"] = slim_douyu_goods(goods)

    if "limited_goods" in data:
        data["limited_goods"] = slim_douyu_limited_goods(data.get("limited_goods"))

    if task_type not in {
        "get_bind_qr",
        "prepare_esports_bind",
        "get_esports_bind_qr",
        "get_xpd_bind_qr",
    }:
        data.pop("url", None)
    if task_type not in {"create_elite_qr", "create_esports_qr", "create_gold_qr"}:
        data.pop("pay_url", None)
    return data


def douyu_task_payload(task: DouyuTask, *, include_detail: bool = False) -> dict:
    """生成 REST 与 WebSocket 共用的斗鱼任务载荷。"""
    account = task.account

    def isoformat(value: datetime | None) -> str | None:
        if value is None:
            return None
        if value.tzinfo is None:
            value = value.replace(tzinfo=UTC)
        return value.isoformat()

    return {
        "id": task.id,
        "batch_id": task.batch_id,
        "account_id": task.account_id,
        "account_username": (account.username if account else "") or "",
        "account_uid": (account.uid if account else "") or "",
        "account_nickname": (account.nickname if account else "") or "",
        "task_type": task.task_type or "",
        "handbook_scope": task.handbook_scope or "legacy",
        "status": task.status or "",
        "message": task.message or "",
        "result": sanitize_douyu_task_result(
            task.result if isinstance(task.result, dict) else None,
            task.task_type or "",
            include_detail=include_detail,
        ),
        "created_by": task.created_by,
        "created_at": isoformat(task.created_at),
        "finished_at": isoformat(task.finished_at),
    }