"""斗鱼任务执行器:手册开通与积分(由 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"] payload = self._task_payload(task) if payload.get("require_opened") and not result["elite_opened"]: self._mark_task( db, task, "failed", "精英宝典未开通,无法继续充值或兑换,请先开通手册", result, ) return 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 # 开通状态也要落库 —— 与精英对称。 # 之前只有 elite 那支写 `account.elite_opened`,电竞的 `esports_opened` # **只在"开通"任务里被设置**,于是详情页的电竞「已开通/未开通」 # 根本不会被一键查询更新(用户报"电竞到期时间拿不到"时顺带查出来的)。 # 判定口径与 create_esports_qr 一致:manual_type >= 1 即已开通。 account.esports_opened = manual_type is not None and manual_type >= 1 # 到期时间也落库 —— 否则界面只能显示"已开通",看不到什么时候到期 expire_raw = result.get("expire_time") # 先规整成字符串再判 isdigit:原来写 `int(expire_raw) if str(expire_raw # or "").strip().isdigit()` —— 判的是 A、转的是 B,类型检查器收窄不了, # 而且同一份值取了两遍。规整到一个局部变量后两个问题一起没了。 expire_text = str(expire_raw or "").strip() expire_int = int(expire_text) if expire_text.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