live-hub-py/core/douyu/activity_client.py

"""斗鱼活动、兑换与充值接口客户端。"""

from __future__ import annotations

import json
import re
import time
from collections.abc import Callable
from typing import Any
from urllib.parse import parse_qs, unquote, urlsplit, urlunsplit

import requests

from .cookie_utils import cookie_pairs, cookie_value, normalize_cookie_pairs, normalize_douyu_cookie


PC_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/133.0.0.0 Safari/537.36"
)


class DouyuActivityError(RuntimeError):
    """斗鱼活动接口异常。"""


class DouyuActivityClient:
    """封装斗鱼活动页、兑换、充值和送礼相关接口。"""

    CSRF_API = "https://www.douyu.com/japi/carnival/nc/common/generateCsrf"
    ACF_CCN_API = "https://www.douyu.com/curl/csrfApi/getCsrfCookie?"
    ACF_CCN_FALLBACK_API = "https://www.douyu.com/curl/csrfNlApi/getCsrfCookie"
    SUBSCRIBE_API = "https://www.douyu.com/wgapi/livenc/asubscribe/userSubStatus"
    ROLE_PARAM_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/roleParam"
    BIND_INFO_API = "https://www.douyu.com/japi/carnivalApi/tencent/bindInfo"
    BIND_INFO_V2_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/bindInfo"
    BIND_API = "https://www.douyu.com/japi/carnivalApi/tencent/bind"
    GOODS_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/storehome"
    CREATE_EXCHANGE_ORDER_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/createExchangeOrder"
    PAY_EXCHANGE_ORDER_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/payExchangeOrder"
    EXCHANGE_ORDER_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/getExchangeOrderList"
    EXCHANGE_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchangeList"
    # 2026-08-30 页面新增链路（对齐浏览器）：
    # 预兑 user/queryExchange -> user/confirmExchange；取消预兑 user/queryCancelExchange -> user/cancelExchange；
    # 预约到货 user/sub / user/cancelSub；批量兑换 batchExchangeLimit；单步兑换 exchange；自动兑换开关。
    ELITE_USER_INFO_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/getUserInfo"
    GOODS_DETAIL_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/storedetail"
    EXCHANGE_GIFT_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchange"
    PRE_EXCHANGE_CHECK_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/queryExchange"
    PRE_EXCHANGE_CONFIRM_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/confirmExchange"
    PRE_EXCHANGE_CANCEL_CHECK_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/queryCancelExchange"
    PRE_EXCHANGE_CANCEL_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/cancelExchange"
    COMMODITY_SUBSCRIBE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/sub"
    COMMODITY_UNSUBSCRIBE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/cancelSub"
    BATCH_EXCHANGE_LIMIT_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/batchExchangeLimit"
    AUTO_CONVERT_QUERY_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/activity/automaticallyConvert"
    AUTO_CONVERT_SAVE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/activity/automaticallyConvertSave"
    EXCHANGE_HOME_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/activity/exchangeHome"
    CREDIT_BALANCE_API = "https://www.douyu.com/japi/oms/web/peace/credit/balance"
    CREDIT_DIFF_API = "https://www.douyu.com/japi/oms/web/peace/credit/diff"
    PEACE_ITEM_API = "https://www.douyu.com/japi/oms/web/peace/item"
    PEACE_TIME_API = "https://www.douyu.com/japi/oms/web/peace/time"
    PAY_QR_API = "https://www.douyu.com/japi/oms/web/pay/getQrCode"
    ESPORTS_BIND_INFO_API = "https://www.douyu.com/wgapi/ordnc/live/memberthird/tencent/bindInfo"
    ESPORTS_BIND_ROLE_PARAM_API = "https://www.douyu.com/wgapi/ordnc/live/memberthird/tencent/bindRoleParam"
    ESPORTS_CSRF_API = "https://www.douyu.com/wgapi/livenc/liveweb/csrfApi/getCsrfCookie"
    ESPORTS_AGREEMENT_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/agreement"
    ESPORTS_ACT_BIND_API = "https://www.douyu.com/wgapi/ordnc/live/memberthird/tencent/actBind"
    ESPORTS_USER_INFO_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/getUserInfo"
    ESPORTS_GOODS_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/storehome"
    ESPORTS_EXCHANGE_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/exchange"
    GOLD_QR_API = "https://cz.douyu.com/m/gold/getQrCode"
    GOLD_ACCOUNT_API = "https://cz.douyu.com/item/gold/account"
    EXCHANGE_BALANCE_API = "https://www.douyu.com/wjapi/nc/exchange/fim"
    DONATE_API = "https://www.douyu.com/japi/gift/donate/mainsite/v3"

    def __init__(
        self,
        cookie: str,
        *,
        logger: Callable[[str], None] | None = None,
        timeout: tuple[float, float] = (8, 20),
        proxies: dict[str, str] | None = None,
    ):
        self.cookie = normalize_douyu_cookie(cookie)
        self.timeout = timeout
        self.logger = logger or (lambda _msg: None)
        self.session = requests.Session()
        # trust_env=False: 不读环境变量代理, 只用显式传入的 proxies (代理配置页控制)
        if proxies:
            self.session.proxies.update(proxies)
        self.session.trust_env = False
        self.session.headers.update({
            "User-Agent": PC_UA,
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9",
        })
        self._load_cookie(self.cookie)

    @staticmethod
    def profile_from_cookie(cookie: str) -> dict[str, str]:
        """从斗鱼 Cookie 提取账号基础信息。"""
        nickname = cookie_value(cookie, "acf_nickname")
        return {
            "uid": cookie_value(cookie, "acf_uid"),
            "nickname": unquote(nickname) if nickname else "",
        }

    def _load_cookie(self, cookie: str) -> None:
        for key, value in cookie_pairs(cookie):
            self.session.cookies.set(key, value, domain=".douyu.com", path="/")

    def _merge_response_cookie(self, response: requests.Response) -> None:
        if not response.cookies:
            return
        self.cookie = normalize_cookie_pairs([
            *cookie_pairs(self.cookie),
            *[(key, value) for key, value in response.cookies.items()],
        ])
        self._load_cookie(self.cookie)

    def _set_cookie_value(self, key: str, value: str) -> None:
        """写入单个 Cookie 值，确保手动 Cookie 请求头与 Session 一致。"""
        self.cookie = normalize_cookie_pairs([
            *cookie_pairs(self.cookie),
            (key, value),
        ])
        self._load_cookie(self.cookie)

    def _headers(self, referer: str = "https://www.douyu.com/") -> dict[str, str]:
        return {
            "User-Agent": PC_UA,
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Origin": "https://www.douyu.com",
            "Referer": referer,
            "Cookie": self.cookie,
        }

    @staticmethod
    def _safe_url(url: str) -> str:
        """移除 query/fragment，避免将 Cookie、签名或一次性参数写入日志。"""
        parsed = urlsplit(url)
        return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))

    def _redirect_error(
        self,
        method: str,
        url: str,
        source: str,
        response: requests.Response,
    ) -> DouyuActivityError:
        location = response.headers.get("Location", "")
        safe_location = self._safe_url(location) if location else "-"
        trace = (
            f"method={method.upper()} url={self._safe_url(url)} "
            f"location={safe_location} history={response.status_code}->{safe_location}"
        )
        self.logger(f"{source}: 已拦截重定向 | {trace}")
        return DouyuActivityError(f"{source} 请求被重定向，Cookie 已失效，请重新登录（{trace}）")

    def _request(
        self,
        method: str,
        url: str,
        *,
        source: str,
        headers: dict[str, str] | None = None,
        **kwargs,
    ) -> requests.Response:
        req_headers = self._headers()
        if headers:
            req_headers.update(headers)
        # 活动 JSON 接口不应自动跳到登录页；忽略调用方传入的覆盖值。
        kwargs.pop("allow_redirects", None)
        try:
            response = self.session.request(
                method,
                url,
                headers=req_headers,
                timeout=kwargs.pop("timeout", self.timeout),
                allow_redirects=False,
                **kwargs,
            )
        except requests.RequestException as exc:
            raise DouyuActivityError(
                f"{source} 请求失败: {method.upper()} {self._safe_url(url)}: {exc}"
            ) from exc
        self._merge_response_cookie(response)
        if 300 <= response.status_code < 400:
            raise self._redirect_error(method, url, source, response)
        self.logger(f"{source}: {response.status_code}")
        response.raise_for_status()
        return response

    def _json(self, response: requests.Response, source: str) -> dict[str, Any]:
        text = response.text.strip()
        if not text:
            raise DouyuActivityError(f"{source} 返回为空")
        try:
            return response.json()
        except json.JSONDecodeError as exc:
            raise DouyuActivityError(f"{source} 返回不是 JSON: {text[:200]}") from exc

    def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict[str, Any]:
        return self._json(self._request(method, url, source=source, **kwargs), source)

    def csrf_token(self, *, force_refresh: bool = False) -> str:
        """获取 cvl_csrf_token；优先复用 Cookie 中已有值，与浏览器行为一致。

        2026-08-30 抓包实证：页面只读 Cookie `cvl_csrf_token` 的值作为 csrfToken
        复用（同一 token 跨多次接口、跨十几秒复用），Cookie 里没有/失效时才调用
        generateCsrf，生成的 token 写回 Cookie 继续复用。
        force_refresh=True 用于服务端提示 token 失效后的强制刷新。
        """
        if not force_refresh:
            token = cookie_value(self.cookie, "cvl_csrf_token")
            if token:
                return token
        payload = self._request_json(
            "post",
            self.CSRF_API,
            "生成 CSRF",
            data="{}",
            headers={
                "Content-Type": "application/json;charset=UTF-8",
                "Referer": "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0",
            },
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "生成 CSRF 失败")
        token = cookie_value(self.cookie, "cvl_csrf_token")
        if not token:
            raise DouyuActivityError("响应 Cookie 中没有 cvl_csrf_token")
        return token

    @staticmethod
    def _is_csrf_error(payload: dict[str, Any]) -> bool:
        """判断响应是否为 csrfToken 校验失败（此时应刷新并重试一次）。"""
        return "csrf" in str(payload.get("msg") or "").lower()

    @staticmethod
    def _peace_page_referer(rid: str = "") -> str:
        """精英手册首页 Referer（对齐浏览器 getUserInfo 等接口）。"""
        if rid:
            return f"https://www.douyu.com/pages/live-peace-handbook/web?ditchname=pass0&roomId={rid}"
        return "https://www.douyu.com/pages/live-peace-handbook/web?ditchname=pass0"

    @staticmethod
    def _peace_shop_referer(rid: str = "") -> str:
        """精英手册商城页 Referer（对齐浏览器 create/pay/storedetail 等接口）。"""
        if rid:
            return f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}"
        return "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0"

    def _post_csrf_form(
        self,
        url: str,
        source: str,
        fields: dict[str, Any],
        referer: str,
    ) -> dict[str, Any]:
        """提交带 csrfToken 的 multipart 表单（对齐浏览器请求层）。

        浏览器 csrf 配置 {tvk: cvl_csrf_token, tn: csrfToken} 会把 Cookie 中的
        token 值拼进表单；这里同样复用 Cookie 值，仅在服务端提示 token 失效时
        刷新一次后重试。
        """
        token = self.csrf_token()
        payload = self._request_json(
            "post",
            url,
            source,
            files=self._multipart_fields({**fields, "csrfToken": token}),
            headers={"Referer": referer},
        )
        if payload.get("error") not in (0, "0") and self._is_csrf_error(payload):
            self.logger(f"{source}: csrfToken 校验失败，刷新后重试一次")
            token = self.csrf_token(force_refresh=True)
            payload = self._request_json(
                "post",
                url,
                source,
                files=self._multipart_fields({**fields, "csrfToken": token}),
                headers={"Referer": referer},
            )
        return payload

    def acf_ccn(self, *, refresh_subscribe: bool = True) -> str:
        """获取 acf_ccn；必要时通过订阅接口刷新一次。"""
        response = self._request(
            "get",
            self.ACF_CCN_API,
            source="获取 acf_ccn",
            headers={
                "Origin": "",
                "Referer": "https://www.douyu.com/pages/ord-task-center?clientType=web&panelSource=1&rid=0",
            },
        )
        ctn = cookie_value(self.cookie, "acf_ccn")
        if not ctn:
            response = self._request(
                "get",
                self.ACF_CCN_FALLBACK_API,
                source="获取 acf_ccn 备用接口",
                headers={"Origin": ""},
            )
            ctn = cookie_value(self.cookie, "acf_ccn")
        if not ctn:
            raise DouyuActivityError(f"获取 acf_ccn 失败: {response.text[:200]}")

        if refresh_subscribe:
            payload = self._request_json(
                "post",
                self.SUBSCRIBE_API,
                "刷新订阅状态",
                data={"subscribe": "cmn", "ctn": ctn},
                headers={
                    "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
                    "Referer": "https://www.douyu.com/",
                },
            )
            if payload.get("error") not in (0, "0"):
                raise DouyuActivityError(payload.get("msg") or "刷新订阅状态失败")
            ctn = cookie_value(self.cookie, "acf_ccn") or ctn
        return ctn

    def get_bind_qr(self, act_alias: str) -> dict[str, Any]:
        """生成腾讯游戏绑定二维码链接（v2 接口）。"""
        ctn = self.acf_ccn(refresh_subscribe=True)
        payload = self._request_json(
            "post",
            self.ROLE_PARAM_API,
            "获取绑定二维码",
            params={"actAlias": act_alias},
            headers={
                "Referer": "https://www.douyu.com/",
            },
        )
        url = (payload.get("data") or {}).get("url") or ""
        if not url:
            raise DouyuActivityError(payload.get("msg") or "获取绑定二维码失败")
        return {"url": url, "ctn": ctn, "raw": payload}

    @staticmethod
    def _first_present(data: dict[str, Any], *keys: str):
        """按候选 key 取第一个非 None 值。"""
        for key in keys:
            if key in data and data.get(key) is not None:
                return data.get(key)
        return None

    def bind_info(self, act_alias: str, *, v2: bool = False) -> dict[str, Any]:
        """查询绑定游戏账号信息。"""
        url = self.BIND_INFO_V2_API if v2 else self.BIND_INFO_API
        payload = self._request_json(
            "get",
            url,
            "查询绑定信息",
            params={"actAlias": act_alias},
            headers={"Origin": "", "Referer": "https://www.douyu.com/"},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "查询绑定信息失败")
        parsed = self._parse_bind_info(payload, act_alias=act_alias, api_version="v2" if v2 else "v1")
        self._log_bind_info(parsed)
        return parsed

    def _log_bind_info(self, parsed: dict[str, Any]) -> None:
        """输出绑定信息接口返回的角色明细（用于排查两个接口/多个 alias 的角色差异）。"""
        self.logger(
            f"查询绑定信息明细 api={parsed.get('api_version') or '-'} "
            f"act={parsed.get('act_alias') or '-'} "
            f"role={parsed.get('role_name') or '-'} "
            f"area={parsed.get('area_name') or '-'} plat={parsed.get('plat_name') or '-'} "
            f"nick={parsed.get('nickname') or '-'} "
            f"bound_account={1 if parsed.get('is_bound_account') else 0} "
            f"bound_role={1 if parsed.get('is_bound_role') else 0} "
            f"bound_act={1 if parsed.get('is_bound_act') else 0} "
            f"need_act={1 if parsed.get('need_bind_act') else 0} "
            f"need_role={1 if parsed.get('need_bind_role') else 0} "
            f"wait={parsed.get('change_role_wait_time') if parsed.get('change_role_wait_time') is not None else '-'} "
            f"can_change_role={parsed.get('can_change_role')} "
            f"can_change_time={parsed.get('can_change_time') if parsed.get('can_change_time') is not None else '-'}"
        )

    def _parse_bind_info(self, payload: dict[str, Any], *, act_alias: str, api_version: str) -> dict[str, Any]:
        """统一解析不同活动的腾讯游戏绑定信息。"""
        data = payload.get("data") or {}
        game_account = data.get("gameAccount") or {}
        game_role = data.get("gameRole") or {}
        tx_act = data.get("txAct") or {}
        bind_account = data.get("bindAccount")
        bind_role = data.get("bindRole")
        bind_act = data.get("bindAct")
        # 换绑倒计时在不同 alias/版本字段可能不一致，尽量兼容。
        change_role_wait_time = self._first_present(
            data,
            "changeRoleWaitTime",
            "change_role_wait_time",
            "changeBindWaitTime",
            "changBindTime",
        )
        if change_role_wait_time is None and isinstance(tx_act, dict):
            change_role_wait_time = self._first_present(
                tx_act,
                "changeRoleWaitTime",
                "changeBindWaitTime",
                "changBindTime",
            )
        can_change_role = self._first_present(
            data,
            "canChangeRole",
            "can_change_role",
            "canChangeBind",
        )
        if can_change_role is None and isinstance(tx_act, dict):
            can_change_role = self._first_present(tx_act, "canChangeRole", "canChangeBind")
        can_change_time = self._first_present(
            data,
            "canChangeTime",
            "can_change_time",
        )
        if can_change_time is None and isinstance(tx_act, dict):
            can_change_time = self._first_present(tx_act, "canChangeTime", "can_change_time")
        return {
            "act_alias": act_alias,
            "api_version": api_version,
            "bind_account": bind_account,
            "bind_role": bind_role,
            "bind_act": bind_act,
            "need_bind_role": data.get("needBindRole"),
            "need_bind_act": data.get("needBindAct"),
            "is_bound_account": str(bind_account) == "1",
            "is_bound_role": str(bind_role) == "1",
            "is_bound_act": str(bind_act) == "1",
            "nickname": game_account.get("nick") or "",
            "role_name": game_role.get("roleName") or "",
            "area_name": game_role.get("areaName") or "",
            "plat_name": game_role.get("platName") or "",
            "tx_act": tx_act if isinstance(tx_act, dict) else {},
            "change_role_wait_time": change_role_wait_time,
            "can_change_role": can_change_role,
            "can_change_time": can_change_time,
            "raw": payload,
        }

    def esports_bind_info(self, act_alias: str) -> dict[str, Any]:
        """电竞手册唯一的角色状态查询，返回角色、绑定状态和换绑冷却。"""
        payload = self._request_json(
            "get",
            self.ESPORTS_BIND_INFO_API,
            "查询电竞手册绑定信息",
            params={"actAlias": act_alias},
            headers={"Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "查询电竞手册绑定信息失败")
        parsed = self._parse_bind_info(payload, act_alias=act_alias, api_version="esports")
        self._log_bind_info(parsed)
        return parsed

    def get_esports_bind_qr(self, act_alias: str) -> dict[str, Any]:
        """生成电竞手册的腾讯角色绑定二维码链接。"""
        payload = self._request_json(
            "get",
            self.ESPORTS_BIND_ROLE_PARAM_API,
            "获取电竞手册绑定二维码",
            params={"actAlias": act_alias},
            headers={"Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "获取电竞手册绑定二维码失败")
        data = payload.get("data") or {}
        # 电竞手册网页实际打开 data.url；gameRoleBindUrl 是兼容旧入口。
        bind_url = data.get("url") or data.get("gameRoleBindUrl") or ""
        if not bind_url:
            raise DouyuActivityError(payload.get("msg") or "电竞手册绑定二维码链接为空")
        return {
            "url": bind_url,
            "bind_url": bind_url,
            "role_param_url": data.get("url") or "",
            "game_role_bind_url": data.get("gameRoleBindUrl") or "",
            "act_alias": act_alias,
            "raw": payload,
        }

    def esports_ctn(self, *, referer: str) -> str:
        """获取电竞手册页面当前 ctn，仅在 Cookie 缺失时走页面 CSRF 接口。"""
        ctn = cookie_value(self.cookie, "acf_ccn")
        if ctn:
            return ctn

        response = self._request(
            "get",
            self.ESPORTS_CSRF_API,
            source="获取电竞手册 ctn",
            headers={"Origin": "", "Referer": referer},
        )
        ctn = cookie_value(self.cookie, "acf_ccn")
        if not ctn:
            raise DouyuActivityError(f"获取电竞手册 ctn 失败: {response.text[:200]}")
        return ctn

    def confirm_esports_bind(self, act_alias: str, *, room_id: str = "9263298") -> dict[str, Any]:
        """自动完成协议前置请求后，通过 actBind 确认电竞手册绑定。"""
        referer = (
            "https://www.douyu.com/pages/live-peace-pel/web"
            f"?roomId={room_id or '9263298'}&isAnchorSide=0"
        )

        # 页面成功流程固定为 agreement -> actBind；协议不作为独立业务状态暴露。
        # 两次 POST 都使用当前 Cookie 中的 acf_ccn，不额外主动刷新 CSRF。
        agreement_ctn = self.esports_ctn(referer=referer)
        agreement = self._request_json(
            "post",
            self.ESPORTS_AGREEMENT_API,
            "同意电竞手册协议",
            files={"ctn": (None, agreement_ctn)},
            headers={"Referer": referer},
        )
        if agreement.get("error") not in (0, "0", None):
            raise DouyuActivityError(agreement.get("msg") or "同意电竞手册协议失败")

        bind_ctn = self.esports_ctn(referer=referer)
        bind = self._request_json(
            "post",
            self.ESPORTS_ACT_BIND_API,
            "完成电竞手册绑定",
            files={
                "actAlias": (None, act_alias),
                "ctn": (None, bind_ctn),
            },
            headers={"Referer": referer},
        )
        if bind.get("error") not in (0, "0", None):
            raise DouyuActivityError(bind.get("msg") or "完成电竞手册绑定失败")
        return {"agreement": agreement, "act_bind": bind}

    def confirm_bind(self, act_alias: str) -> dict[str, Any]:
        """确认绑定当前扫码选择的游戏角色。"""
        token = self.csrf_token()
        payload = self._request_json(
            "post",
            self.BIND_API,
            "确认绑定",
            data={"actAlias": act_alias, "csrfToken": token},
            headers={
                "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
                "Referer": "https://www.douyu.com/",
            },
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "确认绑定失败")
        return {"csrf_token": token, "raw": payload}

    def create_elite_qr(self, *, ctn: str, act_alias: str, amount: int, room_id: str) -> dict[str, Any]:
        """生成精英宝典支付二维码。"""
        # E 语言实现会在 peace/item 和 getQrCode 两次请求里显式拼上 acf_ccn=ctn。
        # 这里也同步 Cookie，避免接口只看 Cookie 时误判未绑定游戏账号。
        self._set_cookie_value("acf_ccn", ctn)
        item_payload = self._request_json(
            "post",
            self.PEACE_ITEM_API,
            "校验精英宝典",
            json={"itemId": 1, "ctn": ctn},
            headers={"Content-Type": "application/json;charset=UTF-8"},
        )
        if item_payload.get("error") not in (0, "0"):
            raise DouyuActivityError(item_payload.get("msg") or "精英宝典校验失败")

        payload = self._request_json(
            "post",
            self.PAY_QR_API,
            "生成精英宝典支付码",
            json={
                "bizType": "peace",
                "biz": {"actAlias": act_alias, "amount": amount, "roomId": int(room_id or 0)},
                "ctn": ctn,
            },
            headers={"Content-Type": "application/json;charset=UTF-8"},
        )
        qr = (payload.get("data") or {}).get("qr") or ""
        if not qr:
            raise DouyuActivityError(payload.get("msg") or "生成精英宝典支付码失败")
        return {"pay_url": qr, "ctn": ctn, "act_alias": act_alias, "item": item_payload, "raw": payload}

    def create_esports_qr(self, *, ctn: str, act_alias: str, amount: int, room_id: str) -> dict[str, Any]:
        """生成电竞手册支付二维码。"""
        self._set_cookie_value("acf_ccn", ctn)
        time_payload = self._request_json(
            "get",
            self.PEACE_TIME_API,
            "校验电竞手册开通时间",
            headers={"Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
        )
        if time_payload.get("error") not in (0, "0"):
            raise DouyuActivityError(time_payload.get("msg") or "电竞手册当前不可开通")

        payload = self._request_json(
            "post",
            self.PAY_QR_API,
            "生成电竞手册支付码",
            json={
                "bizType": "peacePel",
                "biz": {"actAlias": act_alias, "amount": amount, "roomId": int(room_id or 0)},
                "ctn": ctn,
            },
            headers={
                "Content-Type": "application/json;charset=UTF-8",
                "Referer": "https://www.douyu.com/pages/live-peace-pel/web/",
            },
        )
        data = payload.get("data") or {}
        qr = data.get("qr") or ""
        if not qr:
            raise DouyuActivityError(payload.get("msg") or "生成电竞手册支付码失败")
        return {
            "pay_url": qr,
            "pay_token": data.get("token") or "",
            "pay_ttl": data.get("ttl"),
            "ctn": ctn,
            "act_alias": act_alias,
            "time": time_payload,
            "raw": payload,
        }

    def esports_user_info(self, *, manual_id: str) -> dict[str, Any]:
        """查询电竞手册的开通状态和积分。"""
        payload = self._request_json(
            "get",
            self.ESPORTS_USER_INFO_API,
            "查询电竞手册状态",
            params={"manualID": manual_id},
            headers={"Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "查询电竞手册状态失败")
        user_info = (payload.get("data") or {}).get("userInfo") or {}
        return {
            "manual_id": manual_id,
            "manual_type": user_info.get("manualType"),
            "manual_score": user_info.get("manualScore"),
            "expire_time": user_info.get("expireTime"),
            "user_info": user_info,
            "raw": payload,
        }

    @staticmethod
    def _esports_shop_referer(rid: str) -> str:
        return (
            "https://www.douyu.com/pages/live-peace-pel/web/shop"
            f"?roomId={rid or '0'}&isAnchorSide=0"
        )

    def list_esports_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
        """查询电竞手册皮肤商城。"""
        payload = self._request_json(
            "get",
            self.ESPORTS_GOODS_API,
            "刷新电竞皮肤商城",
            params={"manualID": manual_id, "rid": rid},
            headers={
                "Origin": "",
                "Referer": self._esports_shop_referer(rid),
            },
        )
        goods = (payload.get("data") or {}).get("list") or []
        return {
            "goods": goods,
            "score": (payload.get("data") or {}).get("score"),
            "raw": payload,
        }

    def exchange_esports_goods(
        self,
        *,
        manual_id: str,
        rid: str,
        commodity_id: str,
        quantity: int = 1,
        ctn: str | None = None,
    ) -> dict[str, Any]:
        """兑换电竞手册皮肤商城商品。"""
        referer = self._esports_shop_referer(rid)
        ctn_value = ctn or self.esports_ctn(referer=referer)
        self._set_cookie_value("acf_ccn", ctn_value)
        payload = self._request_json(
            "post",
            self.ESPORTS_EXCHANGE_API,
            "兑换电竞皮肤",
            data={
                "manualID": manual_id,
                "commodityID": commodity_id,
                "rid": rid or "0",
                "num": str(max(1, quantity)),
                "ctn": ctn_value,
            },
            headers={
                "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
                "Origin": "",
                "Referer": referer,
            },
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "兑换电竞皮肤失败")
        return {
            "commodity_id": commodity_id,
            "quantity": max(1, quantity),
            "ctn": ctn_value,
            "raw": payload,
        }

    def create_gold_qr(self, *, amount: int, pay_type: int = 1, product_id: str = "DYTV_Product_Gold_1") -> dict[str, Any]:
        """生成鱼翅充值支付二维码。"""
        payload = self._request_json(
            "post",
            self.GOLD_QR_API,
            "生成鱼翅充值码",
            data={
                "pay_type": str(pay_type),
                "product_id": product_id,
                "number": str(amount),
                "toname": "",
                "source": "dy_box",
            },
            headers={
                "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
                "Origin": "https://cz.douyu.com",
                "Referer": "https://cz.douyu.com/",
            },
        )
        data = payload.get("data") or {}
        pay_url = data.get("code_url") or payload.get("code_url") or ""
        token = data.get("token") or payload.get("token") or ""
        if not pay_url:
            raise DouyuActivityError(payload.get("msg") or "生成鱼翅充值码失败")
        return {"pay_url": pay_url, "token": token, "raw": payload}

    def _donate_gift(
        self,
        *,
        gift_count: int,
        room_id: str,
        gift_id: str,
        skin_id: str,
        source: str,
    ) -> dict[str, Any]:
        """赠送直播间礼物。"""
        payload = self._request_json(
            "post",
            self.DONATE_API,
            source,
            data={
                "giftId": gift_id,
                "giftCount": str(gift_count),
                "roomId": room_id,
                "bizExt": json.dumps({"isMainGift": 1}, separators=(",", ":")),
                "skinId": skin_id,
            },
            headers={
                "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
                "Referer": f"https://www.douyu.com/{room_id}",
            },
        )
        if payload.get("error") not in (0, "0") and payload.get("msg") != "success":
            raise DouyuActivityError(payload.get("msg") or f"{source}失败")
        return {"raw": payload}

    def donate_elite_gift(
        self,
        *,
        gift_count: int,
        room_id: str,
        gift_id: str,
        skin_id: str,
    ) -> dict[str, Any]:
        """赠送精英令礼物。"""
        return self._donate_gift(
            gift_count=gift_count,
            room_id=room_id,
            gift_id=gift_id,
            skin_id=skin_id,
            source="赠送精英令",
        )

    def donate_esports_gift(
        self,
        *,
        gift_name: str,
        gift_count: int,
        room_id: str,
        gift_id: str,
        skin_id: str,
    ) -> dict[str, Any]:
        """赠送电竞手册任务指定礼物。"""
        return self._donate_gift(
            gift_count=gift_count,
            room_id=room_id,
            gift_id=gift_id,
            skin_id=skin_id,
            source=f"赠送{gift_name}",
        )

    def query_points(self, *, uid: str, ctn: str) -> dict[str, Any]:
        """查询手册积分余额。"""
        payload = self._request_json(
            "post",
            self.CREDIT_BALANCE_API,
            "查询手册积分",
            json={"uid": str(uid), "ctn": ctn},
            headers={"Content-Type": "application/json;charset=UTF-8"},
        )
        data = payload.get("data") or {}
        return {"points": data.get("balance"), "raw": payload}

    def query_points_diff(self, *, uid: str, ctn: str) -> dict[str, Any]:
        """查询手册积分差额。"""
        payload = self._request_json(
            "post",
            self.CREDIT_DIFF_API,
            "查询手册积分差额",
            json={"userId": str(uid), "ctn": ctn},
            headers={"Content-Type": "application/json;charset=UTF-8"},
        )
        data = payload.get("data") or {}
        return {"diff": data.get("diff"), "raw": payload}

    def list_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
        """刷新活动商品列表。"""
        payload = self._request_json(
            "get",
            self.GOODS_API,
            "刷新商品列表",
            params={"manualID": manual_id, "rid": rid},
            headers={
                "Origin": "",
                "Referer": f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}",
            },
        )
        goods = (payload.get("data") or {}).get("list") or []
        return {"goods": goods, "raw": payload}

    def query_limited_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
        """查询当前账号限兑商品。"""
        result = self.list_goods(manual_id=manual_id, rid=rid)
        limited = [
            item for item in result["goods"]
            if str(item.get("status") or "") == "1"
        ]
        return {"limited_goods": limited, "raw": result["raw"]}

    def exchange_records(self, *, manual_id: str, limit: int = 100, offset: int = 0) -> dict[str, Any]:
        """查询兑换记录。"""
        payload = self._request_json(
            "get",
            self.EXCHANGE_LIST_API,
            "查询兑换记录",
            params={"manualID": manual_id, "limit": limit, "offset": offset},
            headers={"Origin": "", "Referer": "https://www.douyu.com/"},
        )
        records = (payload.get("data") or {}).get("list") or []
        return {"records": records, "raw": payload}

    @staticmethod
    def _multipart_fields(data: dict[str, Any]) -> dict[str, tuple[None, str]]:
        """Build requests multipart fields without uploading files."""
        return {key: (None, str(value)) for key, value in data.items()}

    def create_exchange_order(
        self,
        *,
        manual_id: str,
        rid: str,
        commodity_id: str,
        num: int = 1,
    ) -> dict[str, Any]:
        """Lock stock for an activity commodity before spending points."""
        if num < 1:
            raise DouyuActivityError("锁单数量必须大于 0")
        payload = self._post_csrf_form(
            self.CREATE_EXCHANGE_ORDER_API,
            "锁定兑换商品",
            {
                "manualID": manual_id,
                "rid": rid,
                "commodityID": commodity_id,
                "num": num,
            },
            referer=self._peace_shop_referer(rid),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "锁定兑换商品失败")
        data = payload.get("data") or {}
        order_id = str(data.get("orderId") or "").strip()
        if not order_id:
            raise DouyuActivityError("锁定兑换商品成功，但响应中没有 orderId")
        locked_at = int(time.time())
        try:
            expire_seconds = int(data.get("expireSeconds") or 0)
        except (TypeError, ValueError):
            expire_seconds = 0
        if expire_seconds > 0:
            expires_at = locked_at + expire_seconds
        else:
            expires_at = None
        return {
            "order_id": order_id,
            "commodity_id": commodity_id,
            "commodity_name": data.get("commodityName") or "",
            "commodity_image": data.get("commodityImage") or "",
            "score": data.get("score"),
            "num": data.get("num", num),
            "expire_seconds": data.get("expireSeconds"),
            "locked_at": locked_at,
            "expires_at": expires_at,
            "raw": payload,
        }

    def pay_exchange_order(self, *, manual_id: str, order_id: str, rid: str = "") -> dict[str, Any]:
        """Pay a previously locked exchange order with handbook points."""
        normalized_order_id = str(order_id).strip()
        if not normalized_order_id:
            raise DouyuActivityError("锁单订单号不能为空")
        payload = self._post_csrf_form(
            self.PAY_EXCHANGE_ORDER_API,
            "支付锁单",
            {
                "orderID": normalized_order_id,
                "manualID": manual_id,
            },
            referer=self._peace_shop_referer(rid),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "支付锁单失败")
        data = payload.get("data") or {}
        return {
            "order_id": str(data.get("orderId") or normalized_order_id),
            "exchange_id": str(data.get("exchangeId") or ""),
            "commodity_type": data.get("commodityType"),
            "commodity_image": data.get("commodityImage") or "",
            "exchange_num": data.get("exchangeNum"),
            "s_type": data.get("sType"),
            "g_type": data.get("gType"),
            "raw": payload,
        }

    def exchange_orders(self, *, manual_id: str, page: int = 1, page_size: int = 10) -> dict[str, Any]:
        """Query current locked exchange orders."""
        payload = self._request_json(
            "get",
            self.EXCHANGE_ORDER_LIST_API,
            "查询锁单",
            params={"page": page, "pageSize": page_size, "manualID": manual_id},
            headers={
                "Origin": "",
                "Referer": "https://www.douyu.com/pages/live-peace-handbook/web/lock-order?ditchname=pass0",
            },
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询锁单失败")
        data = payload.get("data") or {}
        return {
            "orders": data.get("list") or [],
            "total": data.get("total") or 0,
            "raw": payload,
        }

    def exchange_goods(self, *, manual_id: str, rid: str, commodity_id: str) -> dict[str, Any]:
        """Lock an activity commodity and immediately pay the locked order."""
        locked = self.create_exchange_order(
            manual_id=manual_id,
            rid=rid,
            commodity_id=commodity_id,
            num=1,
        )
        paid = self.pay_exchange_order(manual_id=manual_id, order_id=locked["order_id"])
        return {
            "commodity_id": commodity_id,
            "order_id": locked["order_id"],
            "exchange_id": paid["exchange_id"],
            "commodity_image": paid["commodity_image"] or locked["commodity_image"],
            "exchange_num": paid["exchange_num"],
            "lock_order": locked,
            "payment": paid,
        }

    def elite_user_info(self, *, manual_id: str, rid: str = "") -> dict[str, Any]:
        """查询精英手册状态（对齐页面 getUserInfo，含 manualType/manualScore）。

        浏览器每次进入手册页都会先请求该接口，用于判断商品是否"精英专享"、
        是否已达兑换上限等状态；兑换前先查一次与页面行为一致。
        """
        payload = self._request_json(
            "get",
            self.ELITE_USER_INFO_API,
            "查询手册状态",
            params={"manualID": manual_id},
            headers={"Origin": "", "Referer": self._peace_page_referer(rid)},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "查询手册状态失败")
        user_info = (payload.get("data") or {}).get("userInfo") or {}
        return {
            "manual_id": manual_id,
            "manual_type": user_info.get("manualType"),
            "manual_score": user_info.get("manualScore"),
            "expire_time": user_info.get("expireTime"),
            "user_info": user_info,
            "raw": payload,
        }

    def query_goods_detail(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]:
        """查询商品详情 storedetail（对齐浏览器 getGoodsDetail）。

        详情里带有兑换状态机所需的全部标记：preExchange.isExists / subscribeType /
        exchangeInfo.userStatus / batchExchange / roomLimited / storeStatus 等。
        """
        payload = self._request_json(
            "get",
            self.GOODS_DETAIL_API,
            "查询商品详情",
            params={"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"},
            headers={"Origin": "", "Referer": self._peace_shop_referer(rid)},
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询商品详情失败")
        return {"detail": payload.get("data") or {}, "raw": payload}

    @staticmethod
    def resolve_exchange_plan(
        detail: dict[str, Any],
        *,
        manual_type: Any = None,
        batch_num_limit: Any = None,
    ) -> dict[str, Any]:
        """对齐浏览器兑换按钮状态机（$e / Ve / Ge）给出自动化兑换计划。

        浏览器判定顺序（2026-08-30 bundle index.4872272b.js / index.0539fbf8.js）：
          存在预扣订单 -> 已达兑换上限 -> 无库存 -> 限定房间 -> 活动未开启(预约/预兑) ->
          精英专享未开通 -> 已预兑等待开放 -> 立即兑换(经典锁单/支付)。
        action:
          classic      经典链路 createExchangeOrder -> payExchangeOrder（批量商品 num 受上限约束）
          pre_exchange 预兑商品（subscribeType=2 且未开启）：user/queryExchange -> user/confirmExchange
          subscribe    预约到货商品（subscribeType=1 且未开启）：user/sub
          wait         已预兑（userStatus=1）但预扣订单尚未生成，等待开放后重试
          blocked      明确不可兑换，text 为原因（文案对齐浏览器/服务端）
        """
        detail = detail or {}
        pre = detail.get("preExchange") or {}
        pre_is_exists = int(pre.get("isExists") or 0)
        exchange_info = detail.get("exchangeInfo") or {}
        user_status = int(exchange_info.get("userStatus") or 0)
        subscribe_type = int(detail.get("subscribeType") or 0)
        status = int(detail.get("status") or 0)
        store_status = int(detail.get("storeStatus") or 0)
        open_status = int(detail.get("openStatus") or 0)
        unlimited = (
            int(detail.get("unlimitedMode") or 0) == 1
            and int(detail.get("unlimitedStatus") or 0) == 1
        )
        room_limited = int(detail.get("roomLimited") or 0) == 1
        is_limit_room = int(detail.get("isLimitRoom") or 0) == 1
        elite_limited = int(detail.get("eliteLimited") or 0) == 1
        batch_exchange = int(detail.get("batchExchange") or 0) > 0
        max_num = 1
        if batch_exchange and batch_num_limit is not None:
            try:
                max_num = max(1, int(batch_num_limit or 1))
            except (TypeError, ValueError):
                max_num = 1

        plan: dict[str, Any] = {
            "action": "classic",
            "text": "",
            "max_num": max_num,
            "batch_exchange": batch_exchange,
            "pre_exchange": bool(pre_is_exists),
            "subscribe_type": subscribe_type,
            "user_status": user_status,
        }
        if pre_is_exists:
            plan.update(action="classic", text="存在预扣订单，继续兑换")
        elif status == 1 or (subscribe_type == 2 and user_status == 2):
            plan.update(action="blocked", text="已达兑换上限")
        elif store_status == 0 and not unlimited:
            plan.update(action="blocked", text="无库存")
        elif room_limited and not is_limit_room:
            plan.update(action="blocked", text="限定房间兑换，请前往指定直播间")
        elif open_status == 0:
            if subscribe_type == 1:
                plan.update(action="subscribe", text="活动未开启，预约到货")
            elif subscribe_type == 2:
                plan.update(action="pre_exchange", text="活动未开启，预兑")
            else:
                plan.update(action="blocked", text="活动未开启")
        elif elite_limited and manual_type is not None and int(manual_type or 0) != 1:
            plan.update(action="blocked", text="需要开通精英手册，才可以兑换本道具")
        elif subscribe_type == 2 and user_status == 1:
            plan.update(action="wait", text="已预兑，等待开放后继续兑换")
        else:
            plan.update(action="classic", text="立即兑换")
        return plan

    def batch_exchange_limit(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]:
        """查询批量兑换上限（对齐页面 batchExchangeLimit，仅 batchExchange=1 商品）。"""
        payload = self._post_csrf_form(
            self.BATCH_EXCHANGE_LIMIT_API,
            "查询批量兑换上限",
            {"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"},
            referer=self._peace_shop_referer(rid),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询批量兑换上限失败")
        data = payload.get("data") or {}
        return {"limit": data.get("limit"), "raw": payload}

    def pre_exchange_check(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]:
        """查询预兑状态（对齐页面 user/queryExchange，预兑弹窗前必查）。"""
        payload = self._request_json(
            "get",
            self.PRE_EXCHANGE_CHECK_API,
            "查询预兑状态",
            params={"manualID": manual_id, "commodityID": commodity_id},
            headers={"Origin": "", "Referer": self._peace_shop_referer()},
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询预兑状态失败")
        return {"data": payload.get("data") or {}, "raw": payload}

    def pre_exchange_cancel_check(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]:
        """查询取消预兑信息（对齐页面 user/queryCancelExchange）。"""
        payload = self._request_json(
            "get",
            self.PRE_EXCHANGE_CANCEL_CHECK_API,
            "查询取消预兑信息",
            params={"manualID": manual_id, "commodityID": commodity_id},
            headers={"Origin": "", "Referer": self._peace_shop_referer()},
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询取消预兑信息失败")
        return {"data": payload.get("data") or {}, "raw": payload}

    def confirm_pre_exchange(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]:
        """确认预兑（对齐页面 user/confirmExchange，成功提示"预兑成功"）。"""
        payload = self._post_csrf_form(
            self.PRE_EXCHANGE_CONFIRM_API,
            "确认预兑",
            {"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"},
            referer=self._peace_shop_referer(rid),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "确认预兑失败")
        return {"raw": payload}

    def cancel_pre_exchange(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]:
        """取消预兑（对齐页面 user/cancelExchange）。"""
        payload = self._post_csrf_form(
            self.PRE_EXCHANGE_CANCEL_API,
            "取消预兑",
            {"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"},
            referer=self._peace_shop_referer(rid),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "取消预兑失败")
        return {"raw": payload}

    def subscribe_commodity(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]:
        """预约到货（对齐页面 user/sub，成功提示"预约成功"）。"""
        payload = self._post_csrf_form(
            self.COMMODITY_SUBSCRIBE_API,
            "预约到货",
            {"manualID": manual_id, "commodityID": commodity_id},
            referer=self._peace_shop_referer(),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "预约到货失败")
        return {"raw": payload}

    def unsubscribe_commodity(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]:
        """取消预约到货（对齐页面 user/cancelSub）。"""
        payload = self._post_csrf_form(
            self.COMMODITY_UNSUBSCRIBE_API,
            "取消预约到货",
            {"manualID": manual_id, "commodityID": commodity_id},
            referer=self._peace_shop_referer(),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "取消预约到货失败")
        return {"raw": payload}

    def exchange_gift(self, *, manual_id: str, commodity_id: str, rid: str = "", num: int = 1) -> dict[str, Any]:
        """单步兑换（对齐页面 postExchangeGift -> /ordnc/activity/peace/exchange）。

        2026-08-30 bundle 中新链路，页面主要用于兑换中心/自动兑换场景；字段按页面
        调用（{manualID, commodityID, num, rid} + csrfToken）对齐。
        """
        payload = self._post_csrf_form(
            self.EXCHANGE_GIFT_API,
            "单步兑换",
            {
                "manualID": manual_id,
                "commodityID": commodity_id,
                "num": max(1, int(num)),
                "rid": rid or "0",
            },
            referer=self._peace_shop_referer(rid),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "兑换失败")
        data = payload.get("data") or {}
        return {
            "exchange_id": str(data.get("exchangeId") or ""),
            "commodity_type": data.get("commodityType"),
            "commodity_image": data.get("commodityImage") or "",
            "exchange_num": data.get("exchangeNum"),
            "s_type": data.get("sType"),
            "g_type": data.get("gType"),
            "raw": payload,
        }

    def auto_convert_switch(self, *, manual_id: str) -> dict[str, Any]:
        """查询自动兑换开关（对齐页面 activity/automaticallyConvert）。"""
        payload = self._request_json(
            "get",
            self.AUTO_CONVERT_QUERY_API,
            "查询自动兑换开关",
            params={"manualID": manual_id},
            headers={"Origin": "", "Referer": self._peace_shop_referer()},
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询自动兑换开关失败")
        return {"data": payload.get("data") or {}, "raw": payload}

    def set_auto_convert(self, *, manual_id: str, switch: Any) -> dict[str, Any]:
        """设置自动兑换开关（对齐页面 automaticallyConvertSave {manualID, switch}）。"""
        payload = self._post_csrf_form(
            self.AUTO_CONVERT_SAVE_API,
            "设置自动兑换开关",
            {"manualID": manual_id, "switch": switch},
            referer=self._peace_shop_referer(),
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "设置自动兑换开关失败")
        return {"raw": payload}

    def exchange_home_info(self) -> dict[str, Any]:
        """查询兑换中心信息（对齐页面 activity/exchangeHome）。"""
        payload = self._request_json(
            "get",
            self.EXCHANGE_HOME_API,
            "查询兑换中心",
            headers={"Origin": "", "Referer": self._peace_shop_referer()},
        )
        if payload.get("error") not in (0, "0"):
            raise DouyuActivityError(payload.get("msg") or "查询兑换中心失败")
        return {"data": payload.get("data") or {}, "raw": payload}

    def gold_account(self) -> dict[str, Any]:
        """查询 cz 侧鱼翅余额。"""
        payload = self._request_json(
            "get",
            self.GOLD_ACCOUNT_API,
            "查询鱼翅余额",
            headers={"Origin": "https://cz.douyu.com", "Referer": "https://cz.douyu.com/"},
        )
        data = payload.get("data") or {}
        return {"gold": data.get("gold"), "raw": payload}

    def exchange_balance(self) -> dict[str, Any]:
        """查询钱包兑换中心余额。"""
        payload = self._request_json(
            "get",
            self.EXCHANGE_BALANCE_API,
            "查询钱包兑换余额",
            params={"appCode": "YJTX"},
            headers={
                "Origin": "",
                "Referer": "https://www.douyu.com/member/walletcenter",
            },
        )
        data = payload.get("data") or {}
        return {"count": data.get("count"), "raw": payload}

    # ---- 和平小店（腾讯道聚城 H5）----
    XPD_EMBED_API = "https://www.douyu.com/japi/carnival/nc/txEmbed/getIframeUrl"
    XPD_H5_REFERER = "https://www.douyu.com/topic/h5/hpxd01"
    XPD_ROLE_PARAM_API = "https://apiv2.douyucdn.cn/japi/carnivalApi/tencent/roleParam"
    XPD_BIND_INFO_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/bindInfo"
    XPD_DAOJU_REFERER = "https://app.daoju.qq.com/"
    XPD_GET_ROLE_API = "https://apps.game.qq.com/daoju/igw/live"
    XPD_ORDER_API = "https://apps.game.qq.com/daoju/igw/live/"
    # 2026-08 抓包(8.6-和平小店)实证:本期走 xn_live_cjm 变体(旧期为 recommend_live/common)
    XPD_RECOMMEND_API = "https://apps.game.qq.com/daoju/v3/recommend_xn_live_cjm/common"
    XPD_BALANCE_API = "https://apps.game.qq.com/daoju/igw/live/"
    XPD_LIVE_BUY_API = "https://apps.game.qq.com/daoju/igw/livebuy/"
    XPD_GOODS_PAGE_SIZE = 10
    XPD_UA = (
        "Mozilla/5.0 (Linux; Android 12; HBN-AL00 Build/V417IR; wv) "
        "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
        "Chrome/101.0.4951.61 Mobile Safari/537.36, Douyu_Android"
    )

    @staticmethod
    def _xpd_int(value: Any) -> int | None:
        try:
            return int(value)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _xpd_parse_var(text: str, varname: str) -> dict[str, Any]:
        """解析道聚城 `var xxx={...};` 形式的响应（容忍等号两侧空白）。"""
        match = re.search(rf"var\s+{varname}\s*=", text)
        if not match:
            raise DouyuActivityError(f"响应中找不到 var {varname}=: {text[:200]}")
        chunk = text[match.end():].strip()
        if chunk.endswith(";"):
            chunk = chunk[:-1]
        try:
            return json.loads(chunk)
        except json.JSONDecodeError as exc:
            raise DouyuActivityError(f"var {varname}= 解析失败: {chunk[:200]}") from exc

    @classmethod
    def _xpd_daoju_headers(cls) -> dict[str, str]:
        return {
            "User-Agent": cls.XPD_UA,
            "Accept": "*/*",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "X-Requested-With": "air.tv.douyu.android",
            "Sec-Fetch-Site": "same-site",
            "Sec-Fetch-Mode": "no-cors",
            "Sec-Fetch-Dest": "script",
            "Referer": cls.XPD_DAOJU_REFERER,
        }

    def _xpd_token(self) -> str:
        """拼斗鱼 token: acf_uid_1_acf_stk_0_acf_ltkid。"""
        uid = cookie_value(self.cookie, "acf_uid")
        stk = cookie_value(self.cookie, "acf_stk")
        ltkid = cookie_value(self.cookie, "acf_ltkid")
        if not uid or not stk or not ltkid:
            raise DouyuActivityError("Cookie 缺少 acf_uid/acf_stk/acf_ltkid，无法获取小店参数")
        return f"{uid}_1_{stk}_0_{ltkid}"

    def xpd_embed_query(self, *, act_alias: str, rid: str) -> dict[str, Any]:
        """从斗鱼换取和平小店 H5 参数（code/sig/timestamp/actId 等，每次现签）。"""
        payload = self._request_json(
            "get",
            self.XPD_EMBED_API,
            "获取小店 H5 参数",
            params={"actAlias": act_alias, "rid": rid, "token": self._xpd_token()},
            headers={"Referer": self.XPD_H5_REFERER},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "获取小店 H5 参数失败")
        data = payload.get("data") or {}
        txurl_h5 = str(data.get("txurlH5") or "")
        if not txurl_h5:
            raise DouyuActivityError(f"获取小店 H5 参数失败: {payload}")
        query = parse_qs(urlsplit(txurl_h5).query)
        embed_query = {key: values[0] for key, values in query.items() if values}
        return {"query": embed_query, "txurl_h5": txurl_h5, "raw": payload}

    def xpd_get_role(
        self,
        *,
        embed_query: dict[str, str],
        act_id: str,
        rid: str = "",
    ) -> dict[str, Any]:
        """用小店 code/sig 换取当前绑定角色信息。

        2026-08 起道聚城校验升级：必须带 isCode=1 + authType=delegate + sAnchorId(房间号)。
        """
        params = {
            "acctype": "livelink",
            "_jsvar": "info",
            "_service": "other.livelink.getrole.out",
            "_biz_code": "cjm",
            "_act_id": act_id,
            "_app_id": "2123",
            "isCode": "1",
            "gameId": embed_query.get("gameId") or "cjm",
            "actId": embed_query.get("actId") or "",
            "appId": embed_query.get("appId") or "bp_cf",
            "livePlatId": embed_query.get("livePlatId") or "douyu",
            "code": embed_query.get("code") or "",
            "timestamp": embed_query.get("timestamp") or "",
            "v": embed_query.get("v") or "",
            "sig": embed_query.get("sig") or "",
            "authType": "delegate",
            "sAnchorId": rid,
            "sVideoId": "",
        }
        response = self._request(
            "get",
            self.XPD_GET_ROLE_API,
            source="查询小店绑定角色",
            params=params,
            headers=self._xpd_daoju_headers(),
        )
        info = self._xpd_parse_var(response.text, "info")
        plat_id = info.get("platId")
        area = info.get("area")
        return {
            "game_open_id": str(info.get("gameOpenId") or ""),
            "role_id": str(info.get("roleId") or ""),
            "role_name": str(info.get("roleName") or ""),
            "type": str(info.get("type") or ""),
            # 0 是合法的 iOS 平台值，不能用 `or` 当作缺失处理。
            "plat_id": str(plat_id) if plat_id is not None else "",
            "area": str(area) if area is not None else "",
            "raw": info,
        }

    def xpd_list_goods(
        self,
        *,
        embed_query: dict[str, str],
        act_id: str,
        openid: str,
        roleid: str,
        areaid: str = "1",
    ) -> dict[str, Any]:
        """查询和平小店全部商品，排除热门抢购分类并拉取全部分页。"""
        base_params = {
            "_service": "act.recommend.query",
            "_biz_code": "cjm",
            "_app_id": "2123",
            "app_id": "2123",
            "fields": (
                "iRelationGoodsid,dtModifyTime,sBuyLimitInfo,sGoodsWaterMark,dtShowBeginTime,"
                "dtShowEndTime,iSort,iGoodsId,sGoodsName,iJbPrice,iJbOrgPrice,iJb2Price,"
                "iJb2OrgPrice,iPrice,iOrgPrice,sGoodsPic,sGoodsDesc,sPayType,dtBeginTime,"
                "dtEndTime,iActionId,sExtShowInfo,sExtInfo,iCategoryId,dtRushBegin,dtRushEnd"
            ),
            "actionFields": "iActionId,sActionName,dtBeginTime,dtEndTime",
            "order_by": "dtShowBeginTime",
            "orderColumnBatch": '[{"key":"iSort","sort":"desc"},{"key":"dtShowBeginTime","sort":"desc"},{"key":"iGoodsId","sort":"asc"}]',
            "desc": "1",
            "_act_id": act_id,
            "actid": act_id,
            # category=76 是“热门抢购”；全部商品页明确将其排除。
            "excludeFields": "iCategoryId_76",
            "userid": openid,
            "roleid": roleid,
            "area": "1",
            "areaid": str(areaid),
            "need_limit": "1",
        }
        goods: list[dict[str, Any]] = []
        seen_ids: set[str] = set()
        pages: list[dict[str, Any]] = []
        offset = 0
        page_num = 1

        while True:
            params = {
                **base_params,
                "page_begin": str(offset),
                "page_num": str(page_num),
                "page_size": str(self.XPD_GOODS_PAGE_SIZE),
            }
            response = self._request(
                "get",
                self.XPD_RECOMMEND_API,
                source=f"刷新小店全部商品第 {page_num} 页",
                params=params,
                headers=self._xpd_daoju_headers(),
            )
            recommend = self._xpd_parse_var(response.text, "recommend")
            if str(recommend.get("errcode") or "0") != "0":
                raise DouyuActivityError(str(recommend.get("errdesc") or "小店商品查询失败"))
            pages.append(recommend)
            items = (((recommend.get("data") or {}).get("client_data") or {}).get("itemsdetail")) or []
            if not isinstance(items, list):
                raise DouyuActivityError("小店商品响应格式异常")

            for item in items:
                if not isinstance(item, dict):
                    continue
                commodity_id = str(item.get("iGoodsId") or "")
                if not commodity_id or commodity_id in seen_ids:
                    continue
                seen_ids.add(commodity_id)
                goods.append(
                    {
                        "commodity_id": commodity_id,
                        "name": str(item.get("sGoodsName") or ""),
                        "price": self._xpd_int(item.get("iPrice")),
                        "org_price": self._xpd_int(item.get("iOrgPrice")),
                        "category": str(item.get("iCategoryId") or ""),
                        "goods_left": self._xpd_int(item.get("iGoodsLeft")),
                        "raw": item,
                    }
                )

            if len(items) < self.XPD_GOODS_PAGE_SIZE:
                break
            offset += self.XPD_GOODS_PAGE_SIZE
            page_num += 1
            if page_num > 100:
                raise DouyuActivityError("小店商品分页超出安全上限")

        return {"goods": goods, "raw": {"pages": pages}}

    def xpd_purchase_records(
        self,
        *,
        embed_query: dict[str, str],
        act_id: str,
        page_size: int = 10,
        max_pages: int = 100,
    ) -> dict[str, Any]:
        """查询和平小店购买记录，接口参数来自道聚城 H5 抓包。"""
        page_size = max(1, min(int(page_size), 50))
        records: list[dict[str, Any]] = []
        pages: list[dict[str, Any]] = []
        total = 0

        for page in range(1, max_pages + 1):
            params = {
                "_service": "order.list",
                "_jsvar": "info",
                "acctype": "livelink",
                "_app_id": "2123",
                "hist": "0",
                "ps": str(page_size),
                "pn": str(page),
                "_biz_code": "cjm",
                "_act_id": str(act_id),
                "authType": "delegate",
                "actId": embed_query.get("actId") or "",
                "appId": embed_query.get("appId") or "bp_cf",
                "gameId": embed_query.get("gameId") or "cjm",
                "livePlatId": embed_query.get("livePlatId") or "douyu",
                "code": embed_query.get("code") or "",
                "timestamp": embed_query.get("timestamp") or "",
                "v": embed_query.get("v") or "",
                "sig": embed_query.get("sig") or "",
                "_sid": "8",
                "set": "8",
            }
            response = self._request(
                "get",
                self.XPD_ORDER_API,
                source=f"查询小店购买记录第 {page} 页",
                params=params,
                headers=self._xpd_daoju_headers(),
            )
            info = self._xpd_parse_var(response.text, "info")
            if str(info.get("ret") or "0") != "0":
                raise DouyuActivityError(str(info.get("msg") or "小店购买记录查询失败"))
            data = info.get("data") if isinstance(info.get("data"), dict) else {}
            page_records = data.get("list") if isinstance(data.get("list"), list) else []
            total = self._xpd_int(data.get("cnt")) or total
            pages.append(info)

            for record in page_records:
                if not isinstance(record, dict):
                    continue
                goods_info = record.get("sGoodsInfo")
                if isinstance(goods_info, str):
                    try:
                        goods_info = json.loads(goods_info)
                    except json.JSONDecodeError:
                        goods_info = {}
                goods_list = goods_info.get("list") if isinstance(goods_info, dict) else []
                goods = (
                    goods_list[0]
                    if isinstance(goods_list, list) and goods_list and isinstance(goods_list[0], dict)
                    else {}
                )
                app_ext = record.get("sAppExt")
                if isinstance(app_ext, str):
                    try:
                        app_ext = json.loads(app_ext)
                    except json.JSONDecodeError:
                        app_ext = {}
                records.append({
                    "serial": str(record.get("sSerialNum") or ""),
                    "goods_name": str(
                        goods.get("sGoodsName")
                        or goods.get("sPacketName")
                        or record.get("sActionDesc")
                        or ""
                    ),
                    "goods_image": str(goods.get("sGoodsPic") or (app_ext or {}).get("sGoodsPic") or ""),
                    "buy_time": str(record.get("dtBuyTime") or ""),
                    "pay_time": str(record.get("dtPayTime") or ""),
                    "status": str(record.get("iStatus") or ""),
                    "price": self._xpd_int(record.get("iPrice")),
                    "pay_amount": self._xpd_int(record.get("iPayAmount")),
                    "pay_type": str(record.get("sPayType") or ""),
                    "role_name": str(record.get("sRoleName") or ""),
                    "send_status": str(goods.get("iSendStatus") or ""),
                    "delivery_message": str(goods.get("deliveryMsg") or ""),
                    "raw": record,
                })

            if len(page_records) < page_size or len(records) >= total:
                break

        return {"records": records, "total": total or len(records), "raw": {"pages": pages}}

    def xpd_bind_qr(self, *, act_alias: str) -> dict[str, Any]:
        """生成和平小店绑定二维码链接（微信扫码进入 livelink 小程序绑定角色）。

        返回 data.url（可生成二维码）与 gameRoleBindUrl（直接角色绑定页）。
        """
        payload = self._request_json(
            "post",
            self.XPD_ROLE_PARAM_API,
            "生成小店绑定二维码",
            params={"client_sys": "android"},
            data={
                "token": self._xpd_token(),
                "uid": cookie_value(self.cookie, "acf_uid") or "",
                "client": "mobile",
                "actAlias": act_alias,
            },
            headers={
                "User-Device": "MDExZTUxMzEyYmYxMmY5MmQ3YmY1M2M4MDA3MDcxMTF8djcuOS4y",
                "aid": "android1",
                "User-Agent": "android/7.9.2 (android 9; ; M973Q)",
                "Accept-Language": "zh-CN",
                "Content-Type": "application/x-www-form-urlencoded",
            },
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "生成小店绑定二维码失败")
        data = payload.get("data") or {}
        url = str(data.get("url") or "")
        if not url:
            raise DouyuActivityError(f"生成小店绑定二维码失败: {payload}")
        return {
            "url": url,
            "game_role_bind_url": str(data.get("gameRoleBindUrl") or ""),
            "nick_name": str((data.get("queryParams") or {}).get("nickName") or ""),
            "raw": payload,
        }

    def xpd_bind_info(self, *, act_alias: str) -> dict[str, Any]:
        """查询和平小店绑定信息（bindInfo）。"""
        payload = self._request_json(
            "get",
            self.XPD_BIND_INFO_API,
            "查询小店绑定信息",
            params={"actAlias": act_alias, "token": self._xpd_token()},
        )
        if payload.get("error") not in (0, "0", None):
            raise DouyuActivityError(payload.get("msg") or "查询小店绑定信息失败")
        data = payload.get("data") or {}
        game_role = data.get("gameRole") if isinstance(data.get("gameRole"), dict) else {}
        game_account = data.get("gameAccount") if isinstance(data.get("gameAccount"), dict) else {}
        return {
            "bind_account": self._xpd_int(data.get("bindAccount")),
            "bind_role": self._xpd_int(data.get("bindRole")),
            "need_bind_role": self._xpd_int(data.get("needBindRole")),
            "nick": str(game_account.get("nick") or ""),
            "role_name": str(game_role.get("roleName") or ""),
            "area_name": str(game_role.get("areaName") or ""),
            "plat_name": str(game_role.get("platName") or ""),
            "raw": payload,
        }

    def xpd_balance(
        self,
        *,
        embed_query: dict[str, str],
        act_id: str,
        openid: str,
        roleid: str,
        plat: str,
        areaid: str = "1",
    ) -> dict[str, Any]:
        """查询和平小店角色点券余额。"""
        params = {
            "_jsvar": "banlanceInfo",
            "_service": "pay.midas.dq.get.ttpp",
            "_app_id": "2123",
            "acctype": "ttpp",
            "areaid": str(areaid),
            "eventid": "",
            "interwork": "0",
            "openid": openid,
            "openkey": "openkey",
            "partition": "0",
            "pay_token": "",
            "plat": str(plat),
            "plat_pc": "0",
            "roleid": roleid,
            "_biz_code": "cjm",
            "_act_id": act_id,
            # 抓包中的接口使用毫秒级时间戳。
            "_time": str(int(time.time() * 1000)),
            "_sid": "6",
        }
        response = self._request(
            "get",
            self.XPD_BALANCE_API,
            source="查询小店点券余额",
            params=params,
            headers=self._xpd_daoju_headers(),
        )
        info = self._xpd_parse_var(response.text, "banlanceInfo")
        return {"balance": self._xpd_int(info.get("balance")), "raw": info}

    def xpd_fragments(
        self,
        *,
        embed_query: dict[str, str],
        act_id: str,
        openid: str,
        roleid: str,
        plat: str,
        areaid: str = "1",
    ) -> dict[str, Any]:
        """查询和平小店角色扭蛋碎片数量（gamecoin jb2）。"""
        params = {
            "_jsvar": "jbInfos",
            "_service": "pay.idip.gamecoin.get",
            "_app_id": "2123",
            "acctype": "ttpp",
            "areaid": str(areaid),
            "eventid": "",
            "interwork": "0",
            "openid": openid,
            "openkey": "openkey",
            "partition": "0",
            "pay_token": "",
            "plat": str(plat),
            "plat_pc": "0",
            "roleid": roleid,
            "_biz_code": "cjm",
            "_act_id": act_id,
            "coin_type": "1",
            "_sid": "6",
        }
        response = self._request(
            "get",
            self.XPD_BALANCE_API,
            source="查询小店扭蛋碎片",
            params=params,
            headers=self._xpd_daoju_headers(),
        )
        info = self._xpd_parse_var(response.text, "jbInfos")
        return {"fragments": self._xpd_int(info.get("jb2")), "raw": info}

    def xpd_exchange_goods(
        self,
        *,
        embed_query: dict[str, str],
        act_id: str,
        rid: str,
        commodity_id: str,
        price: int,
        picture: str = "",
        pay_type: int = 1,
        action_id: str = "",
    ) -> dict[str, Any]:
        """兑换和平小店商品。

        道聚城的 code/sig 为短时授权参数，调用方必须在每次兑换前重新获取。
        该接口会实际扣除点券或扭蛋碎片，因此遇到网络错误不做自动重试。
        """
        if pay_type not in (1, 5):
            raise DouyuActivityError("小店兑换仅支持点券或扭蛋碎片")
        if not commodity_id:
            raise DouyuActivityError("小店兑换缺少商品 ID")
        if price < 0:
            raise DouyuActivityError("小店兑换商品价格无效")

        params = {
            "_service": "buy.plug.svr.web",
            "_plug_id": "7200",
            "_jsvar": "buyInfo",
            "acctype": "livelink",
            "_app_id": "2123",
            "iActionId": action_id or act_id,
            "propid": commodity_id,
            "paytype": str(pay_type),
            "source": rid,
            "sAnchorId": rid,
            "attach": json.dumps({"ch": "0"}, separators=(",", ":")),
            "apptype": "3",
            "_biz_code": "cjm",
            "_act_id": act_id,
            "sLiveUserInfo": json.dumps(
                {"sAnchorId": rid, "sVideoId": 0}, separators=(",", ":"),
            ),
            "appext": json.dumps(
                {"user_price": price, "sGoodsPic": picture, "paytype": pay_type},
                separators=(",", ":"),
            ),
            "isCode": "1",
            "authType": "delegate",
            "_sid": "8",
            "set": "8",
        }
        for key in ("gameId", "actId", "appId", "livePlatId", "code", "timestamp", "v", "sig"):
            value = str(embed_query.get(key) or "")
            if value:
                params[key] = value

        response = self._request(
            "get",
            self.XPD_LIVE_BUY_API,
            source="兑换小店商品",
            params=params,
            headers=self._xpd_daoju_headers(),
        )
        info = self._xpd_parse_var(response.text, "buyInfo")
        if str(info.get("ret")) != "0" or str(info.get("msg") or "").lower() != "ok":
            raise DouyuActivityError(str(info.get("msg") or "小店兑换失败"))
        return {
            "commodity_id": commodity_id,
            "pay_type": pay_type,
            "price": price,
            "serial": str(info.get("serial") or info.get("sSerialNum") or ""),
            "event_id": str(info.get("event_id") or ""),
            "new_balance": self._xpd_int(info.get("newBalance")),
            "raw": info,
        }