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

"""斗鱼任务执行器：商城兑换（由 douyu_runner.py 按功能域拆分）。"""

from __future__ import annotations

import random
import time
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from sqlalchemy.orm import Session

from core.douyu import DouyuActivityClient, DouyuActivityError

from ..models import Account, DouyuEsportsGoodsSnapshot, DouyuGoodsSnapshot, DouyuTask
from .business_events import record_exchange

# 兑换节奏与重试（对齐 8.30 浏览器抓包：锁单->支付间隔约 2.2~4.2s；火爆类错误要长退避而不是秒级连打）
DOUYU_EXCHANGE_PRE_CREATE_JITTER = (0.3, 1.2)
DOUYU_EXCHANGE_LOCK_PAY_DELAY_RANGE = (2.0, 4.0)
DOUYU_EXCHANGE_LOCK_TTL_SECONDS = 300
DOUYU_EXCHANGE_RATE_LIMIT_HINTS = (
    "火爆",
    "频繁",
    "稍后",
    "太热",
    "限流",
    "繁忙",
    "人多",
    "排队",
    "手慢",
)
DOUYU_EXCHANGE_RATE_LIMIT_BACKOFFS = (30, 60, 90)
DOUYU_EXCHANGE_GENERIC_BACKOFFS = (2, 5, 10, 20)


class GoodsMixin:
    """商城兑换域：商品刷新、锁单/支付/兑换、兑换节奏与退避重试。"""

    if TYPE_CHECKING:
        _stop: Any

        def _client(self, cookie: str) -> DouyuActivityClient: ...
        def _task_payload(self, task: DouyuTask) -> dict: ...
        def _upsert_goods(self, db: Session, goods: list[dict]) -> None: ...
        def _upsert_esports_goods(self, db: Session, goods: list[dict]) -> None: ...
        def _refresh_account_points(
            self,
            client: DouyuActivityClient,
            account: Account,
            cookie: str,
            *,
            ctn: str = "",
        ) -> dict: ...
        def _refresh_esports_handbook(
            self, client: DouyuActivityClient, account: Account, *, manual_id: str
        ) -> dict: ...
        def _sleep_interruptible(self, seconds: float) -> bool: ...
        def _push_log(self, level: str, message: str) -> None: ...
        def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...

    def _execute_refresh_goods(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        client = self._client(cookie)
        result = client.list_goods(manual_id=config["manual_id"], rid=config["rid"])
        goods = result["goods"]
        self._upsert_goods(db, goods)
        account.updated_at = datetime.now(UTC)
        self._mark_task(
            db,
            task,
            "success",
            f"已刷新商品 {len(goods)} 个",
            {"goods_count": len(goods), "goods": goods},
        )

    def _execute_refresh_esports_goods(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """刷新电竞手册皮肤商城快照。"""
        client = self._client(cookie)
        result = client.list_esports_goods(
            manual_id=str(config["esports_manual_id"]),
            rid=str(config["room_id"]),
        )
        goods = result["goods"]
        self._upsert_esports_goods(db, goods)
        account.updated_at = datetime.now(UTC)
        self._mark_task(
            db,
            task,
            "success",
            f"已刷新电竞皮肤 {len(goods)} 个",
            {
                "goods_count": len(goods),
                "esports_store_score": result["score"],
                "goods": goods,
            },
        )

    def _execute_lock_goods(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        payload = self._task_payload(task)
        commodity_id = str(
            payload.get("commodity_id") or payload.get("commodityId") or ""
        ).strip()
        if not commodity_id:
            self._mark_task(db, task, "failed", "请选择锁定商品")
            return
        try:
            num = int(payload.get("num") or 1)
        except (TypeError, ValueError):
            num = 1
        client = self._client(cookie)
        result = client.create_exchange_order(
            manual_id=str(config["manual_id"]),
            rid=str(config["rid"]),
            commodity_id=commodity_id,
            num=max(1, num),
        )
        goods = (
            db.query(DouyuGoodsSnapshot)
            .filter(DouyuGoodsSnapshot.commodity_id == commodity_id)
            .first()
        )
        account.updated_at = datetime.now(UTC)
        expire_seconds = result.get("expire_seconds")
        expire_text = f"，{expire_seconds} 秒内有效" if expire_seconds else ""
        self._mark_task(
            db,
            task,
            "success",
            f"锁单成功: {(goods.name if goods else '') or commodity_id}（订单 {result['order_id']}{expire_text}）",
            {
                "goods": goods.raw if goods else None,
                "game_name": account.game_name or "",
                "game_channel": account.game_channel or "",
                "account_name": account.nickname
                or account.username
                or account.uid
                or f"#{account.id}",
                **result,
            },
        )

    def _execute_pay_locked_order(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        payload = self._task_payload(task)
        order_id = str(payload.get("order_id") or payload.get("orderId") or "").strip()
        commodity_id = str(
            payload.get("commodity_id") or payload.get("commodityId") or ""
        ).strip()
        if not order_id:
            self._mark_task(db, task, "failed", "锁单订单号不能为空")
            return
        client = self._client(cookie)
        payment = client.pay_exchange_order(
            manual_id=str(config["manual_id"]),
            order_id=order_id,
            rid=str(config.get("rid") or ""),
        )
        goods = None
        if commodity_id:
            goods = (
                db.query(DouyuGoodsSnapshot)
                .filter(DouyuGoodsSnapshot.commodity_id == commodity_id)
                .first()
            )
        account.updated_at = datetime.now(UTC)
        result = {
            "commodity_id": commodity_id,
            "order_id": order_id,
            "exchange_id": payment["exchange_id"],
            "commodity_image": payment["commodity_image"],
            "exchange_num": payment["exchange_num"],
            "payment": payment,
            "goods": goods.raw if goods else None,
            "game_name": account.game_name or "",
            "game_channel": account.game_channel or "",
            "account_name": account.nickname
            or account.username
            or account.uid
            or f"#{account.id}",
        }
        try:
            ctn = client.acf_ccn(refresh_subscribe=False)
            result.update(
                self._refresh_account_points(client, account, cookie, ctn=ctn)
            )
        except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            self._push_log("warning", f"支付锁单后刷新积分失败: {exc}")
        # 两步式兑换（lock_goods → pay_locked_order）的**支付**这一步才是真兑换。
        # 这一步此前从不发事件，时间线上只有「锁单」没有「支付」。
        # `pay_exchange_order` 的响应里没有 score，用商品快照上的标价。
        score = None
        if goods is not None and isinstance(goods.score, (int, float)):
            num = payment.get("exchange_num")
            score = int(goods.score) * (
                int(num) if isinstance(num, (int, float)) else 1
            )
        record_exchange(
            db,
            account,
            task_id=task.id,
            name=(goods.name if goods else "") or commodity_id or order_id,
            score=score,
            commodity_id=commodity_id,
            order_id=order_id,
            exchange_id=str(payment.get("exchange_id") or ""),
            game_name=account.game_name or "",
            game_channel=account.game_channel or "",
            points_after=account.points,
            scope="elite",
        )
        self._mark_task(
            db,
            task,
            "success",
            f"锁单支付成功: {(goods.name if goods else '') or commodity_id or order_id}",
            result,
        )

    @staticmethod
    def _is_rate_limited_error(message: str) -> bool:
        """识别频控/火爆类错误文案，命中时按长退避重试。"""
        return any(hint in message for hint in DOUYU_EXCHANGE_RATE_LIMIT_HINTS)

    def _pay_exchange_with_backoff(
        self,
        client: DouyuActivityClient,
        *,
        manual_id: str,
        order_id: str,
        rid: str,
        locked_at: float,
    ) -> dict | None:
        """支付锁单，按错误类型退避重试（不再 0.3s 无脑连打）。

        频控/火爆类错误：30/60/90 秒退避，最多 4 次尝试且不超出锁单有效期；
        其他错误：2/5/10/20 秒退避，最多 5 次尝试。
        """
        for attempt in range(6):
            if self._stop.is_set():
                return None
            try:
                return client.pay_exchange_order(
                    manual_id=manual_id,
                    order_id=order_id,
                    rid=rid,
                )
            except DouyuActivityError as exc:
                last_error = str(exc)
                if self._is_rate_limited_error(last_error):
                    if attempt >= len(DOUYU_EXCHANGE_RATE_LIMIT_BACKOFFS):
                        raise DouyuActivityError(
                            f"锁单 {order_id} 支付被频控拦截(已退避重试{attempt}次): {last_error}"
                        ) from exc
                    backoff = DOUYU_EXCHANGE_RATE_LIMIT_BACKOFFS[attempt]
                else:
                    if attempt >= len(DOUYU_EXCHANGE_GENERIC_BACKOFFS):
                        raise DouyuActivityError(
                            f"锁单 {order_id} 支付失败(已重试{attempt}次): {last_error}"
                        ) from exc
                    backoff = DOUYU_EXCHANGE_GENERIC_BACKOFFS[attempt]
                # 超出锁单有效期（服务端 300s）前留 15s 余量，放弃继续重试
                remaining = locked_at + DOUYU_EXCHANGE_LOCK_TTL_SECONDS - time.time()
                if remaining - backoff < 15:
                    raise DouyuActivityError(
                        f"锁单 {order_id} 支付失败且临近过期({int(remaining)}s): {last_error}"
                    ) from exc
                kind = "频控" if self._is_rate_limited_error(last_error) else "错误"
                self._push_log(
                    "info",
                    f"  锁单 {order_id} 支付{kind}退避 {backoff:.0f}s 后重试 {attempt + 1}/"
                    f"{len(DOUYU_EXCHANGE_RATE_LIMIT_BACKOFFS) if self._is_rate_limited_error(last_error) else len(DOUYU_EXCHANGE_GENERIC_BACKOFFS) + 1}: {last_error}",
                )
                if not self._sleep_interruptible(backoff):
                    return None
        return None

    def _execute_exchange_goods(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        payload = self._task_payload(task)
        commodity_id = str(
            payload.get("commodity_id") or payload.get("commodityId") or ""
        ).strip()
        if not commodity_id:
            self._mark_task(db, task, "failed", "请选择兑换商品")
            return
        client = self._client(cookie)
        manual_id = str(config["manual_id"])
        rid = str(config.get("rid") or "")
        result: dict = {"commodity_id": commodity_id}

        # ---- 对齐浏览器：先查手册状态 + 商品详情，按页面状态机选兑换链路 ----
        manual_type = None
        try:
            user_info = client.elite_user_info(manual_id=manual_id, rid=rid)
            manual_type = user_info.get("manual_type")
        except DouyuActivityError as exc:
            self._push_log("warning", f"查询手册状态失败，按经典流程继续: {exc}")
        detail = None
        try:
            detail = client.query_goods_detail(
                manual_id=manual_id, commodity_id=commodity_id, rid=rid
            )["detail"]
        except DouyuActivityError as exc:
            self._push_log("warning", f"查询商品详情失败，按经典流程继续: {exc}")

        batch_num_limit = None
        if detail and int(detail.get("batchExchange") or 0) > 0:
            try:
                batch_num_limit = client.batch_exchange_limit(
                    manual_id=manual_id, commodity_id=commodity_id, rid=rid
                )["limit"]
            except DouyuActivityError as exc:
                self._push_log("warning", f"查询批量兑换上限失败: {exc}")

        plan = DouyuActivityClient.resolve_exchange_plan(
            detail or {}, manual_type=manual_type, batch_num_limit=batch_num_limit
        )
        result["plan"] = plan
        if detail:
            result["detail"] = detail
        action = plan["action"]

        def finish_failed(message: str) -> None:
            self._mark_task(
                db,
                task,
                "failed",
                message,
                {"commodity_id": commodity_id, "plan": plan, "detail": detail},
            )

        if action in ("blocked", "wait"):
            finish_failed(f"兑换失败: {plan['text']}")
            return

        goods = (
            db.query(DouyuGoodsSnapshot)
            .filter(DouyuGoodsSnapshot.commodity_id == commodity_id)
            .first()
        )
        name = (goods.name if goods else "") or commodity_id

        if action == "subscribe":
            try:
                client.subscribe_commodity(
                    manual_id=manual_id, commodity_id=commodity_id
                )
            except DouyuActivityError as exc:
                finish_failed(f"预约到货失败: {exc}")
                return
            account.updated_at = datetime.now(UTC)
            self._mark_task(
                db,
                task,
                "success",
                f"已预约到货: {name} ({plan['text']})",
                {"commodity_id": commodity_id, "plan": plan, "detail": detail},
            )
            return

        if action == "pre_exchange":
            try:
                check = client.pre_exchange_check(
                    manual_id=manual_id, commodity_id=commodity_id
                )
                user_status = int((check.get("data") or {}).get("userStatus") or 0)
                if user_status == 1 and not plan["pre_exchange"]:
                    self._mark_task(
                        db,
                        task,
                        "success",
                        f"已预兑: {name}，等待开放后继续兑换",
                        {
                            "commodity_id": commodity_id,
                            "plan": plan,
                            "detail": detail,
                            "check": check,
                        },
                    )
                    return
                confirm = client.confirm_pre_exchange(
                    manual_id=manual_id, commodity_id=commodity_id, rid=rid
                )
            except DouyuActivityError as exc:
                finish_failed(f"预兑失败: {exc}")
                return
            account.updated_at = datetime.now(UTC)
            self._mark_task(
                db,
                task,
                "success",
                f"预兑成功: {name}，开放后可继续兑换",
                {
                    "commodity_id": commodity_id,
                    "plan": plan,
                    "detail": detail,
                    "check": check,
                    "confirm": confirm,
                },
            )
            return

        # ---- 经典链路：锁单 -> 人类节奏等待 -> 支付（错误分类退避重试）----
        try:
            num = max(1, int(payload.get("num") or 1))
        except (TypeError, ValueError):
            num = 1
        num = min(num, int(plan.get("max_num") or 1))

        # 少量随机抖动（浏览器点击商品到锁单之间有人工间隔）
        self._push_log("info", f"  兑换 {name}: {plan['text']}")
        if not self._sleep_interruptible(
            random.uniform(*DOUYU_EXCHANGE_PRE_CREATE_JITTER)
        ):
            self._mark_task(db, task, "stopped", "任务已停止", result)
            return
        try:
            locked = client.create_exchange_order(
                manual_id=manual_id,
                rid=rid,
                commodity_id=commodity_id,
                num=num,
            )
        except DouyuActivityError as exc:
            # 锁单本身被频控时做一次长退避重试，避免无脑连打
            backoff = (
                DOUYU_EXCHANGE_RATE_LIMIT_BACKOFFS[0]
                if self._is_rate_limited_error(str(exc))
                else 3
            )
            self._push_log(
                "info", f"  锁定兑换商品失败({exc})，{backoff:.0f}s 后重试一次"
            )
            if not self._sleep_interruptible(backoff):
                self._mark_task(db, task, "stopped", "任务已停止", result)
                return
            try:
                locked = client.create_exchange_order(
                    manual_id=manual_id,
                    rid=rid,
                    commodity_id=commodity_id,
                    num=num,
                )
            except DouyuActivityError as exc2:
                finish_failed(f"锁定兑换商品失败: {exc2}")
                return
        result.update(
            {**locked, "lock_order": locked, "goods": goods.raw if goods else None}
        )

        # 锁单成功后按浏览器节奏（抓包实测 2.24s~4.2s）等待再支付
        delay = random.uniform(*DOUYU_EXCHANGE_LOCK_PAY_DELAY_RANGE)
        self._push_log(
            "info", f"  锁单成功(订单 {locked['order_id']})，{delay:.1f}s 后支付"
        )
        if not self._sleep_interruptible(delay):
            self._mark_task(
                db, task, "stopped", "任务已停止，商品锁单仍可能有效", locked
            )
            return
        try:
            payment = self._pay_exchange_with_backoff(
                client,
                manual_id=manual_id,
                order_id=locked["order_id"],
                rid=rid,
                locked_at=locked["locked_at"],
            )
        except DouyuActivityError as exc:
            finish_failed(str(exc))
            return
        if payment is None:
            self._mark_task(
                db, task, "stopped", "任务已停止，商品锁单仍可能有效", locked
            )
            return

        result.update(
            {
                "order_id": locked["order_id"],
                "exchange_id": payment["exchange_id"],
                "commodity_image": payment["commodity_image"]
                or locked["commodity_image"],
                "exchange_num": payment["exchange_num"],
                "payment": payment,
            }
        )
        account.updated_at = datetime.now(UTC)
        # 兑换成功后自动刷新积分，更新账号最新积分信息（失败不阻断兑换成功）
        points_refresh = None
        try:
            ctn = client.acf_ccn(refresh_subscribe=False)
            points_refresh = self._refresh_account_points(
                client, account, cookie, ctn=ctn
            )
        except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            self._push_log("warning", f"兑换后刷新积分失败: {exc}")
        # ⚠️ **只有这一条 success 是真的兑换**。同一个 handler 里还有三条
        # 「预约到货 / 已预兑 / 预兑确认」的 success —— 它们不消耗积分、也没换到东西，
        # 所以**不发事件**。旧收集器只按 status=="success" 收集，导致每预约一次
        # 就生成一条金额为空的 exchange，把 accounts 的 exchange_count 灌大。
        record_exchange(
            db,
            account,
            task_id=task.id,
            name=name,
            score=locked.get("score"),
            commodity_id=commodity_id,
            order_id=str(locked.get("order_id") or ""),
            exchange_id=str(payment.get("exchange_id") or ""),
            game_name=account.game_name or "",
            game_channel=account.game_channel or "",
            points_after=(points_refresh or {}).get("points"),
            scope="elite",
        )
        self._mark_task(
            db,
            task,
            "success",
            f"兑换成功: {name}",
            {
                "goods": goods.raw if goods else None,
                "game_name": account.game_name or "",
                "game_channel": account.game_channel or "",
                "account_name": account.nickname
                or account.username
                or account.uid
                or f"#{account.id}",
                **result,
                **(points_refresh or {}),
            },
        )

    def _execute_exchange_esports_goods(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """兑换电竞手册皮肤，并同步电竞积分。"""
        payload = self._task_payload(task)
        commodity_id = str(
            payload.get("commodity_id") or payload.get("commodityId") or ""
        ).strip()
        if not commodity_id:
            self._mark_task(db, task, "failed", "请选择电竞皮肤")
            return
        try:
            quantity = max(1, int(payload.get("quantity") or payload.get("num") or 1))
        except (TypeError, ValueError):
            self._mark_task(db, task, "failed", "兑换数量必须是正整数")
            return

        manual_id = str(config.get("esports_manual_id") or "").strip()
        room_id = str(config.get("room_id") or "").strip()
        if not manual_id or not room_id:
            self._mark_task(db, task, "failed", "请先配置电竞手册 manualID 和房间 ID")
            return

        client = self._client(cookie)
        baseline_points = account.esports_points
        try:
            baseline = self._refresh_esports_handbook(
                client, account, manual_id=manual_id
            )
            baseline_points = baseline["esports_points"]
            db.commit()
        except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            self._push_log("warning", f"兑换电竞皮肤前刷新积分失败: {exc}")

        result = client.exchange_esports_goods(
            manual_id=manual_id,
            rid=room_id,
            commodity_id=commodity_id,
            quantity=quantity,
        )
        goods = (
            db.query(DouyuEsportsGoodsSnapshot)
            .filter(DouyuEsportsGoodsSnapshot.commodity_id == commodity_id)
            .first()
        )
        result["goods"] = goods.raw if goods else None
        result["esports_points_baseline"] = baseline_points
        try:
            points_result = self._refresh_esports_handbook(
                client, account, manual_id=manual_id
            )
            result.update(points_result)
            result["esports_points_after_exchange"] = points_result["esports_points"]
        except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            result["esports_points_refresh_error"] = str(exc)

        account.updated_at = datetime.now(UTC)
        name = (goods.name if goods else "") or commodity_id
        message = f"兑换电竞皮肤成功: {name}"
        if quantity > 1:
            message += f" x{quantity}"
        if account.esports_points is not None:
            message += f"，电竞积分: {account.esports_points}"
        result["game_name"] = account.esports_game_name or ""
        result["game_channel"] = account.esports_game_channel or ""
        result["account_name"] = (
            account.nickname or account.username or account.uid or f"#{account.id}"
        )
        # 电竞皮肤兑换 —— **此前从不发事件**（不在旧收集器的 allowlist 里），
        # 时间线上一直是空白。
        # 消耗的电竞积分用「兑换前后积分差」实测，而不是读猜测的字段。
        after_points = result.get("esports_points_after_exchange")
        spent = (
            int(baseline_points) - int(after_points)
            if isinstance(baseline_points, (int, float))
            and isinstance(after_points, (int, float))
            else None
        )
        record_exchange(
            db,
            account,
            task_id=task.id,
            name=name,
            score=spent if spent and spent > 0 else None,
            commodity_id=commodity_id,
            game_name=account.esports_game_name or "",
            game_channel=account.esports_game_channel or "",
            points_after=account.esports_points,
            scope="esports",
        )
        self._mark_task(db, task, "success", message, result)

    def _execute_query_limited_goods(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        client = self._client(cookie)
        result = client.query_limited_goods(
            manual_id=str(config["manual_id"]), rid=str(config["rid"])
        )
        limited = result["limited_goods"]
        names = [
            str(item.get("commodityName") or "")
            for item in limited
            if item.get("commodityName")
        ]
        message = (
            "无限制商品"
            if not names
            else f"限兑 {len(names)} 个: {', '.join(names[:5])}"
        )
        account.updated_at = datetime.now(UTC)
        self._mark_task(
            db,
            task,
            "success",
            message,
            {"limited_count": len(limited), "limited_goods": limited},
        )

    def _execute_query_exchange_records(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        client = self._client(cookie)
        result = client.exchange_records(manual_id=str(config["manual_id"]))
        records = result["records"]
        account.updated_at = datetime.now(UTC)
        self._mark_task(
            db,
            task,
            "success",
            f"兑换记录 {len(records)} 条" if records else "暂无兑换记录",
            result,
        )

    def _execute_prefetch_csrf_token(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        client = self._client(cookie)
        token = client.csrf_token()
        account.updated_at = datetime.now(UTC)
        self._mark_task(
            db,
            task,
            "success",
            "获取 csrf_token 成功",
            {"csrf_token": token, "cookie": client.cookie},
        )