diff --git a/core/douyu/__init__.py b/core/douyu/__init__.py index 48cb635..e6a7216 100644 --- a/core/douyu/__init__.py +++ b/core/douyu/__init__.py @@ -5,6 +5,7 @@ from .login_api import LoginAPIStrategy from .login_api_wgapi import WgapiLoginAPI from .login_api_iframe import IframeLoginAPI from .email_verifier import EmailVerifier +from .activity_client import DouyuActivityClient, DouyuActivityError __all__ = [ "DouyuLogin", @@ -13,4 +14,6 @@ __all__ = [ "WgapiLoginAPI", "IframeLoginAPI", "EmailVerifier", + "DouyuActivityClient", + "DouyuActivityError", ] diff --git a/core/douyu/activity_client.py b/core/douyu/activity_client.py new file mode 100644 index 0000000..1e308d0 --- /dev/null +++ b/core/douyu/activity_client.py @@ -0,0 +1,465 @@ +"""斗鱼活动、兑换与充值接口客户端。""" + +from __future__ import annotations + +import json +import random +import string +from collections.abc import Callable +from typing import Any +from urllib.parse import unquote + +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/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" + EXCHANGE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchange" + EXCHANGE_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchangeList" + 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" + PAY_QR_API = "https://www.douyu.com/japi/oms/web/pay/getQrCode" + 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), + ): + self.cookie = normalize_douyu_cookie(cookie) + self.timeout = timeout + self.logger = logger or (lambda _msg: None) + self.session = requests.Session() + 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 _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, + } + + 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) + response = self.session.request( + method, + url, + headers=req_headers, + timeout=kwargs.pop("timeout", self.timeout), + **kwargs, + ) + self._merge_response_cookie(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) -> str: + """获取 cvl_csrf_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 + + 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]: + """生成腾讯游戏绑定二维码链接。""" + ctn = self.acf_ccn(refresh_subscribe=True) + payload = self._request_json( + "post", + self.ROLE_PARAM_API, + "获取绑定二维码", + data={"actAlias": act_alias, "ctn": ctn}, + headers={ + "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", + "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} + + 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/"}, + ) + data = payload.get("data") or {} + game_account = data.get("gameAccount") or {} + game_role = data.get("gameRole") or {} + return { + "bind_account": data.get("bindAccount"), + "bind_role": data.get("bindRole"), + "bind_act": data.get("bindAct"), + "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 "", + "change_role_wait_time": data.get("changeRoleWaitTime"), + "can_change_role": data.get("canChangeRole"), + "raw": payload, + } + + 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]: + """生成精英宝典支付二维码。""" + 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, "item": item_payload, "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_elite_gift( + self, + *, + gift_count: int, + room_id: str, + gift_id: str, + skin_id: str, + ) -> dict[str, Any]: + """赠送精英令礼物。""" + payload = self._request_json( + "post", + self.DONATE_API, + "赠送精英令", + 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 "赠送精英令失败") + return {"raw": payload} + + 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} + + def exchange_goods(self, *, manual_id: str, rid: str, commodity_id: str, ctn: str | None = None) -> dict[str, Any]: + """兑换活动商品。""" + ctn_value = ctn or cookie_value(self.cookie, "acf_ccn") or self.acf_ccn(refresh_subscribe=False) + token = self.csrf_token() + randstr = "".join(random.choices(string.ascii_letters + string.digits, k=16)) + data = { + "manualID": manual_id, + "rid": rid, + "commodityID": commodity_id, + "ctn": ctn_value, + "csrfToken": token, + "randstr": randstr, + "token[error][code]": "-1", + "token[data]": "", + } + payload = self._request_json( + "post", + self.EXCHANGE_API, + "兑换商品", + data=data, + headers={ + "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", + "Referer": f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}", + "sec-ch-ua": f'"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Windows"', + "Sec-Fetch-Site": "same-origin", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Dest": "empty", + }, + ) + if payload.get("error") not in (0, "0"): + raise DouyuActivityError(payload.get("msg") or "兑换商品失败") + return { + "commodity_id": commodity_id, + "ctn": ctn_value, + "csrf_token": token, + "randstr": randstr, + "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} diff --git a/core/douyu/cookie_utils.py b/core/douyu/cookie_utils.py new file mode 100644 index 0000000..6d0040b --- /dev/null +++ b/core/douyu/cookie_utils.py @@ -0,0 +1,61 @@ +"""斗鱼 Cookie 规范化工具。""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping + +import requests + + +def cookie_pairs(cookie: str) -> list[tuple[str, str]]: + """把浏览器 Cookie 字符串拆成 key/value 对。""" + pairs: list[tuple[str, str]] = [] + for part in (cookie or "").split(";"): + item = part.strip() + if not item or "=" not in item: + continue + key, value = item.split("=", 1) + key = key.strip() + if key: + pairs.append((key, value.strip())) + return pairs + + +def normalize_cookie_pairs(pairs: Iterable[tuple[str, str]]) -> str: + """按 Cookie key 去重，保留最后一次出现的值。""" + ordered_keys: list[str] = [] + values: dict[str, str] = {} + for raw_key, raw_value in pairs: + key = str(raw_key).strip() + if not key: + continue + if key not in values: + ordered_keys.append(key) + values[key] = "" if raw_value is None else str(raw_value).strip() + return "; ".join(f"{key}={values[key]}" for key in ordered_keys) + + +def normalize_douyu_cookie( + cookie: str | Mapping[str, str] | requests.cookies.RequestsCookieJar | None, +) -> str: + """把 Cookie 字符串、dict 或 CookieJar 转成去重后的浏览器 Cookie 字符串。""" + if not cookie: + return "" + if isinstance(cookie, str): + return normalize_cookie_pairs(cookie_pairs(cookie)) + return normalize_cookie_pairs(cookie.items()) + + +def cookie_value( + cookie: str | Mapping[str, str] | requests.cookies.RequestsCookieJar | None, + key: str, +) -> str: + """从 Cookie 中读取 key；有重复时以最后一次出现为准。""" + if not cookie or not key: + return "" + pairs = cookie_pairs(cookie) if isinstance(cookie, str) else list(cookie.items()) + value = "" + for item_key, item_value in pairs: + if item_key == key: + value = "" if item_value is None else str(item_value).strip() + return value diff --git a/web/backend/main.py b/web/backend/main.py index c392e57..e9b90a6 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -11,7 +11,7 @@ from fastapi.responses import FileResponse from starlette.middleware.base import BaseHTTPMiddleware from .database import init_db -from .routers import auth, users, accounts, account_check, login, proxy, cookies, huya +from .routers import auth, users, accounts, account_check, login, proxy, cookies, huya, douyu from .schemas import AppInfo from .version import get_app_version from utils import setup_logger @@ -30,12 +30,16 @@ async def lifespan(app: FastAPI): from .database import SessionLocal from .services.huya_service import cleanup_orphan_huya_tasks + from .services.douyu_service import cleanup_orphan_douyu_tasks db = SessionLocal() try: cleaned = cleanup_orphan_huya_tasks(db, message="任务已中断（服务重启）") if cleaned: logger.info(f"启动清理虎牙残留任务: {cleaned} 条") + cleaned_douyu = cleanup_orphan_douyu_tasks(db, message="任务已中断（服务重启）") + if cleaned_douyu: + logger.info(f"启动清理斗鱼残留任务: {cleaned_douyu} 条") finally: db.close() yield @@ -84,6 +88,7 @@ app.include_router(login.router) app.include_router(proxy.router) app.include_router(cookies.router) app.include_router(huya.router) +app.include_router(douyu.router) @app.get("/api/health") diff --git a/web/backend/migrations/versions/20260724_0008_douyu_activity_tasks.py b/web/backend/migrations/versions/20260724_0008_douyu_activity_tasks.py new file mode 100644 index 0000000..8338d09 --- /dev/null +++ b/web/backend/migrations/versions/20260724_0008_douyu_activity_tasks.py @@ -0,0 +1,150 @@ +"""新增斗鱼活动任务表 + +Revision ID: 20260724_0008 +Revises: 20260712_0007 +Create Date: 2026-07-24 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260724_0008" +down_revision: Union[str, None] = "20260712_0007" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _has_table(bind, table_name: str) -> bool: + return sa.inspect(bind).has_table(table_name) + + +def _columns(bind, table_name: str) -> set[str]: + if not _has_table(bind, table_name): + return set() + return {column["name"] for column in sa.inspect(bind).get_columns(table_name)} + + +def _indexes(bind, table_name: str) -> set[str]: + if not _has_table(bind, table_name): + return set() + return {index["name"] for index in sa.inspect(bind).get_indexes(table_name)} + + +def _add_column_if_missing(bind, table_name: str, column: sa.Column) -> None: + if column.name not in _columns(bind, table_name): + op.add_column(table_name, column) + + +def _create_index_if_missing(bind, name: str, table_name: str, columns: list[str], unique: bool = False) -> None: + if name not in _indexes(bind, table_name): + op.create_index(name, table_name, columns, unique=unique) + + +def upgrade() -> None: + bind = op.get_bind() + + _add_column_if_missing(bind, "accounts", sa.Column("uid", sa.String(length=32), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("nickname", sa.String(length=128), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("points", sa.Integer(), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("game_name", sa.String(length=128), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("game_channel", sa.String(length=128), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("gold_balance", sa.Integer(), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("exchange_balance", sa.Integer(), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("bind_status", sa.String(length=32), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("change_role_wait_time", sa.Integer(), nullable=True)) + _add_column_if_missing(bind, "accounts", sa.Column("updated_at", sa.DateTime(), nullable=True)) + _create_index_if_missing(bind, "ix_accounts_uid", "accounts", ["uid"]) + + if not _has_table(bind, "douyu_tasks"): + op.create_table( + "douyu_tasks", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("batch_id", sa.String(length=64), nullable=False), + sa.Column("account_id", sa.Integer(), nullable=False), + sa.Column("task_type", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=32), nullable=True), + sa.Column("message", sa.String(length=512), nullable=True), + sa.Column("result", sa.JSON(), nullable=True), + sa.Column("created_by", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=True), + sa.Column("finished_at", sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(["account_id"], ["accounts.id"]), + sa.ForeignKeyConstraint(["created_by"], ["users.id"]), + sa.PrimaryKeyConstraint("id"), + ) + _create_index_if_missing(bind, "ix_douyu_tasks_batch_id", "douyu_tasks", ["batch_id"]) + _create_index_if_missing(bind, "ix_douyu_tasks_task_type", "douyu_tasks", ["task_type"]) + + if not _has_table(bind, "douyu_config"): + op.create_table( + "douyu_config", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("manual_id", sa.String(length=64), nullable=True), + sa.Column("rid", sa.String(length=64), nullable=True), + sa.Column("bind_act_alias", sa.String(length=64), nullable=True), + sa.Column("confirm_act_alias", sa.String(length=64), nullable=True), + sa.Column("legacy_act_alias", sa.String(length=64), nullable=True), + sa.Column("room_id", sa.String(length=64), nullable=True), + sa.Column("elite_amount", sa.Integer(), nullable=True), + sa.Column("gold_pay_type", sa.Integer(), nullable=True), + sa.Column("gift_id", sa.String(length=64), nullable=True), + sa.Column("skin_id", sa.String(length=64), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + + if not _has_table(bind, "douyu_goods_snapshot"): + op.create_table( + "douyu_goods_snapshot", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("commodity_id", sa.String(length=64), nullable=False), + sa.Column("name", sa.String(length=256), nullable=True), + sa.Column("score", sa.Integer(), nullable=True), + sa.Column("status", sa.String(length=32), nullable=True), + sa.Column("raw", sa.JSON(), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + _create_index_if_missing( + bind, + "ix_douyu_goods_snapshot_commodity_id", + "douyu_goods_snapshot", + ["commodity_id"], + ) + + +def downgrade() -> None: + bind = op.get_bind() + if _has_table(bind, "douyu_goods_snapshot"): + indexes = _indexes(bind, "douyu_goods_snapshot") + if "ix_douyu_goods_snapshot_commodity_id" in indexes: + op.drop_index("ix_douyu_goods_snapshot_commodity_id", table_name="douyu_goods_snapshot") + op.drop_table("douyu_goods_snapshot") + if _has_table(bind, "douyu_config"): + op.drop_table("douyu_config") + if _has_table(bind, "douyu_tasks"): + indexes = _indexes(bind, "douyu_tasks") + if "ix_douyu_tasks_task_type" in indexes: + op.drop_index("ix_douyu_tasks_task_type", table_name="douyu_tasks") + if "ix_douyu_tasks_batch_id" in indexes: + op.drop_index("ix_douyu_tasks_batch_id", table_name="douyu_tasks") + op.drop_table("douyu_tasks") + + account_columns = _columns(bind, "accounts") + for column in [ + "updated_at", + "change_role_wait_time", + "bind_status", + "exchange_balance", + "gold_balance", + "game_channel", + "game_name", + "points", + "nickname", + "uid", + ]: + if column in account_columns: + op.drop_column("accounts", column) diff --git a/web/backend/models.py b/web/backend/models.py index 298855a..91ae99b 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -55,10 +55,21 @@ class Account(Base): assigned_to = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) tag = Column(String(64), default="") remark = Column(String(256), default="") + uid = Column(String(32), default="", index=True) + nickname = Column(String(128), default="") + points = Column(Integer, nullable=True) + game_name = Column(String(128), default="") + game_channel = Column(String(128), default="") + gold_balance = Column(Integer, nullable=True) + exchange_balance = Column(Integer, nullable=True) + bind_status = Column(String(32), default="") + change_role_wait_time = Column(Integer, nullable=True) created_at = Column(DateTime, default=_utcnow) + updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow) assigned_user = relationship("User", back_populates="assigned_accounts", foreign_keys=[assigned_to]) login_tasks = relationship("LoginTask", back_populates="account") + douyu_tasks = relationship("DouyuTask", back_populates="account") class LoginTask(Base): @@ -78,6 +89,55 @@ class LoginTask(Base): account = relationship("Account", back_populates="login_tasks") +class DouyuTask(Base): + """斗鱼业务任务""" + __tablename__ = "douyu_tasks" + + id = Column(Integer, primary_key=True, autoincrement=True) + batch_id = Column(String(64), nullable=False, index=True) + account_id = Column(Integer, ForeignKey("accounts.id"), nullable=False) + task_type = Column(String(64), nullable=False, index=True) + status = Column(String(32), default="pending") + message = Column(String(512), default="") + result = Column(JSON, nullable=True) + created_by = Column(Integer, ForeignKey("users.id"), nullable=False) + created_at = Column(DateTime, default=_utcnow) + finished_at = Column(DateTime, nullable=True) + + account = relationship("Account", back_populates="douyu_tasks") + + +class DouyuConfig(Base): + """斗鱼业务配置""" + __tablename__ = "douyu_config" + + id = Column(Integer, primary_key=True, autoincrement=True) + manual_id = Column(String(64), default="G4KA4Qnz4LDp7") + rid = Column(String(64), default="9263298") + bind_act_alias = Column(String(64), default="20250213NQCYX") + confirm_act_alias = Column(String(64), default="20260120QYOOB") + legacy_act_alias = Column(String(64), default="cjm") + room_id = Column(String(64), default="9263298") + elite_amount = Column(Integer, default=3000) + gold_pay_type = Column(Integer, default=1) + gift_id = Column(String(64), default="23643") + skin_id = Column(String(64), default="2942") + updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow) + + +class DouyuGoodsSnapshot(Base): + """斗鱼兑换商品快照""" + __tablename__ = "douyu_goods_snapshot" + + id = Column(Integer, primary_key=True, autoincrement=True) + commodity_id = Column(String(64), nullable=False, index=True) + name = Column(String(256), default="") + score = Column(Integer, nullable=True) + status = Column(String(32), default="") + raw = Column(JSON, nullable=True) + updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow) + + class HuyaAccount(Base): """虎牙账号""" __tablename__ = "huya_accounts" diff --git a/web/backend/permissions.py b/web/backend/permissions.py index bef9350..2ff47dc 100644 --- a/web/backend/permissions.py +++ b/web/backend/permissions.py @@ -23,6 +23,9 @@ PERMISSIONS = { # Cookie "cookie:view": "查看 Cookie", "cookie:export": "导出 Cookie", + # 斗鱼活动 + "douyu:task": "斗鱼任务管理", + "douyu:config": "斗鱼配置管理", # 虎牙 "huya:account": "虎牙账号管理（兼容旧权限）", "huya:view_all": "查看所有虎牙账号", @@ -57,6 +60,8 @@ ROLE_PERMISSIONS = { "login:view_all", "cookie:view", "cookie:export", + "douyu:task", + "douyu:config", "huya:account", "huya:view_all", "huya:import", diff --git a/web/backend/routers/douyu.py b/web/backend/routers/douyu.py new file mode 100644 index 0000000..ec38e45 --- /dev/null +++ b/web/backend/routers/douyu.py @@ -0,0 +1,324 @@ +"""斗鱼活动任务路由。""" + +from __future__ import annotations + +import asyncio +import threading +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect +from sqlalchemy import or_ +from sqlalchemy.orm import Session, joinedload + +from ..database import SessionLocal, get_db +from ..deps import authenticate_websocket, get_current_user, require_permission +from ..models import Account, DouyuConfig, DouyuGoodsSnapshot, DouyuTask, User +from ..permissions import user_has_permission +from ..schemas import ( + DouyuConfigOut, + DouyuConfigUpdate, + DouyuGoodsOut, + DouyuTaskAccountOut, + DouyuTaskBatchRequest, + DouyuTaskOut, +) +from ..services.douyu_runner import DouyuBatchRunner, douyu_batch_registry +from ..services.douyu_service import ( + DOUYU_CONFIG_FIELDS, + SUPPORTED_DOUYU_TASK_TYPES, + apply_douyu_config_defaults, + cleanup_orphan_douyu_tasks, + cookie_account_ids_query, + create_douyu_planned_tasks, + douyu_config_value, + ensure_douyu_config, +) + + +router = APIRouter(prefix="/api/douyu", tags=["斗鱼活动"]) + + +def _can_view_all(user: User) -> bool: + return user_has_permission(user, "account:view_all") + + +def _visible_task_accounts_query(db: Session, current: User): + """返回当前用户可用于斗鱼任务的账号查询。""" + cookie_ids = cookie_account_ids_query(db).subquery() + query = ( + db.query(Account) + .options(joinedload(Account.assigned_user)) + .filter(Account.id.in_(cookie_ids)) + ) + if _can_view_all(current): + return query + if user_has_permission(current, "account:view_assigned"): + return query.filter(Account.assigned_to == current.id) + raise HTTPException(status_code=403, detail="无权查看斗鱼账号") + + +def _account_out(account: Account) -> DouyuTaskAccountOut: + return DouyuTaskAccountOut( + id=account.id, + username=account.username, + uid=account.uid or "", + nickname=account.nickname or "", + tag=account.tag or "", + points=account.points, + game_name=account.game_name or "", + game_channel=account.game_channel or "", + gold_balance=account.gold_balance, + exchange_balance=account.exchange_balance, + bind_status=account.bind_status or "", + change_role_wait_time=account.change_role_wait_time, + assigned_to=account.assigned_to, + assigned_username=account.assigned_user.username if account.assigned_user else None, + ) + + +def _task_out(task: DouyuTask) -> DouyuTaskOut: + account = task.account + return DouyuTaskOut( + id=task.id, + batch_id=task.batch_id, + account_id=task.account_id, + account_username=account.username if account else "", + account_uid=account.uid if account else "", + account_nickname=account.nickname if account else "", + task_type=task.task_type, + status=task.status or "", + message=task.message or "", + result=task.result if isinstance(task.result, dict) else None, + created_by=task.created_by, + created_at=task.created_at, + finished_at=task.finished_at, + ) + + +def _config_out(config: DouyuConfig) -> DouyuConfigOut: + return DouyuConfigOut( + manual_id=douyu_config_value("manual_id", config.manual_id), + rid=douyu_config_value("rid", config.rid), + bind_act_alias=douyu_config_value("bind_act_alias", config.bind_act_alias), + confirm_act_alias=douyu_config_value("confirm_act_alias", config.confirm_act_alias), + legacy_act_alias=douyu_config_value("legacy_act_alias", config.legacy_act_alias), + room_id=douyu_config_value("room_id", config.room_id), + elite_amount=douyu_config_value("elite_amount", config.elite_amount), + gold_pay_type=douyu_config_value("gold_pay_type", config.gold_pay_type), + gift_id=douyu_config_value("gift_id", config.gift_id), + skin_id=douyu_config_value("skin_id", config.skin_id), + updated_at=config.updated_at, + ) + + +@router.get("/task-types") +def task_types(current: User = Depends(require_permission("douyu:task"))): + """返回斗鱼任务类型。""" + return SUPPORTED_DOUYU_TASK_TYPES + + +@router.get("/accounts", response_model=list[DouyuTaskAccountOut]) +def list_task_accounts( + search: str = Query(""), + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +): + """查看可执行斗鱼任务的账号（必须有成功 Cookie）。""" + query = _visible_task_accounts_query(db, current) + search_text = (search or "").strip() + if search_text: + pattern = f"%{search_text}%" + query = query.filter(or_( + Account.username.ilike(pattern), + Account.uid.ilike(pattern), + Account.nickname.ilike(pattern), + Account.tag.ilike(pattern), + Account.game_name.ilike(pattern), + )) + rows = query.order_by(Account.id.desc()).limit(500).all() + return [_account_out(account) for account in rows] + + +@router.get("/config", response_model=DouyuConfigOut) +def get_config( + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:config")), +): + """获取斗鱼活动配置。""" + return _config_out(ensure_douyu_config(db)) + + +@router.put("/config", response_model=DouyuConfigOut) +def update_config( + req: DouyuConfigUpdate, + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:config")), +): + """更新斗鱼活动配置。""" + config = ensure_douyu_config(db) + for field in DOUYU_CONFIG_FIELDS: + value = getattr(req, field) + if value is None: + continue + setattr(config, field, value.strip() if isinstance(value, str) else value) + apply_douyu_config_defaults(config) + config.updated_at = datetime.now(timezone.utc) + db.commit() + db.refresh(config) + return _config_out(config) + + +@router.get("/goods", response_model=list[DouyuGoodsOut]) +def list_goods( + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:task")), +): + """查看已缓存的斗鱼商品快照。""" + rows = db.query(DouyuGoodsSnapshot).order_by(DouyuGoodsSnapshot.id.asc()).all() + return rows + + +@router.post("/tasks/batch") +async def create_task_batch( + req: DouyuTaskBatchRequest, + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:task")), +): + """创建斗鱼任务记录并启动后台执行器。""" + if not req.account_ids: + raise HTTPException(status_code=400, detail="请选择斗鱼账号") + + cleanup_orphan_douyu_tasks( + db, + active_batch_ids=douyu_batch_registry.active_ids(), + statuses=("pending", "running"), + message="任务已中断（无执行器接管）", + ) + + try: + batch_id, count = create_douyu_planned_tasks( + db, + req.account_ids, + req.task_type, + current.id, + req.payload, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + if count == 0: + raise HTTPException(status_code=400, detail="没有可执行的斗鱼账号，请先登录获取 Cookie") + + log_queue = asyncio.Queue() + loop = asyncio.get_running_loop() + thread_db = SessionLocal() + runner = DouyuBatchRunner( + db=thread_db, + batch_id=batch_id, + task_type=req.task_type, + payload=req.payload, + log_queue=log_queue, + loop=loop, + concurrency=req.concurrency, + ) + douyu_batch_registry.register(batch_id, log_queue, loop, runner) + + thread = threading.Thread(target=runner.run, daemon=True) + thread.start() + + return {"batch_id": batch_id, "count": count, "success": True} + + +@router.get("/tasks", response_model=list[DouyuTaskOut]) +def list_tasks( + batch_id: str | None = None, + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:task")), +): + """查看斗鱼任务记录。""" + cleanup_orphan_douyu_tasks( + db, + active_batch_ids=douyu_batch_registry.active_ids(), + statuses=("pending", "running"), + message="任务已中断（无执行器接管）", + ) + query = db.query(DouyuTask).options(joinedload(DouyuTask.account)) + if batch_id: + query = query.filter(DouyuTask.batch_id == batch_id) + rows = query.order_by(DouyuTask.id.desc()).limit(300).all() + return [_task_out(task) for task in rows] + + +@router.get("/tasks/{task_id}", response_model=DouyuTaskOut) +def get_task( + task_id: int, + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:task")), +): + """获取单条斗鱼任务详情。""" + task = ( + db.query(DouyuTask) + .options(joinedload(DouyuTask.account)) + .filter(DouyuTask.id == task_id) + .first() + ) + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + return _task_out(task) + + +@router.post("/stop/{batch_id}") +def stop_batch( + batch_id: str, + db: Session = Depends(get_db), + current: User = Depends(require_permission("douyu:task")), +): + """停止正在运行的斗鱼批次。""" + batch = douyu_batch_registry.get(batch_id) + if batch: + if batch.get("finished"): + douyu_batch_registry.pop(batch_id) + cleaned = cleanup_orphan_douyu_tasks(db, batch_id=batch_id, message="批次已结束") + if cleaned: + return {"message": f"批次已结束，已清理 {cleaned} 个残留任务", "success": True} + raise HTTPException(status_code=404, detail="批次已结束") + batch["runner"].stop() + return {"message": "已发送停止信号", "success": True} + + cleaned = cleanup_orphan_douyu_tasks(db, batch_id=batch_id, message="任务已停止（批次不存在）") + if cleaned: + return {"message": f"已清理 {cleaned} 个残留任务", "success": True} + raise HTTPException(status_code=404, detail="批次不存在或已结束") + + +@router.websocket("/ws/{batch_id}") +async def ws_douyu_logs(websocket: WebSocket, batch_id: str): + """斗鱼实时日志推送通道。""" + user = authenticate_websocket(websocket) + if not user: + await websocket.close(code=1008, reason="未授权") + return + await websocket.accept() + + batch = douyu_batch_registry.get(batch_id) + if not batch: + await websocket.send_json({"level": "error", "message": "批次不存在或已结束"}) + await websocket.close() + return + + log_queue: asyncio.Queue = batch["log_queue"] + try: + while True: + try: + msg = await asyncio.wait_for(log_queue.get(), timeout=30) + await websocket.send_json(msg) + if msg.get("level") == "result": + await asyncio.sleep(0.1) + break + except asyncio.TimeoutError: + await websocket.send_json({"level": "heartbeat", "message": ""}) + except WebSocketDisconnect: + pass + finally: + latest = douyu_batch_registry.get(batch_id) + if latest and latest.get("finished"): + douyu_batch_registry.pop(batch_id) diff --git a/web/backend/schemas.py b/web/backend/schemas.py index 2aca665..35d392a 100644 --- a/web/backend/schemas.py +++ b/web/backend/schemas.py @@ -514,6 +514,134 @@ class HuyaRechargeGoodsOut(BaseModel): } +# ---- 斗鱼活动任务 ---- +class DouyuConfigOut(BaseModel): + manual_id: str = "G4KA4Qnz4LDp7" + rid: str = "9263298" + bind_act_alias: str = "20250213NQCYX" + confirm_act_alias: str = "20260120QYOOB" + legacy_act_alias: str = "cjm" + room_id: str = "9263298" + elite_amount: int = 3000 + gold_pay_type: int = 1 + gift_id: str = "23643" + skin_id: str = "2942" + updated_at: Optional[datetime] = None + + @model_serializer + def _serialize(self) -> dict[str, Any]: + return { + "manual_id": self.manual_id, + "rid": self.rid, + "bind_act_alias": self.bind_act_alias, + "confirm_act_alias": self.confirm_act_alias, + "legacy_act_alias": self.legacy_act_alias, + "room_id": self.room_id, + "elite_amount": self.elite_amount, + "gold_pay_type": self.gold_pay_type, + "gift_id": self.gift_id, + "skin_id": self.skin_id, + "updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None, + } + + +class DouyuConfigUpdate(BaseModel): + manual_id: Optional[str] = None + rid: Optional[str] = None + bind_act_alias: Optional[str] = None + confirm_act_alias: Optional[str] = None + legacy_act_alias: Optional[str] = None + room_id: Optional[str] = None + elite_amount: Optional[int] = Field(None, ge=1) + gold_pay_type: Optional[int] = Field(None, ge=1, le=9) + gift_id: Optional[str] = None + skin_id: Optional[str] = None + + +class DouyuTaskBatchRequest(BaseModel): + account_ids: list[int] + task_type: str + concurrency: int = Field(3, ge=1, le=10) + payload: dict[str, Any] = Field(default_factory=dict) + + +class DouyuTaskOut(BaseModel): + id: int + batch_id: str + account_id: int + account_username: str = "" + account_uid: str = "" + account_nickname: str = "" + task_type: str + status: str + message: str = "" + result: Optional[dict[str, Any]] = None + created_by: int + created_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + + model_config = ConfigDict(from_attributes=True) + + @model_serializer + def _serialize(self) -> dict[str, Any]: + return { + "id": self.id, + "batch_id": self.batch_id, + "account_id": self.account_id, + "account_username": self.account_username, + "account_uid": self.account_uid, + "account_nickname": self.account_nickname, + "task_type": self.task_type, + "status": self.status, + "message": self.message, + "result": self.result, + "created_by": self.created_by, + "created_at": _ensure_tz(self.created_at).isoformat() if self.created_at else None, + "finished_at": _ensure_tz(self.finished_at).isoformat() if self.finished_at else None, + } + + +class DouyuGoodsOut(BaseModel): + id: int + commodity_id: str + name: str = "" + score: Optional[int] = None + status: str = "" + raw: Optional[dict[str, Any]] = None + updated_at: Optional[datetime] = None + + model_config = ConfigDict(from_attributes=True) + + @model_serializer + def _serialize(self) -> dict[str, Any]: + return { + "id": self.id, + "commodity_id": self.commodity_id, + "name": self.name, + "score": self.score, + "status": self.status, + "raw": self.raw, + "updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None, + } + + +class DouyuTaskAccountOut(BaseModel): + id: int + username: str + uid: str = "" + nickname: str = "" + tag: str = "" + points: Optional[int] = None + game_name: str = "" + game_channel: str = "" + gold_balance: Optional[int] = None + exchange_balance: Optional[int] = None + bind_status: str = "" + change_role_wait_time: Optional[int] = None + assigned_to: Optional[int] = None + assigned_username: Optional[str] = None + + # ---- 代理配置 ---- class ProxyConfigOut(BaseModel): enabled: bool = False diff --git a/web/backend/services/douyu_runner.py b/web/backend/services/douyu_runner.py new file mode 100644 index 0000000..881a5a7 --- /dev/null +++ b/web/backend/services/douyu_runner.py @@ -0,0 +1,490 @@ +"""斗鱼活动任务批次执行器。""" + +from __future__ import annotations + +import asyncio +import threading +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +from typing import Optional + +from loguru import logger +from sqlalchemy.orm import Session, joinedload + +from core.douyu import DouyuActivityClient, DouyuActivityError + +from ..database import SessionLocal +from ..models import Account, DouyuGoodsSnapshot, DouyuTask +from .douyu_service import ( + DOUYU_CONFIG_FIELDS, + account_uid, + douyu_config_value, + ensure_douyu_config, + latest_success_cookie, + update_account_profile_from_cookie, +) + + +class DouyuBatchRunner: + """批量执行斗鱼活动任务，通过队列推送实时日志。""" + + def __init__( + self, + db: Session, + batch_id: str, + task_type: str, + payload: Optional[dict] = None, + log_queue: Optional[asyncio.Queue] = None, + loop: Optional[asyncio.AbstractEventLoop] = None, + concurrency: int = 3, + ): + self.db = db + self.batch_id = batch_id + self.task_type = task_type + self.payload = payload or {} + self.log_queue = log_queue + self.loop = loop + self.concurrency = max(1, min(concurrency, 10)) + self._stop = threading.Event() + self._counter_lock = threading.Lock() + self._started = 0 + + def stop(self): + self._stop.set() + + def _push_log(self, level: str, message: str): + if level == "result": + try: + douyu_batch_registry.mark_finished(self.batch_id) + except NameError: + pass + if level != "result" and message: + log_func = getattr(logger, level, logger.info) + log_func(f"[douyu] {message}") + if self.log_queue and self.loop: + asyncio.run_coroutine_threadsafe( + self.log_queue.put({"level": level, "message": message}), + self.loop, + ) + + @staticmethod + def _account_name(account: Account) -> str: + return account.nickname or account.username or account.uid or f"#{account.id}" + + @staticmethod + def _to_int(value) -> int | None: + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _format_wait_time(seconds: int | None) -> str: + if seconds is None: + return "" + seconds = max(0, int(seconds)) + days, rem = divmod(seconds, 86400) + hours, rem = divmod(rem, 3600) + minutes, sec = divmod(rem, 60) + if days: + return f"{days}天{hours}小时{minutes}分" + if hours: + return f"{hours}小时{minutes}分{sec}秒" + return f"{minutes}分{sec}秒" + + def _mark_task( + self, + db: Session, + task: DouyuTask, + status: str, + message: str, + result: dict | None = None, + ) -> None: + task.status = status + task.message = message[:512] + if result is not None: + task.result = result + task.finished_at = datetime.now(timezone.utc) + db.commit() + + def _upsert_goods(self, db: Session, goods: list[dict]) -> None: + now = datetime.now(timezone.utc) + for raw in goods: + commodity_id = str(raw.get("commodityId") or raw.get("commodity_id") or "") + if not commodity_id: + continue + row = ( + db.query(DouyuGoodsSnapshot) + .filter(DouyuGoodsSnapshot.commodity_id == commodity_id) + .first() + ) + score = self._to_int(raw.get("score")) + if row is None: + row = DouyuGoodsSnapshot(commodity_id=commodity_id) + db.add(row) + row.name = str(raw.get("commodityName") or raw.get("name") or "") + row.score = score + row.status = str(raw.get("status") or "") + row.raw = raw + row.updated_at = now + db.commit() + + def _config_info(self, db: Session) -> dict: + config = ensure_douyu_config(db) + return {field: douyu_config_value(field, getattr(config, field, None)) for field in DOUYU_CONFIG_FIELDS} + + def _task_payload(self, task: DouyuTask) -> dict: + result = task.result if isinstance(task.result, dict) else {} + payload = result.get("payload") if isinstance(result.get("payload"), dict) else {} + return {**payload, **self.payload} + + def _client(self, cookie: str) -> DouyuActivityClient: + return DouyuActivityClient(cookie, logger=lambda msg: self._push_log("debug", msg)) + + def _execute_refresh_goods(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + result = client.list_goods(manual_id=config["manual_id"], rid=config["rid"]) + goods = result["goods"] + self._upsert_goods(db, goods) + account.bind_status = account.bind_status or "active" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", f"已刷新商品 {len(goods)} 个", {"goods_count": len(goods), "goods": goods}) + + def _execute_get_bind_qr(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + result = client.get_bind_qr(str(config["bind_act_alias"])) + account.bind_status = "bind_qr_generated" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", "绑定二维码已生成", result) + + def _execute_confirm_bind(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + before = client.bind_info(str(config["legacy_act_alias"]), v2=True) + result = client.confirm_bind(str(config["confirm_act_alias"])) + after = client.bind_info(str(config["confirm_act_alias"]), v2=False) + role_name = after.get("role_name") or before.get("role_name") or "" + account.game_name = role_name or account.game_name + account.game_channel = " / ".join(part for part in [after.get("area_name"), after.get("plat_name")] if part) + account.bind_status = "bind_confirmed" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", f"绑定成功: {role_name or '已确认'}", {"before": before, "confirm": result, "bind_info": after}) + + 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) + result = client.create_elite_qr( + ctn=ctn, + act_alias=str(config["confirm_act_alias"]), + amount=int(config["elite_amount"]), + room_id=str(config["room_id"]), + ) + account.bind_status = "elite_qr_created" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", "精英宝典支付码已生成", result) + + def _execute_create_gold_qr(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + payload = self._task_payload(task) + amount = int(payload.get("amount") or payload.get("gold_amount") or 1) + client = self._client(cookie) + result = client.create_gold_qr(amount=amount, pay_type=int(config["gold_pay_type"])) + account.bind_status = "gold_qr_created" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", f"鱼翅 {amount} 元支付码已生成", result) + + def _execute_donate_elite_gift(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + payload = self._task_payload(task) + gift_count = int(payload.get("gift_count") or payload.get("count") or 1) + client = self._client(cookie) + result = client.donate_elite_gift( + gift_count=gift_count, + room_id=str(payload.get("room_id") or config["room_id"]), + gift_id=str(payload.get("gift_id") or config["gift_id"]), + skin_id=str(payload.get("skin_id") or config["skin_id"]), + ) + account.bind_status = "gift_donated" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", f"赠送精英令成功: {gift_count}", 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) + uid = account_uid(account, cookie) + if not uid: + self._mark_task(db, task, "failed", "Cookie 中没有 acf_uid，无法查询积分") + return + result = client.query_points(uid=uid, ctn=ctn) + points = self._to_int(result.get("points")) + account.uid = uid + account.points = points + update_account_profile_from_cookie(account, cookie) + account.bind_status = "points_queried" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", f"积分: {points if points is not None else '-'}", result) + + def _execute_exchange_goods(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + import time as time_mod + payload = self._task_payload(task) + commodity_id = str(payload.get("commodity_id") or payload.get("commodityId") or "").strip() + if not commodity_id: + self._mark_task(db, task, "failed", "请选择兑换商品") + return + client = self._client(cookie) + ctn = client.acf_ccn(refresh_subscribe=False) + result = None + last_error = "" + for attempt in range(8 + 1): + if self._stop.is_set(): + self._mark_task(db, task, "stopped", "任务已停止") + return + try: + result = client.exchange_goods( + manual_id=str(config["manual_id"]), + rid=str(config["rid"]), + commodity_id=commodity_id, + ctn=ctn, + ) + break + except DouyuActivityError as exc: + last_error = str(exc) + if attempt >= 8: + self._mark_task(db, task, "failed", f"兑换失败(已重试{attempt}次): {last_error}") + return + error_lower = last_error.lower() + if any(kw in error_lower for kw in ("无效", "太快", "csrf")): + self._push_log("info", f" 重试 {attempt + 1}/8: {last_error}，刷新 csrf_token...") + try: + token = client.csrf_token() + self._push_log("debug", f" csrf_token 已刷新: {token[:12]}...") + except Exception: + pass + else: + self._push_log("info", f" 重试 {attempt + 1}/8: {last_error}") + time_mod.sleep(0.3) + if result is None: + self._mark_task(db, task, "failed", f"兑换失败: {last_error}") + return + goods = ( + db.query(DouyuGoodsSnapshot) + .filter(DouyuGoodsSnapshot.commodity_id == commodity_id) + .first() + ) + account.bind_status = "goods_exchanged" + account.updated_at = datetime.now(timezone.utc) + self._mark_task( + db, + task, + "success", + f"兑换成功: {(goods.name if goods else '') or commodity_id}", + {"goods": goods.raw if goods else None, **result}, + ) + + def _execute_query_game_name(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + result = client.bind_info(str(config["confirm_act_alias"]), v2=False) + if not result.get("role_name"): + result = client.bind_info(str(config["legacy_act_alias"]), v2=True) + role_name = result.get("role_name") or "" + account.game_name = role_name + account.game_channel = " / ".join(part for part in [result.get("area_name"), result.get("plat_name")] if part) + account.bind_status = "game_queried" if role_name else "game_not_bound" + account.change_role_wait_time = self._to_int(result.get("change_role_wait_time")) + account.updated_at = datetime.now(timezone.utc) + message = f"游戏名: {role_name}" if role_name else "未获取到游戏名" + self._mark_task(db, task, "success" if role_name else "failed", message, result) + + def _execute_query_change_bind_time(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + result = client.bind_info(str(config["confirm_act_alias"]), v2=True) + wait_time = self._to_int(result.get("change_role_wait_time")) + account.change_role_wait_time = wait_time + account.bind_status = "change_time_queried" + account.updated_at = datetime.now(timezone.utc) + result["change_role_wait_text"] = self._format_wait_time(wait_time) + self._mark_task(db, task, "success", f"换绑剩余: {result['change_role_wait_text'] or '-'}", result) + + def _execute_query_limited_goods(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + result = client.query_limited_goods(manual_id=str(config["manual_id"]), rid=str(config["rid"])) + limited = result["limited_goods"] + names = [str(item.get("commodityName") or "") for item in limited if item.get("commodityName")] + message = "无限制商品" if not names else f"限兑 {len(names)} 个: {', '.join(names[:5])}" + account.bind_status = "limited_goods_queried" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", message, {"limited_count": len(limited), "limited_goods": limited}) + + def _execute_query_gold_balance(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + gold = client.gold_account() + exchange = client.exchange_balance() + account.gold_balance = self._to_int(gold.get("gold")) + account.exchange_balance = self._to_int(exchange.get("count")) + account.bind_status = "gold_balance_queried" + account.updated_at = datetime.now(timezone.utc) + self._mark_task( + db, + task, + "success", + f"鱼翅余额: {account.gold_balance if account.gold_balance is not None else '-'}", + {"gold": gold, "exchange_balance": exchange}, + ) + + def _execute_query_exchange_records(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + result = client.exchange_records(manual_id=str(config["manual_id"])) + records = result["records"] + account.bind_status = "exchange_records_queried" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", f"兑换记录 {len(records)} 条" if records else "暂无兑换记录", result) + + def _execute_prefetch_csrf_token(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + client = self._client(cookie) + token = client.csrf_token() + account.bind_status = "csrf_token_ready" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(db, task, "success", "获取 csrf_token 成功", {"csrf_token": token, "cookie": client.cookie}) + + def _execute_one(self, task_id: int, config: dict, total: int): + worker_db = SessionLocal() + try: + task = ( + worker_db.query(DouyuTask) + .options(joinedload(DouyuTask.account)) + .filter(DouyuTask.id == task_id) + .first() + ) + if not task or self._stop.is_set(): + return + account = task.account + task.status = "running" + task.message = "执行中" + worker_db.commit() + + with self._counter_lock: + self._started += 1 + current = self._started + + self._push_log("info", f"[{current}/{total}] 开始: {self._account_name(account)}") + cookie = latest_success_cookie(worker_db, account.id) + if not cookie: + self._mark_task(worker_db, task, "failed", "账号没有成功登录 Cookie") + self._push_log("warning", f"[{current}] {self._account_name(account)} 无 Cookie") + return + update_account_profile_from_cookie(account, cookie) + + handler = { + "refresh_goods": self._execute_refresh_goods, + "get_bind_qr": self._execute_get_bind_qr, + "confirm_bind": self._execute_confirm_bind, + "create_elite_qr": self._execute_create_elite_qr, + "create_gold_qr": self._execute_create_gold_qr, + "donate_elite_gift": self._execute_donate_elite_gift, + "query_points": self._execute_query_points, + "exchange_goods": self._execute_exchange_goods, + "query_game_name": self._execute_query_game_name, + "query_change_bind_time": self._execute_query_change_bind_time, + "query_limited_goods": self._execute_query_limited_goods, + "query_gold_balance": self._execute_query_gold_balance, + "query_exchange_records": self._execute_query_exchange_records, + "prefetch_csrf_token": self._execute_prefetch_csrf_token, + }.get(task.task_type) + if handler is None: + self._mark_task(worker_db, task, "failed", "不支持的任务类型") + return + + handler(worker_db, task, account, cookie, config) + self._push_log("success", f"[{current}] {self._account_name(account)} {task.message}") + except DouyuActivityError as exc: + if "task" in locals() and task: + self._mark_task(worker_db, task, "failed", str(exc)) + self._push_log("warning", f"斗鱼任务失败: {exc}") + except Exception as exc: + if "task" in locals() and task: + self._mark_task(worker_db, task, "error", str(exc)) + self._push_log("error", f"斗鱼任务异常: {exc}") + finally: + worker_db.close() + + def run(self): + """执行批次任务。""" + self._push_log("info", f"斗鱼任务批次 {self.batch_id} 开始") + try: + config = self._config_info(self.db) + tasks = ( + self.db.query(DouyuTask) + .filter(DouyuTask.batch_id == self.batch_id, DouyuTask.status == "planned") + .order_by(DouyuTask.id.asc()) + .all() + ) + if not tasks: + self._push_log("warning", "没有可执行的斗鱼任务") + self._push_log("result", "") + return + + for task in tasks: + task.status = "pending" + task.message = "等待执行" + self.db.commit() + + total = len(tasks) + with ThreadPoolExecutor(max_workers=self.concurrency) as executor: + futures = [] + for task in tasks: + if self._stop.is_set(): + break + futures.append(executor.submit(self._execute_one, task.id, config, total)) + for future in as_completed(futures): + try: + future.result() + except Exception as exc: + self._push_log("error", f"Worker 异常: {exc}") + + if self._stop.is_set(): + self._push_log("warning", f"斗鱼任务批次 {self.batch_id} 已停止") + else: + self._push_log("info", f"斗鱼任务批次 {self.batch_id} 完成") + self._push_log("result", "") + finally: + self.db.close() + + +class DouyuBatchRegistry: + """管理运行中的斗鱼任务批次。""" + + def __init__(self): + self._batches: dict[str, dict] = {} + + def register(self, batch_id: str, log_queue: asyncio.Queue, + loop: asyncio.AbstractEventLoop, runner: DouyuBatchRunner): + self._batches[batch_id] = { + "log_queue": log_queue, + "loop": loop, + "runner": runner, + "finished": False, + "updated_at": time.time(), + } + + def get(self, batch_id: str): + return self._batches.get(batch_id) + + def pop(self, batch_id: str): + return self._batches.pop(batch_id, None) + + def mark_finished(self, batch_id: str): + if batch_id in self._batches: + self._batches[batch_id]["finished"] = True + self._batches[batch_id]["updated_at"] = time.time() + + def active_ids(self) -> set[str]: + return { + batch_id + for batch_id, info in self._batches.items() + if not info.get("finished") + } + + +douyu_batch_registry = DouyuBatchRegistry() diff --git a/web/backend/services/douyu_service.py b/web/backend/services/douyu_service.py new file mode 100644 index 0000000..769524c --- /dev/null +++ b/web/backend/services/douyu_service.py @@ -0,0 +1,195 @@ +"""斗鱼活动任务服务。""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from core.douyu.activity_client import DouyuActivityClient +from core.douyu.cookie_utils import cookie_value + +from ..models import Account, DouyuConfig, DouyuTask, LoginTask + + +SUPPORTED_DOUYU_TASK_TYPES = { + "get_bind_qr": "获取绑定二维码", + "confirm_bind": "确认绑定", + "create_elite_qr": "开通精英宝典30", + "create_gold_qr": "充值鱼翅", + "donate_elite_gift": "赠送精英令", + "query_points": "一键查询积分", + "exchange_goods": "兑换商品", + "query_game_name": "一键获取游戏名", + "query_change_bind_time": "一键查询换绑时间", + "query_limited_goods": "一键查询限兑商品", + "query_gold_balance": "一键查询鱼刺余额", + "refresh_goods": "刷新商品列表", + "query_exchange_records": "一键查询兑换记录", + "prefetch_csrf_token": "一键获取兑换 CSRF Token", +} + + +DOUYU_CONFIG_DEFAULTS = { + "manual_id": "G4KA4Qnz4LDp7", + "rid": "9263298", + "bind_act_alias": "20250213NQCYX", + "confirm_act_alias": "20260120QYOOB", + "legacy_act_alias": "cjm", + "room_id": "9263298", + "elite_amount": 3000, + "gold_pay_type": 1, + "gift_id": "23643", + "skin_id": "2942", +} + +DOUYU_CONFIG_FIELDS = tuple(DOUYU_CONFIG_DEFAULTS.keys()) +DOUYU_ACTIVE_TASK_STATUSES = ("planned", "pending", "running") + + +def douyu_config_value(field: str, value): + """读取配置值；空值自动回退到默认值。""" + default = DOUYU_CONFIG_DEFAULTS[field] + if isinstance(default, int): + try: + return int(value if value is not None else default) + except (TypeError, ValueError): + return default + text = str(value or "").strip() + return text or str(default) + + +def apply_douyu_config_defaults(config: DouyuConfig) -> bool: + """补齐斗鱼配置默认值，返回是否发生变更。""" + changed = False + for field in DOUYU_CONFIG_FIELDS: + normalized = douyu_config_value(field, getattr(config, field, None)) + if getattr(config, field, None) != normalized: + setattr(config, field, normalized) + changed = True + return changed + + +def ensure_douyu_config(db: Session) -> DouyuConfig: + """获取单条斗鱼配置，不存在则创建。""" + config = db.query(DouyuConfig).first() + if config: + if apply_douyu_config_defaults(config): + db.commit() + db.refresh(config) + return config + config = DouyuConfig(**DOUYU_CONFIG_DEFAULTS) + db.add(config) + db.commit() + db.refresh(config) + return config + + +def latest_success_cookie(db: Session, account_id: int) -> str: + """读取账号最近一次成功登录 Cookie。""" + task = ( + db.query(LoginTask) + .filter( + LoginTask.account_id == account_id, + LoginTask.status == "success", + LoginTask.cookie != "", + ) + .order_by(LoginTask.id.desc()) + .first() + ) + return task.cookie if task else "" + + +def cookie_account_ids_query(db: Session): + """返回拥有成功 Cookie 的斗鱼账号 ID 查询。""" + return ( + db.query(LoginTask.account_id) + .filter(LoginTask.status == "success", LoginTask.cookie != "") + .distinct() + ) + + +def visible_douyu_task_accounts(db: Session, account_ids: list[int]) -> list[Account]: + """只保留存在成功 Cookie 的斗鱼账号。""" + if not account_ids: + return [] + cookie_ids = cookie_account_ids_query(db).subquery() + return ( + db.query(Account) + .filter(Account.id.in_(account_ids), Account.id.in_(cookie_ids)) + .all() + ) + + +def update_account_profile_from_cookie(account: Account, cookie: str) -> None: + """从 Cookie 回填 uid/nickname。""" + profile = DouyuActivityClient.profile_from_cookie(cookie) + if profile["uid"]: + account.uid = profile["uid"] + if profile["nickname"]: + account.nickname = profile["nickname"] + + +def account_uid(account: Account, cookie: str) -> str: + """优先从账号字段读取 uid，缺失时从 Cookie 中取。""" + return account.uid or cookie_value(cookie, "acf_uid") + + +def create_douyu_planned_tasks( + db: Session, + account_ids: list[int], + task_type: str, + created_by: int, + payload: dict | None = None, +) -> tuple[str, int]: + """创建斗鱼任务记录，等待后台执行器消费。""" + if task_type not in SUPPORTED_DOUYU_TASK_TYPES: + raise ValueError("不支持的任务类型") + + accounts = visible_douyu_task_accounts(db, account_ids) + if task_type == "refresh_goods" and accounts: + # 商品快照是全局数据，一个可用 CK 足够；没有 CK 时前端无法选账号创建任务。 + accounts = accounts[:1] + + batch_id = uuid.uuid4().hex[:12] + payload = payload or {} + for account in accounts: + db.add(DouyuTask( + batch_id=batch_id, + account_id=account.id, + task_type=task_type, + status="planned", + message="任务已创建，等待执行", + result={"payload": payload} if payload else None, + created_by=created_by, + )) + db.commit() + return batch_id, len(accounts) + + +def cleanup_orphan_douyu_tasks( + db: Session, + *, + active_batch_ids: set[str] | None = None, + batch_id: str | None = None, + statuses: tuple[str, ...] = DOUYU_ACTIVE_TASK_STATUSES, + message: str = "任务已中断（服务重启或批次丢失）", +) -> int: + """清理没有执行器接管的斗鱼任务。""" + query = db.query(DouyuTask).filter(DouyuTask.status.in_(statuses)) + if batch_id: + query = query.filter(DouyuTask.batch_id == batch_id) + elif active_batch_ids is not None and active_batch_ids: + query = query.filter(~DouyuTask.batch_id.in_(list(active_batch_ids))) + + tasks = query.all() + if not tasks: + return 0 + now = datetime.now(timezone.utc) + for task in tasks: + task.status = "stopped" + task.message = message + task.finished_at = now + db.commit() + return len(tasks) diff --git a/web/frontend/src/App.tsx b/web/frontend/src/App.tsx index 0ce93fa..ced6cb1 100644 --- a/web/frontend/src/App.tsx +++ b/web/frontend/src/App.tsx @@ -17,6 +17,7 @@ const LoginTasksPage = lazy(() => import('./pages/LoginTasksPage')); const ProxyPage = lazy(() => import('./pages/ProxyPage')); const UsersPage = lazy(() => import('./pages/UsersPage')); const CookiePage = lazy(() => import('./pages/CookiePage')); +const DouyuTasksPage = lazy(() => import('./pages/DouyuTasksPage')); const HuyaAccountsPage = lazy(() => import('./pages/HuyaAccountsPage')); const HuyaAssignmentsPage = lazy(() => import('./pages/HuyaAssignmentsPage')); const HuyaCookiePage = lazy(() => import('./pages/HuyaCookiePage')); @@ -71,6 +72,7 @@ function AppContent() { )} /> )} /> )} /> + )} /> )} /> )} /> )} /> diff --git a/web/frontend/src/api/douyu.ts b/web/frontend/src/api/douyu.ts new file mode 100644 index 0000000..52653a0 --- /dev/null +++ b/web/frontend/src/api/douyu.ts @@ -0,0 +1,25 @@ +import api from './client'; +import type { + DouyuConfig, + DouyuGoodsItem, + DouyuTaskAccountItem, + DouyuTaskBatchRequest, + DouyuTaskBatchResult, + DouyuTaskItem, + MessageResponse, +} from './types'; + +export const douyuApi = { + taskTypes: () => api.get, Record>('/douyu/task-types'), + listAccounts: (params?: { search?: string }) => + api.get('/douyu/accounts', { params }), + getConfig: () => api.get('/douyu/config'), + updateConfig: (data: Partial) => api.put('/douyu/config', data), + listGoods: () => api.get('/douyu/goods'), + createTasks: (data: DouyuTaskBatchRequest) => + api.post('/douyu/tasks/batch', data), + listTasks: (batchId?: string) => + api.get('/douyu/tasks', { params: batchId ? { batch_id: batchId } : {} }), + getTask: (taskId: number) => api.get(`/douyu/tasks/${taskId}`), + stopBatch: (batchId: string) => api.post(`/douyu/stop/${batchId}`), +}; diff --git a/web/frontend/src/api/modules.ts b/web/frontend/src/api/modules.ts index 6a7c637..393bd35 100644 --- a/web/frontend/src/api/modules.ts +++ b/web/frontend/src/api/modules.ts @@ -4,6 +4,7 @@ export { accountApi } from './accounts'; export { appApi } from './app'; export { authApi } from './auth'; export { cookieApi } from './cookies'; +export { douyuApi } from './douyu'; export { huyaApi } from './huya'; export { loginApi } from './login'; export { proxyApi } from './proxy'; diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index 442ac10..4482ae4 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -168,6 +168,78 @@ export interface CookieItem { account_password: string; } +// ==================== Douyu Activity ==================== + +export interface DouyuTaskAccountItem { + id: number; + username: string; + uid: string; + nickname: string; + tag: string; + points: number | null; + game_name: string; + game_channel: string; + gold_balance: number | null; + exchange_balance: number | null; + bind_status: string; + change_role_wait_time: number | null; + assigned_to: number | null; + assigned_username: string | null; +} + +export interface DouyuConfig { + manual_id: string; + rid: string; + bind_act_alias: string; + confirm_act_alias: string; + legacy_act_alias: string; + room_id: string; + elite_amount: number; + gold_pay_type: number; + gift_id: string; + skin_id: string; + updated_at: string | null; +} + +export interface DouyuTaskBatchRequest { + account_ids: number[]; + task_type: string; + concurrency?: number; + payload?: Record; +} + +export interface DouyuTaskBatchResult { + batch_id: string; + count: number; + success: boolean; +} + +export interface DouyuTaskItem { + id: number; + batch_id: string; + account_id: number; + account_username: string; + account_uid: string; + account_nickname: string; + task_type: string; + status: string; + message: string; + result: Record | null; + created_by: number; + created_at: string | null; + finished_at: string | null; +} + +export interface DouyuGoodsItem { + id: number; + commodity_id: string; + name: string; + score: number | null; + status: string; + raw: Record | null; + updated_at: string | null; +} + // ==================== Huya ==================== export interface HuyaAccountItem { diff --git a/web/frontend/src/layouts/MainLayout.tsx b/web/frontend/src/layouts/MainLayout.tsx index 9572f36..be268fa 100644 --- a/web/frontend/src/layouts/MainLayout.tsx +++ b/web/frontend/src/layouts/MainLayout.tsx @@ -70,6 +70,9 @@ export default function MainLayout({ onLogout }: { onLogout?: () => void }) { if (can('cookie:view')) { douyuItems.push({ key: '/cookies', label: 'Cookie 管理', icon: }); } + if (can('douyu:task')) { + douyuItems.push({ key: '/douyu/tasks', label: '任务操作台', icon: }); + } // 虎牙 if (canAny(['huya:account', 'huya:view_all', 'huya:view_assigned'])) { diff --git a/web/frontend/src/pages/DouyuTasksPage.tsx b/web/frontend/src/pages/DouyuTasksPage.tsx new file mode 100644 index 0000000..fc97157 --- /dev/null +++ b/web/frontend/src/pages/DouyuTasksPage.tsx @@ -0,0 +1,516 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import { + Button, Card, Col, Form, Input, InputNumber, Modal, QRCode, Row, Select, Space, Table, Tag, Typography, theme, +} from 'antd'; +import type { TableProps } from 'antd'; +import { + CheckCircleOutlined, CreditCardOutlined, FieldTimeOutlined, GiftOutlined, LinkOutlined, + QrcodeOutlined, ReloadOutlined, SearchOutlined, SettingOutlined, ShoppingOutlined, StopOutlined, +} from '@ant-design/icons'; +import { + douyuApi, + type DouyuConfig, + type DouyuGoodsItem, + type DouyuTaskAccountItem, + type DouyuTaskItem, +} from '../api/modules'; +import RealtimeLogPanel from '../components/RealtimeLogPanel'; +import { useWebSocketLogs } from '../hooks/useWebSocketLogs'; +import { formatTime } from '../utils/time'; +import { getErrorMessage } from '../utils/error'; +import { message } from '../utils/antdMessage'; + +const { Text } = Typography; + +const FALLBACK_TASK_TYPES: Record = { + get_bind_qr: '获取绑定二维码', + confirm_bind: '确认绑定', + create_elite_qr: '开通精英宝典30', + create_gold_qr: '充值鱼翅', + donate_elite_gift: '赠送精英令', + query_points: '一键查询积分', + exchange_goods: '兑换商品', + query_game_name: '一键获取游戏名', + query_change_bind_time: '一键查询换绑时间', + query_limited_goods: '一键查询限兑商品', + query_gold_balance: '一键查询鱼刺余额', + refresh_goods: '刷新商品列表', + query_exchange_records: '一键查询兑换记录', + prefetch_csrf_token: '一键获取兑换 CSRF Token', +}; + +const TASK_ICONS: Record = { + get_bind_qr: , + confirm_bind: , + create_elite_qr: , + create_gold_qr: , + donate_elite_gift: , + query_points: , + exchange_goods: , + query_game_name: , + query_change_bind_time: , + query_limited_goods: , + query_gold_balance: , + refresh_goods: , + query_exchange_records: , + prefetch_csrf_token: , +}; + +const STATUS_COLORS: Record = { + planned: 'default', + pending: 'default', + running: 'processing', + success: 'success', + failed: 'error', + error: 'error', + stopped: 'warning', +}; + +const STATUS_LABELS: Record = { + planned: '已计划', + pending: '等待中', + running: '执行中', + success: '成功', + failed: '失败', + error: '异常', + stopped: '已停止', +}; + +function accountLabel(account: DouyuTaskAccountItem): string { + const name = account.nickname || account.username || account.uid || `#${account.id}`; + const tag = account.tag ? ` [${account.tag}]` : ''; + const game = account.game_name ? ` / ${account.game_name}` : ''; + return `${name}${tag}${game}`; +} + +function resultText(result: Record | null | undefined, key: string): string { + const value = result?.[key]; + return typeof value === 'string' ? value : ''; +} + +function resultNumber(result: Record | null | undefined, key: string): number | null { + const value = result?.[key]; + if (typeof value === 'number') return value; + if (typeof value === 'string' && value.trim()) return Number(value) || null; + return null; +} + +function taskPayUrl(task: DouyuTaskItem | null): string { + return resultText(task?.result, 'pay_url') || resultText(task?.result, 'url'); +} + +function goodsLabel(item: DouyuGoodsItem): string { + return `${item.name || item.commodity_id}${item.score ? ` / ${item.score}积分` : ''}`; +} + +const defaultConfig: DouyuConfig = { + manual_id: 'G4KA4Qnz4LDp7', + rid: '9263298', + bind_act_alias: '20250213NQCYX', + confirm_act_alias: '20260120QYOOB', + legacy_act_alias: 'cjm', + room_id: '9263298', + elite_amount: 3000, + gold_pay_type: 1, + gift_id: '23643', + skin_id: '2942', + updated_at: null, +}; + +export default function DouyuTasksPage() { + const { token } = theme.useToken(); + const [accounts, setAccounts] = useState([]); + const [goods, setGoods] = useState([]); + const [tasks, setTasks] = useState([]); + const [taskTypes, setTaskTypes] = useState>(FALLBACK_TASK_TYPES); + const [config, setConfig] = useState(defaultConfig); + const [selectedAccountIds, setSelectedAccountIds] = useState([]); + const [selectedGoodsId, setSelectedGoodsId] = useState(''); + const [concurrency, setConcurrency] = useState(3); + const [goldAmount, setGoldAmount] = useState(1); + const [giftCount, setGiftCount] = useState(1); + const [loading, setLoading] = useState(false); + const [runningBatchId, setRunningBatchId] = useState(''); + const [configOpen, setConfigOpen] = useState(false); + const [payTask, setPayTask] = useState(null); + const [configForm] = Form.useForm(); + const logs = useWebSocketLogs(); + + const loadData = useCallback(async () => { + setLoading(true); + try { + const [accountResult, goodsResult, taskResult, configResult, taskTypeResult] = await Promise.allSettled([ + douyuApi.listAccounts(), + douyuApi.listGoods(), + douyuApi.listTasks(), + douyuApi.getConfig(), + douyuApi.taskTypes(), + ]); + if (accountResult.status === 'fulfilled') setAccounts(accountResult.value); + if (goodsResult.status === 'fulfilled') setGoods(goodsResult.value); + if (taskResult.status === 'fulfilled') setTasks(taskResult.value); + if (configResult.status === 'fulfilled') setConfig(configResult.value); + if (taskTypeResult.status === 'fulfilled') setTaskTypes(taskTypeResult.value); + const errors = [ + accountResult.status === 'rejected' ? `账号: ${getErrorMessage(accountResult.reason)}` : '', + goodsResult.status === 'rejected' ? `商品: ${getErrorMessage(goodsResult.reason)}` : '', + taskResult.status === 'rejected' ? `任务: ${getErrorMessage(taskResult.reason)}` : '', + configResult.status === 'rejected' ? `配置: ${getErrorMessage(configResult.reason)}` : '', + ].filter(Boolean); + if (errors.length) message.warning(errors.join('；')); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + loadData(); + }, [loadData]); + + useEffect(() => { + if (configOpen) configForm.setFieldsValue(config); + }, [configOpen, config, configForm]); + + const activeTask = tasks.find((item) => ['planned', 'pending', 'running'].includes(item.status)); + + const goodsOptions = useMemo( + () => goods.map((item) => ({ value: item.commodity_id, label: goodsLabel(item) })), + [goods], + ); + + const startTask = async (taskType: string) => { + if (selectedAccountIds.length === 0) { + message.warning('请先选择斗鱼账号'); + return; + } + if (taskType === 'exchange_goods' && !selectedGoodsId) { + message.warning('请先选择兑换商品'); + return; + } + const payload: Record = {}; + if (taskType === 'exchange_goods') { + payload.commodity_id = selectedGoodsId; + } + if (taskType === 'create_gold_qr') { + payload.amount = goldAmount; + } + if (taskType === 'donate_elite_gift') { + payload.gift_count = giftCount; + } + + try { + const result = await douyuApi.createTasks({ + account_ids: selectedAccountIds, + task_type: taskType, + concurrency, + payload, + }); + setRunningBatchId(result.batch_id); + logs.connect(`/api/douyu/ws/${result.batch_id}`, { + clear: true, + onResult: () => { + setRunningBatchId(''); + loadData(); + }, + }); + message.success(`已创建 ${result.count} 个任务`); + setTimeout(loadData, 500); + } catch (error) { + message.error(getErrorMessage(error)); + } + }; + + const stopTask = async () => { + if (!runningBatchId) return; + try { + await douyuApi.stopBatch(runningBatchId); + message.success('已发送停止信号'); + } catch (error) { + message.error(getErrorMessage(error)); + } + }; + + const saveConfig = async () => { + try { + const values = await configForm.validateFields(); + const saved = await douyuApi.updateConfig(values); + setConfig(saved); + setConfigOpen(false); + message.success('配置已保存'); + } catch (error) { + message.error(getErrorMessage(error)); + } + }; + + const openPayTask = (task: DouyuTaskItem) => { + const url = taskPayUrl(task); + if (!url) { + message.warning('任务结果中没有二维码链接'); + return; + } + setPayTask(task); + }; + + const accountColumns: TableProps['columns'] = [ + { + title: '账号', + dataIndex: 'username', + width: 190, + render: (_, record) => ( + + {record.nickname || record.username} + UID {record.uid || '-'} + + ), + }, + { + title: '游戏名', + dataIndex: 'game_name', + width: 180, + render: (_, record) => record.game_name ? ( + + {record.game_name} + {record.game_channel || '-'} + + ) : 未查, + }, + { + title: '积分', + dataIndex: 'points', + width: 90, + render: (value) => value ?? -, + sorter: (a, b) => (a.points ?? -1) - (b.points ?? -1), + }, + { + title: '鱼翅', + dataIndex: 'gold_balance', + width: 90, + render: (value) => value ?? -, + }, + { + title: '状态', + dataIndex: 'bind_status', + width: 120, + render: (value) => value ? {value} : -, + }, + ]; + + const taskColumns: TableProps['columns'] = [ + { + title: '账号', + dataIndex: 'account_username', + width: 150, + render: (_, record) => record.account_nickname || record.account_username || record.account_uid, + }, + { + title: '任务', + dataIndex: 'task_type', + width: 170, + render: (value) => taskTypes[value] || value, + }, + { + title: '状态', + dataIndex: 'status', + width: 90, + render: (value) => {STATUS_LABELS[value] || value}, + }, + { + title: '结果', + dataIndex: 'message', + ellipsis: true, + render: (_, record) => { + const payUrl = taskPayUrl(record); + const points = resultNumber(record.result, 'points'); + return ( + + {record.message || '-'} + {typeof points === 'number' && 积分 {points}} + {payUrl && ( + + )} + + ); + }, + }, + { + title: '时间', + dataIndex: 'finished_at', + width: 170, + render: (_, record) => formatTime(record.finished_at || record.created_at), + }, + ]; + + return ( +
+ +
+

斗鱼任务操作台

+ 绑定、宝典、鱼翅、积分、兑换和余额任务 +
+ + + + {runningBatchId && ( + + )} + + + + + + 已选 {selectedAccountIds.length}} + > + setSelectedAccountIds(keys.map(Number)), + }} + columns={accountColumns} + dataSource={accounts} + pagination={{ pageSize: 8, showSizeChanger: false }} + scroll={{ x: 700 }} + /> + + + + + + + + + {[ + 'get_bind_qr', + 'confirm_bind', + 'create_elite_qr', + 'query_points', + 'query_game_name', + 'query_change_bind_time', + 'query_limited_goods', + 'query_gold_balance', + ].map((key) => ( + + + + ))} + + + +
+ + + setConfigOpen(false)} + onOk={saveConfig} + okText="保存" + cancelText="取消" + width={720} + > +
+ + + + + + + + + + + + + + + + setPayTask(null)} + footer={null} + centered + > + {payTask && ( + + {taskTypes[payTask.task_type] || payTask.task_type} + + + {taskPayUrl(payTask)} + + + )} + + + ); +}