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

"""斗鱼任务执行器：和平小店（由 douyu_runner.py 按功能域拆分）。"""

from __future__ import annotations

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

from sqlalchemy.orm import Session

from core.douyu import DouyuActivityClient, DouyuActivityError

from ..models import Account, DouyuTask, DouyuXpdGoodsSnapshot
from .account_binding import (
    BIND_BOUND,
    BIND_PENDING,
    BIND_UNBOUND,
    SCOPE_PEACE,
    set_bind_status,
)
from .business_events import record_exchange, task_actor

DOUYU_XPD_BIND_POLL_SECONDS = 300
DOUYU_XPD_BIND_POLL_INTERVAL = 5


class XpdMixin:
    """和平小店域：绑定、商品、余额、碎片与兑换。"""

    if TYPE_CHECKING:
        _stop: Any

        @staticmethod
        def _to_int(value: Any) -> int | None: ...

        def _client(self, cookie: str) -> DouyuActivityClient: ...
        def _task_payload(self, task: DouyuTask) -> dict: ...
        def _upsert_xpd_goods(self, db: Session, goods: list[dict]) -> None: ...
        def _push_log(self, level: str, message: str) -> None: ...
        def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
        def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ...

    def _xpd_role_context(self, client: DouyuActivityClient, config: dict) -> dict:
        """获取小店 H5 参数 + 绑定角色信息，小店任务共用。"""
        embed = client.xpd_embed_query(
            act_alias=str(config["xpd_act_alias"]),
            rid=str(config["xpd_rid"]),
        )
        role = client.xpd_get_role(
            embed_query=embed["query"],
            act_id=str(config["xpd_act_id"]),
            rid=str(config["xpd_rid"]),
        )
        return {"embed": embed, "role": role}

    def _xpd_area_id(self, role: dict, account: Account) -> int:
        """角色大区: 优先使用接口值，微信=1、手Q=2，未知回退已存值。"""
        raw_area = role.get("area")
        if raw_area not in (None, ""):
            try:
                area_id = int(raw_area)
            except (TypeError, ValueError):
                area_id = 0
            if area_id > 0:
                return area_id
        role_type = str(role.get("type") or "")
        if role_type == "wx":
            return 1
        if role_type == "qq":
            return 2
        return account.xpd_area_id or 1

    def _apply_xpd_role_to_account(
        self, account: Account, role: dict, area_id: int
    ) -> None:
        account.xpd_game_name = (
            str(role.get("role_name") or "") or account.xpd_game_name
        )
        account.xpd_openid = str(role.get("game_open_id") or "") or account.xpd_openid
        account.xpd_role_id = str(role.get("role_id") or "") or account.xpd_role_id
        account.xpd_plat_id = self._to_int(role.get("plat_id"))
        account.xpd_area_id = area_id
        account.updated_at = datetime.now(UTC)

    def _execute_query_xpd_role(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """查询和平小店绑定角色。"""
        client = self._client(cookie)
        ctx = self._xpd_role_context(client, config)
        role = ctx["role"]
        if not role.get("role_id"):
            self._mark_task(db, task, "failed", "未获取到小店绑定角色")
            return
        area_id = self._xpd_area_id(role, account)
        self._apply_xpd_role_to_account(account, role, area_id)
        set_bind_status(account, SCOPE_PEACE, BIND_BOUND)
        db.commit()
        role_text = str(role.get("role_name") or "-")
        channel = (
            "微信"
            if role.get("type") == "wx"
            else ("手Q" if role.get("type") == "qq" else str(role.get("type") or "-"))
        )
        self._mark_task(
            db,
            task,
            "success",
            f"小店角色: {role_text}（{channel}）",
            {"role": role, "area_id": area_id},
        )

    def _execute_get_xpd_bind_qr(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """生成和平小店绑定二维码并轮询等待微信扫码绑定/换绑完成。

        识别到新角色后仅标记"待确认"，不自动回写账号，由用户手动确认绑定。
        """
        client = self._client(cookie)
        act_alias = str(config.get("xpd_act_alias") or "").strip()
        if not act_alias:
            self._mark_task(db, task, "failed", "请先配置小店活动代号 actAlias")
            return
        result = client.xpd_bind_qr(act_alias=act_alias)
        set_bind_status(account, SCOPE_PEACE, BIND_PENDING)
        account.updated_at = datetime.now(UTC)
        db.commit()
        # 记录绑定前状态：已绑定账号生成二维码后必须等扫码换绑，不能立即成功
        try:
            before = client.xpd_bind_info(act_alias=act_alias)
            result["before_bound"] = bool(before.get("bind_role"))
            result["before_role_name"] = str(before.get("role_name") or "")
            result["before_area_name"] = str(before.get("area_name") or "")
            result["before_plat_name"] = str(before.get("plat_name") or "")
        except Exception:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            result["before_bound"] = False
            result["before_role_name"] = ""
        result["bind_polling"] = True
        self._update_task_progress(
            db,
            task,
            "running",
            "二维码已生成，请微信扫码在小程序中绑定角色",
            result,
        )
        state = self._wait_xpd_bind(db, task, client, act_alias, result)
        result["bind_polling"] = False
        if state == "stopped":
            self._mark_task(db, task, "stopped", "任务已停止", result)
            return
        if state == "pending":
            role_text = str(result.get("role_name") or "-")
            self._mark_task(
                db, task, "success", f"已识别角色: {role_text}，待确认绑定", result
            )
            return
        self._mark_task(
            db,
            task,
            "failed",
            "未检测到小店绑定（二维码仍有效，可再次生成后扫码）",
            result,
        )

    def _wait_xpd_bind(
        self,
        db: Session,
        task: DouyuTask,
        client: DouyuActivityClient,
        act_alias: str,
        result: dict,
    ) -> str:
        """轮询 bindInfo 检测绑定/换绑角色，识别到后停在"待确认"，不自动回写账号。

        - 绑定前未绑定：检测到 bind_role=1 即识别到待确认角色
        - 绑定前已绑定（换绑）：检测到角色名变化才算换绑完成，角色不变继续等
        返回 "pending"=已识别待确认角色, "stopped"=任务停止, "timeout"=超时未识别。
        """
        before_bound = bool(result.get("before_bound"))
        before_role_name = str(result.get("before_role_name") or "")
        deadline = time.monotonic() + DOUYU_XPD_BIND_POLL_SECONDS
        poll_count = 0
        while not self._stop.is_set() and time.monotonic() <= deadline:
            try:
                info = client.xpd_bind_info(act_alias=act_alias)
                poll_count += 1
                result["bind_poll_count"] = poll_count
                result["bind_polling"] = True
                role_name = str(info.get("role_name") or "")
                bound_now = bool(info.get("bind_role"))
                changed = (
                    before_bound and bool(role_name) and role_name != before_role_name
                )
                if (not before_bound and bound_now and role_name) or changed:
                    result.update(
                        {key: value for key, value in info.items() if key != "raw"}
                    )
                    result["bind_polling"] = False
                    result["xpd_pending_confirm"] = True
                    return "pending"
                self._update_task_progress(
                    db,
                    task,
                    "running",
                    f"等待扫码绑定（第 {poll_count} 次）",
                    result,
                )
            except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
                poll_count += 1
                result["bind_poll_count"] = poll_count
                result["bind_poll_error"] = str(exc)
                self._update_task_progress(
                    db,
                    task,
                    "running",
                    f"等待扫码绑定: {exc}",
                    result,
                )
            if self._stop.wait(DOUYU_XPD_BIND_POLL_INTERVAL):
                break
        result["bind_polling"] = False
        return "stopped" if self._stop.is_set() else "timeout"

    def _execute_confirm_xpd_bind(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """确认和平小店绑定：回查 bindInfo，确认绑定角色后将账号回写为已绑定。"""
        client = self._client(cookie)
        act_alias = str(config.get("xpd_act_alias") or "").strip()
        if not act_alias:
            self._mark_task(db, task, "failed", "请先配置小店活动代号 actAlias")
            return
        result = client.xpd_bind_info(act_alias=act_alias)
        role_name = str(result.get("role_name") or "")
        if not result.get("bind_role") or not role_name:
            set_bind_status(account, SCOPE_PEACE, BIND_UNBOUND)
            account.updated_at = datetime.now(UTC)
            db.commit()
            self._mark_task(
                db, task, "failed", "尚未检测到小店绑定角色，请先扫码绑定", result
            )
            return
        # 优先用完整角色信息回写（与查询角色一致），失败时回退 bindInfo 角色名
        try:
            ctx = self._xpd_role_context(client, config)
            role = ctx["role"]
            if role.get("role_id"):
                self._apply_xpd_role_to_account(
                    account, role, self._xpd_area_id(role, account)
                )
            else:
                account.xpd_game_name = role_name
                account.updated_at = datetime.now(UTC)
        except Exception:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            account.xpd_game_name = role_name
            account.updated_at = datetime.now(UTC)
        set_bind_status(account, SCOPE_PEACE, BIND_BOUND)
        account.updated_at = datetime.now(UTC)
        db.commit()
        result["xpd_pending_confirm"] = False
        result["xpd_bound"] = True
        self._mark_task(db, task, "success", f"小店绑定成功: {role_name}", result)

    def _execute_query_xpd_bind_info(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """查询和平小店绑定信息（bindInfo）。

        仅查询展示，不回写账号；确认绑定由 confirm_xpd_bind 任务完成。
        """
        client = self._client(cookie)
        act_alias = str(config.get("xpd_act_alias") or "").strip()
        if not act_alias:
            self._mark_task(db, task, "failed", "请先配置小店活动代号 actAlias")
            return
        result = client.xpd_bind_info(act_alias=act_alias)
        status = "已绑定" if result.get("bind_role") else "未绑定"
        text = str(result.get("role_name") or result.get("nick") or "-")
        self._mark_task(
            db,
            task,
            "success",
            f"小店绑定: {status} ({text})",
            result,
        )

    def _execute_refresh_xpd_goods(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """刷新和平小店商品列表快照（全局数据，任一可用 CK 即可）。"""
        client = self._client(cookie)
        ctx = self._xpd_role_context(client, config)
        role = ctx["role"]
        if not role.get("role_id"):
            self._mark_task(db, task, "failed", "未获取到小店绑定角色")
            return
        area_id = self._xpd_area_id(role, account)
        result = client.xpd_list_goods(
            embed_query=ctx["embed"]["query"],
            act_id=str(config["xpd_act_id"]),
            openid=str(role.get("game_open_id") or ""),
            roleid=str(role.get("role_id") or ""),
            areaid=str(area_id),
        )
        goods = result["goods"]
        self._upsert_xpd_goods(db, goods)
        self._apply_xpd_role_to_account(account, role, area_id)
        db.commit()
        self._mark_task(
            db,
            task,
            "success",
            f"已刷新小店商品 {len(goods)} 个",
            {"goods_count": len(goods), "goods": goods},
        )

    def _execute_query_xpd_balance(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """查询和平小店点券余额。"""
        client = self._client(cookie)
        ctx = self._xpd_role_context(client, config)
        role = ctx["role"]
        if not role.get("role_id"):
            self._mark_task(db, task, "failed", "未获取到小店绑定角色")
            return
        area_id = self._xpd_area_id(role, account)
        role_plat = role.get("plat_id")
        plat = str(role_plat) if role_plat not in (None, "") else "1"
        result = client.xpd_balance(
            embed_query=ctx["embed"]["query"],
            act_id=str(config["xpd_act_id"]),
            openid=str(role.get("game_open_id") or ""),
            roleid=str(role.get("role_id") or ""),
            plat=plat,
            areaid=str(area_id),
        )
        balance = result.get("balance")
        self._apply_xpd_role_to_account(account, role, area_id)
        account.xpd_balance = balance
        db.commit()
        if balance is None:
            self._mark_task(db, task, "failed", "未获取到小店点券余额")
            return
        self._mark_task(
            db,
            task,
            "success",
            f"小店点券余额: {balance}",
            {"balance": balance, "role": role, "area_id": area_id},
        )

    def _execute_query_xpd_fragments(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """查询和平小店扭蛋碎片数量。

        优先现查角色；getrole 受限（Livelink 风控/失效）时回退账号已存角色，
        保证已绑定账号仍可查询。
        """
        client = self._client(cookie)
        act_id = str(config["xpd_act_id"])
        embed_query: dict = {}
        openid = str(account.xpd_openid or "")
        roleid = str(account.xpd_role_id or "")
        stored_plat = account.xpd_plat_id
        plat = str(stored_plat) if stored_plat is not None else "1"
        areaid = str(account.xpd_area_id or 1)
        role: dict = {}
        try:
            ctx = self._xpd_role_context(client, config)
            embed_query = ctx["embed"]["query"]
            role = ctx["role"] if isinstance(ctx.get("role"), dict) else {}
            if role.get("role_id"):
                role_area = self._xpd_area_id(role, account)
                openid = str(role.get("game_open_id") or "") or openid
                roleid = str(role.get("role_id") or "") or roleid
                role_plat = role.get("plat_id")
                plat = str(role_plat) if role_plat not in (None, "") else plat
                areaid = str(role_area) or areaid
                self._apply_xpd_role_to_account(account, role, role_area)
        except Exception as exc:  # noqa: BLE001
            self._push_log("debug", f"小店角色信息补全失败: {exc}")
        if not openid or not roleid:
            self._mark_task(
                db, task, "failed", "未获取到小店绑定角色，请先生成二维码扫码绑定"
            )
            return
        result = client.xpd_fragments(
            embed_query=embed_query,
            act_id=act_id,
            openid=openid,
            roleid=roleid,
            plat=plat,
            areaid=areaid,
        )
        fragments = result.get("fragments")
        account.xpd_fragments = fragments
        account.updated_at = datetime.now(UTC)
        db.commit()
        if fragments is None:
            self._mark_task(db, task, "failed", "未获取到小店扭蛋碎片数量")
            return
        self._mark_task(
            db,
            task,
            "success",
            f"小店扭蛋碎片: {fragments}",
            {"fragments": fragments, "role": role, "area_id": int(areaid)},
        )

    def _execute_query_xpd_purchase_records(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """查询和平小店道聚城购买记录。"""
        client = self._client(cookie)
        embed = client.xpd_embed_query(
            act_alias=str(config["xpd_act_alias"]),
            rid=str(config["xpd_rid"]),
        )
        result = client.xpd_purchase_records(
            embed_query=embed["query"],
            act_id=str(config["xpd_act_id"]),
        )
        account.updated_at = datetime.now(UTC)
        total = result.get("total") or len(result.get("records") or [])
        self._mark_task(
            db,
            task,
            "success",
            f"小店兑换记录 {total} 条" if total else "暂无小店兑换记录",
            result,
        )

    def _execute_exchange_xpd_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:
            pay_type = int(payload.get("pay_type") or 1)
        except (TypeError, ValueError):
            self._mark_task(db, task, "failed", "兑换货币参数无效")
            return
        if pay_type not in (1, 5):
            self._mark_task(db, task, "failed", "小店兑换仅支持点券或扭蛋碎片")
            return

        goods = (
            db.query(DouyuXpdGoodsSnapshot)
            .filter(DouyuXpdGoodsSnapshot.commodity_id == commodity_id)
            .first()
        )
        if not goods:
            self._mark_task(db, task, "failed", "未找到小店商品快照，请先刷新商品列表")
            return
        goods_snapshot = goods.raw if isinstance(goods.raw, dict) else {}
        nested_raw = goods_snapshot.get("raw")
        goods_raw = (
            cast(dict[str, Any], nested_raw)
            if isinstance(nested_raw, dict)
            else cast(dict[str, Any], goods_snapshot)
        )
        price_key = "iPrice" if pay_type == 1 else "iJb2Price"
        price = self._to_int(goods_raw.get(price_key))
        if price is None:
            price = goods.price if pay_type == 1 else None
        if price is None or price <= 0:
            currency = "点券" if pay_type == 1 else "扭蛋碎片"
            self._mark_task(db, task, "failed", f"该商品不支持使用{currency}兑换")
            return
        # iGoodsLeft=-1 表示活动未公开库存，不是售罄；只有 0 才阻止兑换。
        if goods.goods_left == 0:
            self._mark_task(db, task, "failed", "该商品库存不足，请刷新商品列表后重试")
            return

        client = self._client(cookie)
        embed = client.xpd_embed_query(
            act_alias=str(config["xpd_act_alias"]),
            rid=str(config["xpd_rid"]),
        )
        role: dict = {}
        try:
            role = client.xpd_get_role(
                embed_query=embed["query"],
                act_id=str(config["xpd_act_id"]),
                rid=str(config["xpd_rid"]),
            )
            if role.get("role_id"):
                self._apply_xpd_role_to_account(
                    account, role, self._xpd_area_id(role, account)
                )
        except DouyuActivityError as exc:
            self._push_log("warning", f"小店兑换前刷新角色失败，使用已保存角色: {exc}")
        if not role.get("role_id") and not account.xpd_role_id:
            self._mark_task(
                db, task, "failed", "未获取到小店绑定角色，请先生成二维码扫码绑定"
            )
            return

        result = client.xpd_exchange_goods(
            embed_query=embed["query"],
            act_id=str(config["xpd_act_id"]),
            rid=str(config["xpd_rid"]),
            commodity_id=commodity_id,
            price=price,
            picture=str(goods_raw.get("sGoodsPic") or ""),
            pay_type=pay_type,
            action_id=str(goods_raw.get("iActionId") or ""),
        )
        if pay_type == 1 and result.get("new_balance") is not None:
            account.xpd_balance = result["new_balance"]
        if pay_type == 5 and result.get("new_balance") is not None:
            account.xpd_fragments = result["new_balance"]
        account.updated_at = datetime.now(UTC)
        db.commit()

        currency = "点券" if pay_type == 1 else "扭蛋碎片"
        display_role = role.get("role_name") or account.xpd_game_name or ""
        channel = (
            "微信" if (role.get("type") == "wx" or account.xpd_area_id == 1) else "手Q"
        )
        result.update(
            {
                "goods": {**goods_raw, "commodityName": goods.name or ""},
                "game_name": display_role,
                "game_channel": channel,
                "account_name": account.nickname
                or account.username
                or account.uid
                or f"#{account.id}",
                "currency": currency,
            }
        )
        # 和平小店兑换 —— 此前从不发事件（不在旧收集器的 allowlist 里）。
        # 消耗的是点券或扭蛋碎片，不是积分；金额用本地已知的 price（快照标价，
        # 已在上面校验过 > 0），作用域标成 peace 以便和精英/电竞区分。
        record_exchange(
            db,
            account,
            task_id=task.id,
            **task_actor(task),
            name=goods.name or commodity_id,
            score=price,
            commodity_id=commodity_id,
            game_name=display_role,
            game_channel=channel,
            points_after=result.get("new_balance"),
            scope="peace",
        )
        self._mark_task(
            db,
            task,
            "success",
            f"兑换小店商品成功: {goods.name or commodity_id}（{price}{currency}）",
            result,
        )