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

"""斗鱼任务执行器：手册开通与积分（由 douyu_runner.py 按功能域拆分）。"""

from __future__ import annotations

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

from sqlalchemy.orm import Session

from core.douyu import DouyuActivityClient, DouyuActivityError

from ..models import Account, DouyuTask
from .account_events_service import EVENT_ELITE_OPENED, EVENT_ESPORTS_OPENED, emit
from .douyu_runner_core import DOUYU_PAYMENT_POLL_INTERVAL, DOUYU_PAYMENT_POLL_SECONDS
from .douyu_service import account_uid, update_account_profile_from_cookie


class ManualMixin:
    """手册域：精英/电竞手册开通支付、积分查询与到账轮询。"""

    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 _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: ...

        @classmethod
        def _confirm_act_alias(cls, config: dict) -> str: ...

        @classmethod
        def _bind_qr_act_alias(cls, config: dict) -> str: ...

    def _refresh_account_points(
        self,
        client: DouyuActivityClient,
        account: Account,
        cookie: str,
        *,
        ctn: str | None = None,
    ) -> dict:
        """刷新账号积分并写回账号表。"""
        uid = account_uid(account, cookie)
        if not uid:
            raise DouyuActivityError("Cookie 中没有 acf_uid，无法查询积分")
        ctn_value = ctn or client.acf_ccn(refresh_subscribe=False)
        result = client.query_points(uid=uid, ctn=ctn_value)
        points = self._to_int(result.get("points"))
        account.uid = uid
        account.points = points
        update_account_profile_from_cookie(account, cookie)
        account.updated_at = datetime.now(UTC)
        return {"points": points, "points_query": result}

    @staticmethod
    def _format_expire_text(expire_time: int | None) -> str:
        """把到期时间戳格式化成中文描述，缺失或为 0 时返回空串。"""
        if not expire_time or expire_time <= 0:
            return ""
        expire_at = datetime.fromtimestamp(expire_time, tz=UTC).astimezone()
        return f"（到期 {expire_at:%Y-%m-%d %H:%M}）"

    def _refresh_elite_handbook(
        self,
        client: DouyuActivityClient,
        account: Account,
        *,
        manual_id: str,
        rid: str = "",
    ) -> dict:
        """查询精英宝典开通状态（manualType/expireTime）并写回账号。

        积分区分不了「开通后积分花完」和「从未开通」，manualType 才是权威口径：
        0=未开通，>=1=已开通。
        """
        result = client.elite_user_info(manual_id=manual_id, rid=rid)
        manual_type = self._to_int(result.get("manual_type")) or 0
        manual_score = self._to_int(result.get("manual_score"))
        expire_time = self._to_int(result.get("expire_time")) or 0
        opened = manual_type >= 1
        account.elite_opened = opened
        account.elite_expire_time = expire_time or None
        if manual_score is not None:
            account.points = manual_score
        account.updated_at = datetime.now(UTC)
        return {
            "elite_manual_type": manual_type,
            "elite_manual_score": manual_score,
            "elite_expire_time": expire_time,
            "elite_opened": opened,
            "elite_user_info": result,
            "points": manual_score,
        }

    def _execute_query_elite_status(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        """查询精英宝典是否已开通、到期时间与积分。"""
        manual_id = str(config.get("manual_id") or "").strip()
        if not manual_id:
            self._mark_task(db, task, "failed", "请先配置精英宝典 manualID")
            return
        client = self._client(cookie)
        result = self._refresh_elite_handbook(
            client,
            account,
            manual_id=manual_id,
            rid=str(config.get("rid") or ""),
        )
        account.updated_at = datetime.now(UTC)
        state = "已开通" if result["elite_opened"] else "未开通"
        expire_text = self._format_expire_text(result["elite_expire_time"])
        points = result["elite_manual_score"]
        self._mark_task(
            db,
            task,
            "success",
            f"精英宝典{state}{expire_text}，积分: {points if points is not None else '-'}",
            result,
        )

    def _wait_points_after_payment(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        client: DouyuActivityClient,
        cookie: str,
        ctn: str,
        result: dict,
        *,
        baseline_points: int | None = None,
    ) -> bool:
        """等待宝典支付到账。

        判定口径是「积分比开通前有增长」：旧口径 `积分 >= 300` 会让账上本来就
        有 300+ 积分的账号在出码后第一轮轮询就被判成功，二维码刚弹出来就消失
        （线上 2~4 秒 success 的案例都是这种）。拿不到基线时才退回旧口径。
        """
        deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
        result["payment_polling"] = True
        result["payment_target_points"] = 300
        result["elite_points_baseline"] = baseline_points
        poll_count = 0
        last_points = None
        while not self._stop.is_set() and time.monotonic() <= deadline:
            try:
                points_result = self._refresh_account_points(
                    client, account, cookie, ctn=ctn
                )
                db.commit()
                poll_count += 1
                last_points = points_result["points"]
                result.update(points_result)
                result["payment_poll_count"] = poll_count
                result["payment_polling"] = True
                if last_points is not None:
                    arrived = (
                        last_points > baseline_points
                        if baseline_points is not None
                        else last_points >= 300
                    )
                    if arrived:
                        result["payment_polling"] = False
                        result["elite_opened"] = True
                        return True
                self._update_task_progress(
                    db,
                    task,
                    "running",
                    f"精英宝典支付码已生成，等待开通到账（当前积分 {last_points if last_points is not None else '-'}）",
                    result,
                )
            except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
                poll_count += 1
                result["payment_poll_count"] = poll_count
                result["payment_poll_error"] = str(exc)
                self._update_task_progress(
                    db, task, "running", f"等待开通到账: {exc}", result
                )
            if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
                break
        result["payment_polling"] = False
        result["elite_opened"] = False
        result["points"] = last_points
        return False

    def _refresh_esports_handbook(
        self,
        client: DouyuActivityClient,
        account: Account,
        *,
        manual_id: str,
    ) -> dict:
        """刷新电竞手册开通状态并将积分写回账号。"""
        result = client.esports_user_info(manual_id=manual_id)
        manual_type = self._to_int(result.get("manual_type"))
        manual_score = self._to_int(result.get("manual_score"))
        account.esports_points = manual_score
        # 到期时间也落库 —— 否则界面只能显示"已开通"，看不到什么时候到期
        expire_raw = result.get("expire_time")
        expire_int = (
            int(expire_raw) if str(expire_raw or "").strip().isdigit() else None
        )
        # ⚠️ 斗鱼用 0 表示「没有到期时间」。直接存 0 的话界面会显示 1970/1/1
        # （这个坑在积分到期上踩过一次，这里别再踩）。
        account.esports_expire_time = (
            expire_int if expire_int and expire_int > 0 else None
        )
        account.updated_at = datetime.now(UTC)
        return {
            "esports_manual_type": manual_type,
            "esports_manual_score": manual_score,
            "esports_expire_time": result.get("expire_time"),
            "esports_user_info": result,
            "esports_points": manual_score,
            "points": manual_score,
        }

    def _wait_esports_open_after_payment(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        client: DouyuActivityClient,
        *,
        manual_id: str,
        result: dict,
        baseline_manual_type: int | None,
        baseline_manual_score: int | None,
    ) -> bool:
        """等待电竞手册支付到账，以 manualType=1 或积分变化作为成功条件。"""
        deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
        result["payment_polling"] = True
        result["esports_manual_type_baseline"] = baseline_manual_type
        result["esports_manual_score_baseline"] = baseline_manual_score
        poll_count = 0
        last_manual_type = baseline_manual_type
        last_manual_score = baseline_manual_score

        while not self._stop.is_set() and time.monotonic() <= deadline:
            try:
                handbook_result = self._refresh_esports_handbook(
                    client,
                    account,
                    manual_id=manual_id,
                )
                db.commit()
                poll_count += 1
                last_manual_type = handbook_result["esports_manual_type"]
                last_manual_score = handbook_result["esports_manual_score"]
                result.update(handbook_result)
                result["payment_poll_count"] = poll_count
                result["payment_polling"] = True
                opened = (last_manual_type is not None and last_manual_type >= 1) or (
                    baseline_manual_score is not None
                    and last_manual_score is not None
                    and last_manual_score > baseline_manual_score
                )
                if opened:
                    result["payment_polling"] = False
                    result["esports_opened"] = True
                    return True
                self._update_task_progress(
                    db,
                    task,
                    "running",
                    "电竞手册支付码已生成，等待开通到账"
                    f"（类型 {last_manual_type if last_manual_type is not None else '-'}，"
                    f"积分 {last_manual_score if last_manual_score is not None else '-'}）",
                    result,
                )
            except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
                poll_count += 1
                result["payment_poll_count"] = poll_count
                result["payment_poll_error"] = str(exc)
                self._update_task_progress(
                    db, task, "running", f"等待电竞手册到账: {exc}", result
                )
            if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
                break

        result["payment_polling"] = False
        result["esports_opened"] = False
        result["esports_manual_type"] = last_manual_type
        result["esports_manual_score"] = last_manual_score
        result["esports_points"] = last_manual_score
        result["points"] = last_manual_score
        return False

    def _execute_create_elite_qr(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        client = self._client(cookie)
        ctn = str(self._task_payload(task).get("ctn") or "")
        if not ctn:
            ctn = client.acf_ccn(refresh_subscribe=True)
        act_alias = self._confirm_act_alias(config) or self._bind_qr_act_alias(config)
        if not act_alias:
            self._mark_task(db, task, "failed", "请先配置开通宝典活动 actAlias")
            return

        # 开通前先查状态（对齐电竞手册）：已开通的账号直接给出结论，不再生成
        # 支付码 —— 既避免二维码刚弹出来就因「积分本来就够」被判成功而消失，
        # 也避免给已开通账号生成一张无意义的真实支付单。
        manual_id = str(config.get("manual_id") or "").strip()
        baseline_points: int | None = None
        if manual_id:
            try:
                baseline = self._refresh_elite_handbook(
                    client,
                    account,
                    manual_id=manual_id,
                    rid=str(config.get("rid") or ""),
                )
                db.commit()
            except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
                self._push_log(
                    "warning", f"开通前查询精英宝典状态失败，按未开通流程继续: {exc}"
                )
            else:
                if baseline["elite_opened"]:
                    expire_text = self._format_expire_text(
                        baseline["elite_expire_time"]
                    )
                    points = baseline["elite_manual_score"]
                    account.updated_at = datetime.now(UTC)
                    self._mark_task(
                        db,
                        task,
                        "success",
                        f"精英宝典已开通{expire_text}，积分: "
                        f"{points if points is not None else '-'}",
                        {
                            **baseline,
                            "elite_already_opened": True,
                            "payment_polling": False,
                        },
                    )
                    return
                baseline_points = baseline["elite_manual_score"]
                # 到账轮询读的是 credit/balance，基线也用同一个接口取，
                # 避免两个口径不一致时把「余额本来就高」误判成刚到账。
                try:
                    baseline_points = self._refresh_account_points(
                        client, account, cookie, ctn=ctn
                    )["points"]
                    db.commit()
                except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
                    self._push_log(
                        "warning", f"开通前查询积分基线失败，改用手册积分: {exc}"
                    )

        result = client.create_elite_qr(
            ctn=ctn,
            act_alias=act_alias,
            amount=int(config["elite_amount"]),
            room_id=str(config["room_id"]),
        )
        account.updated_at = datetime.now(UTC)
        self._update_task_progress(
            db, task, "running", "精英宝典支付码已生成，等待开通到账", result
        )
        opened = self._wait_points_after_payment(
            db,
            task,
            account,
            client,
            cookie,
            ctn,
            result,
            baseline_points=baseline_points,
        )
        if self._stop.is_set():
            self._mark_task(db, task, "stopped", "任务已停止", result)
            return
        if opened:
            account.elite_opened = True
            account.updated_at = datetime.now(UTC)
            # elite_opened 是**覆盖式**布尔值：开通之后再操作也看不出「什么时候开的」。
            # 在这里记一条事件，补上开通时间。
            # 开通是**资产减少**（花积分）。只记余额、不记花掉多少的话，
            # 账号的积分流水就对不上账 —— 进项有金额、出项没有。
            spent = _points_spent(baseline_points, account.points)
            emit(
                db,
                account_id=account.id,
                event_type=EVENT_ELITE_OPENED,
                title=f"精英宝典开通 -{spent}" if spent else "精英宝典开通",
                occurred_at=account.updated_at,
                amount=-spent if spent else None,
                detail={
                    "points": account.points,
                    "expire_time": account.elite_expire_time,
                    "spent": spent,
                },
                source="manual",
                source_ref=(
                    f"elite_opened:{account.id}:{int(account.updated_at.timestamp())}"
                ),
            )
            self._mark_task(
                db, task, "success", f"精英宝典已开通，积分: {account.points}", result
            )
            return
        self._mark_task(
            db,
            task,
            "failed",
            f"未检测到精英宝典开通到账，当前积分: {account.points if account.points is not None else '-'}",
            result,
        )

    def _execute_create_esports_qr(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """生成电竞手册支付二维码，并通过活动状态确认开通到账。"""
        client = self._client(cookie)
        ctn = str(self._task_payload(task).get("ctn") or "")
        if not ctn:
            ctn = client.acf_ccn(refresh_subscribe=True)
        act_alias = str(config.get("esports_act_alias") or "").strip()
        manual_id = str(config.get("esports_manual_id") or "").strip()
        if not act_alias or not manual_id:
            self._mark_task(
                db, task, "failed", "请先配置电竞手册活动 actAlias 和 manualID"
            )
            return

        baseline_manual_type = None
        baseline_manual_score = None
        baseline_result: dict = {}
        try:
            baseline_result = self._refresh_esports_handbook(
                client, account, manual_id=manual_id
            )
            baseline_manual_type = baseline_result["esports_manual_type"]
            baseline_manual_score = baseline_result["esports_manual_score"]
            db.commit()
        except Exception as exc:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
            self._push_log("warning", f"生成电竞手册支付码前查询活动状态失败: {exc}")

        if baseline_manual_type is not None and baseline_manual_type >= 1:
            # 这里已确认开通：写独立状态列（以前只把它塞进 esports_bind_status）
            account.esports_opened = True
            account.updated_at = datetime.now(UTC)
            self._mark_task(
                db,
                task,
                "success",
                f"电竞手册已开通，积分: {baseline_manual_score if baseline_manual_score is not None else '-'}",
                {**baseline_result, "esports_opened": True, "payment_polling": False},
            )
            return

        result = client.create_esports_qr(
            ctn=ctn,
            act_alias=act_alias,
            amount=int(config["esports_amount"]),
            room_id=str(config["room_id"]),
        )
        result.update(baseline_result)
        account.updated_at = datetime.now(UTC)
        self._update_task_progress(
            db, task, "running", "电竞手册支付码已生成，等待开通到账", result
        )
        opened = self._wait_esports_open_after_payment(
            db,
            task,
            account,
            client,
            manual_id=manual_id,
            result=result,
            baseline_manual_type=baseline_manual_type,
            baseline_manual_score=baseline_manual_score,
        )
        if self._stop.is_set():
            self._mark_task(db, task, "stopped", "任务已停止", result)
            return
        if opened:
            account.updated_at = datetime.now(UTC)
            # 与精英宝典同理：状态字段是覆盖式的，这里补上「什么时候开的」。
            # 注意只在**真正开通**这一支发事件；上面「基线查到已开通」那一支是检测，
            # 每次都发会刷屏。
            # 同理：开通电竞手册也要记清花了多少积分
            spent = _points_spent(baseline_manual_score, account.esports_points)
            emit(
                db,
                account_id=account.id,
                event_type=EVENT_ESPORTS_OPENED,
                title=f"电竞手册开通 -{spent}" if spent else "电竞手册开通",
                occurred_at=account.updated_at,
                amount=-spent if spent else None,
                detail={"points": account.esports_points, "spent": spent},
                source="manual",
                source_ref=(
                    f"esports_opened:{account.id}:{int(account.updated_at.timestamp())}"
                ),
            )
            self._mark_task(
                db,
                task,
                "success",
                f"电竞手册已开通，积分: {account.esports_points if account.esports_points is not None else '-'}",
                result,
            )
            return
        self._mark_task(
            db,
            task,
            "failed",
            "未检测到电竞手册开通到账"
            f"，类型: {result.get('esports_manual_type', '-')}，"
            f"积分: {result.get('esports_manual_score', '-')}",
            result,
        )

    def _execute_query_esports_points(
        self,
        db: Session,
        task: DouyuTask,
        account: Account,
        cookie: str,
        config: dict,
    ):
        """查询电竞手册积分。"""
        manual_id = str(config.get("esports_manual_id") or "").strip()
        if not manual_id:
            self._mark_task(db, task, "failed", "请先配置电竞手册 manualID")
            return
        client = self._client(cookie)
        result = self._refresh_esports_handbook(client, account, manual_id=manual_id)
        account.updated_at = datetime.now(UTC)
        points = result["esports_points"]
        self._mark_task(
            db,
            task,
            "success",
            f"电竞积分: {points if points is not None else '-'}",
            result,
        )

    def _execute_query_points(
        self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
    ):
        client = self._client(cookie)
        ctn = client.acf_ccn(refresh_subscribe=False)
        result = self._refresh_account_points(client, account, cookie, ctn=ctn)
        account.updated_at = datetime.now(UTC)
        points = result["points"]
        self._mark_task(
            db,
            task,
            "success",
            f"积分: {points if points is not None else '-'}",
            result,
        )


def _points_spent(before: int | None, after: int | None) -> int | None:
    """开通前后积分余额的差（正数 = 花掉的）。

    用**实测的余额差**而不是"开通要多少分"的常量 —— 活动价、折扣、
    赠送都会让实际扣减不同，只有观测到的差是真的。
    任一侧拿不到就返回 None（事件就不写金额，而不是编一个）。
    """
    if not isinstance(before, (int, float)) or not isinstance(after, (int, float)):
        return None
    delta = int(before) - int(after)
    return delta if delta > 0 else None