diff --git a/core/huya/__init__.py b/core/huya/__init__.py index 10be1dc..bfeb306 100644 --- a/core/huya/__init__.py +++ b/core/huya/__init__.py @@ -6,4 +6,19 @@ from .wss_client import HuyaWssClient __all__ = [ "HuyaHttpClient", "HuyaWssClient", + "GetUserScoreReq", + "GetUserScoreResp", ] + + +def __getattr__(name: str): + if name in {"GetUserScoreReq", "GetUserScoreResp"}: + from .activity_structs import GetUserScoreReq, GetUserScoreResp + + values = { + "GetUserScoreReq": GetUserScoreReq, + "GetUserScoreResp": GetUserScoreResp, + } + globals().update(values) + return values[name] + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/core/huya/activity_structs.py b/core/huya/activity_structs.py new file mode 100644 index 0000000..b3bccbf --- /dev/null +++ b/core/huya/activity_structs.py @@ -0,0 +1,130 @@ +"""虎牙活动积分相关 JCE 结构。""" + +from .taf_protocol import TafInputStream, TafOutputStream, TafStruct, TafType + + +class ActivityUserId(TafStruct): + """活动组件使用的用户标识。""" + + def __init__(self): + self.lUid: int = 0 + self.sGuid: str = "" + self.sToken: str = "" + self.sHuYaUA: str = "web&1.0&huya" + self.sCookie: str = "" + self.iTokenType: int = 0 + self.sDeviceInfo: str = "" + + def write_to(self, os: TafOutputStream): + os.write_int64(0, self.lUid) + os.write_string(1, self.sGuid) + os.write_string(2, self.sToken) + os.write_string(3, self.sHuYaUA) + os.write_string(4, self.sCookie) + os.write_int32(5, self.iTokenType) + os.write_string(6, self.sDeviceInfo) + + def read_from(self, ins: TafInputStream): + self.lUid = ins.read_int64(0, default=self.lUid) + self.sGuid = ins.read_string(1, default=self.sGuid) + self.sToken = ins.read_string(2, default=self.sToken) + self.sHuYaUA = ins.read_string(3, default=self.sHuYaUA) + self.sCookie = ins.read_string(4, default=self.sCookie) + self.iTokenType = ins.read_int32(5, default=self.iTokenType) + self.sDeviceInfo = ins.read_string(6, default=self.sDeviceInfo) + + +class GetUserScoreReq(TafStruct): + """webActUI.getUserScore 请求。""" + + def __init__(self): + self.userId = ActivityUserId() + self.sid: int = 0 + + def write_to(self, os: TafOutputStream): + os.write_struct(0, self.userId) + os.write_int32(1, self.sid) + + def read_from(self, ins: TafInputStream): + self.userId = ins.read_struct(0, ActivityUserId) or self.userId + self.sid = ins.read_int32(1, default=self.sid) + + +class GetUserScoreResp(TafStruct): + """webActUI.getUserScore 响应。""" + + def __init__(self): + self.status: int = 0 + self.msg: str = "" + self.gainScore: int = 0 + self.usedScore: int = 0 + self.newGainScore: int = 0 + self.newUsedScore: int = 0 + self.score: int = 0 + self.isOpenExchangeTip: int = 0 + + @property + def available_score(self) -> int: + """按现网页组件逻辑计算可用积分。""" + if self.newGainScore or self.newUsedScore: + return max(0, int(self.newGainScore) - int(self.newUsedScore)) + return int(self.score) + + def read_from(self, ins: TafInputStream): + self.status = ins.read_int32(0, default=self.status) + self.msg = ins.read_string(1, default=self.msg) + self.gainScore = ins.read_int32(2, default=self.gainScore) + self.usedScore = ins.read_int32(3, default=self.usedScore) + self.newGainScore = ins.read_int64(4, default=self.newGainScore) + self.newUsedScore = ins.read_int64(5, default=self.newUsedScore) + self.score = ins.read_int64(6, default=self.score) + self.isOpenExchangeTip = ins.read_int32(7, default=self.isOpenExchangeTip) + + def write_to(self, os: TafOutputStream): + os.write_int32(0, self.status) + os.write_string(1, self.msg) + os.write_int32(2, self.gainScore) + os.write_int32(3, self.usedScore) + os.write_int64(4, self.newGainScore) + os.write_int64(5, self.newUsedScore) + os.write_int64(6, self.score) + os.write_int32(7, self.isOpenExchangeTip) + + def to_dict(self) -> dict: + return { + "status": self.status, + "msg": self.msg, + "gain_score": self.gainScore, + "used_score": self.usedScore, + "new_gain_score": self.newGainScore, + "new_used_score": self.newUsedScore, + "score": self.score, + "available_score": self.available_score, + "is_open_exchange_tip": self.isOpenExchangeTip, + } + + +def _self_check(): + rsp = GetUserScoreResp() + rsp.status = 200 + rsp.msg = "请求成功" + rsp.gainScore = 300 + rsp.usedScore = 120 + rsp.newGainScore = 300 + rsp.newUsedScore = 120 + rsp.score = 180 + + os = TafOutputStream() + os.write_struct(0, rsp) + ins = TafInputStream(os.get_bytes()) + _, dtype = ins.read_head() + assert dtype == TafType.STRUCT_BEGIN + + parsed = GetUserScoreResp() + parsed.read_from(ins) + assert parsed.status == 200 + assert parsed.available_score == 180 + + +if __name__ == "__main__": + _self_check() diff --git a/core/huya/http_client.py b/core/huya/http_client.py index f4c3653..46e5d9e 100644 --- a/core/huya/http_client.py +++ b/core/huya/http_client.py @@ -162,6 +162,16 @@ class HuyaHttpClient: "input_1": str(item_count), } + @staticmethod + def _build_activity_user(uid: int, cookie: str): + """构造活动组件 UserId,与当前积分兑换组件保持一致。""" + from .activity_structs import ActivityUserId + user = ActivityUserId() + user.lUid = uid + user.sHuYaUA = "web&1.0&huya" + user.sCookie = cookie or "" + return user + def call_rpc(self, service: str, method: str, req_struct: TafStruct, rsp_class=None, uid: int = 0, guid: str = "", cookie: str = "", @@ -233,6 +243,22 @@ class HuyaHttpClient: # ---------- 业务便捷方法 ---------- + def query_user_score(self, uid: int, cookie: str, sid: int, timeout: float = 15.0): + """查询活动可用积分。""" + from .activity_structs import GetUserScoreReq, GetUserScoreResp + req = GetUserScoreReq() + req.userId = self._build_activity_user(uid, cookie) + req.sid = sid + return self.call_rpc( + "webActUI", + "getUserScore", + req, + GetUserScoreResp, + uid=uid, + cookie=cookie, + timeout=timeout, + ) + def get_goods_info(self, uid, guid, cookie, pid, spu_id, sku_id=0, game_id="", source_id="yellowcarlist", scene=7): from .shop_structs import GetGoodsInfoReqV5, GoodsInfoRsp diff --git a/web/backend/routers/huya.py b/web/backend/routers/huya.py index 3459425..840d23a 100644 --- a/web/backend/routers/huya.py +++ b/web/backend/routers/huya.py @@ -1,11 +1,13 @@ """虎牙基础管理路由""" +import asyncio +import threading from datetime import datetime, timezone -from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket +from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect from sqlalchemy.orm import Session, joinedload -from ..database import get_db +from ..database import SessionLocal, get_db from ..deps import authenticate_websocket, require_permission from ..models import HuyaAccount, HuyaConfig, HuyaGoodsSnapshot, HuyaTask, User from ..schemas import ( @@ -23,6 +25,7 @@ from ..services.huya_service import ( ensure_huya_config, import_huya_cookies, ) +from ..services.huya_runner import HuyaBatchRunner, huya_batch_registry router = APIRouter(prefix="/api/huya", tags=["虎牙"]) @@ -197,12 +200,12 @@ def list_goods( @router.post("/tasks/batch") -def create_task_batch( +async def create_task_batch( req: HuyaTaskBatchRequest, db: Session = Depends(get_db), current: User = Depends(require_permission("huya:task")), ): - """创建虎牙任务记录,真实执行器后续接入。""" + """创建虎牙任务记录并启动后台执行器。""" if not req.account_ids: raise HTTPException(status_code=400, detail="请选择虎牙账号") try: @@ -217,6 +220,24 @@ def create_task_batch( raise HTTPException(status_code=400, detail=str(exc)) from exc if count == 0: raise HTTPException(status_code=400, detail="没有有效的虎牙账号") + + log_queue = asyncio.Queue() + loop = asyncio.get_running_loop() + thread_db = SessionLocal() + runner = HuyaBatchRunner( + db=thread_db, + batch_id=batch_id, + task_type=req.task_type, + payload=req.payload, + log_queue=log_queue, + loop=loop, + concurrency=req.concurrency, + ) + huya_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} @@ -234,17 +255,47 @@ def list_tasks( return [_task_out(task) for task in tasks] +@router.post("/stop/{batch_id}") +def stop_batch( + batch_id: str, + current: User = Depends(require_permission("huya:task")), +): + """停止正在运行的虎牙批次。""" + batch = huya_batch_registry.get(batch_id) + if batch: + batch["runner"].stop() + return {"message": "已发送停止信号", "success": True} + raise HTTPException(status_code=404, detail="批次不存在或已结束") + + @router.websocket("/ws/{batch_id}") async def ws_huya_logs(websocket: WebSocket, batch_id: str): - """虎牙实时日志占位通道。""" + """虎牙实时日志推送通道。""" user = authenticate_websocket(websocket) if not user: await websocket.close(code=1008, reason="未授权") return await websocket.accept() - await websocket.send_json({ - "level": "warning", - "message": f"虎牙批次 {batch_id} 已创建,真实执行器尚未接入", - }) - await websocket.send_json({"level": "result", "message": ""}) - await websocket.close() + + batch = huya_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: + huya_batch_registry.pop(batch_id) diff --git a/web/backend/services/huya_runner.py b/web/backend/services/huya_runner.py new file mode 100644 index 0000000..c59cfc2 --- /dev/null +++ b/web/backend/services/huya_runner.py @@ -0,0 +1,290 @@ +"""虎牙任务批次执行器。""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +from typing import Optional + +from sqlalchemy.orm import Session, joinedload + +from core.huya import HuyaHttpClient +from ..database import SessionLocal +from ..models import HuyaAccount, HuyaConfig, HuyaTask +from .huya_service import cookie_value + + +class HuyaBatchRunner: + """批量执行虎牙任务,通过队列推送实时日志。""" + + 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 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_info: dict) -> str: + return ( + account_info.get("nickname") + or account_info.get("username") + or account_info.get("uid") + or f"#{account_info.get('account_id')}" + ) + + @staticmethod + def _to_int(value) -> int: + text = str(value or "").strip() + return int(text) if text.isdigit() else 0 + + def _resolve_uid(self, account_info: dict) -> int: + cookie = account_info.get("cookie") or "" + return ( + self._to_int(account_info.get("yyuid")) + or self._to_int(account_info.get("uid")) + or self._to_int(cookie_value(cookie, "yyuid")) + or self._to_int(cookie_value(cookie, "udb_uid")) + ) + + def _mark_task( + self, + worker_db: Session, + task: HuyaTask, + status: str, + message: str, + result: Optional[dict] = None, + ): + task.status = status + task.message = message + task.result = result + task.finished_at = datetime.now(timezone.utc) + worker_db.commit() + + def _execute_query_points( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + account_info: dict, + config_info: dict, + ): + sid = str(self.payload.get("sid") or config_info.get("sid") or "").strip() + if not sid: + self._mark_task(worker_db, task, "failed", "请先配置虎牙活动 SID") + return + + sid_int = self._to_int(sid) + if not sid_int: + self._mark_task(worker_db, task, "failed", f"虎牙活动 SID 无效: {sid}") + return + + uid = self._resolve_uid(account_info) + if not uid: + self._mark_task(worker_db, task, "failed", "无法从账号或 Cookie 解析 yyuid") + return + + cookie = account_info.get("cookie") or "" + if not cookie: + self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") + return + + client = HuyaHttpClient(logger=lambda _msg: None) + response = client.query_user_score(uid=uid, cookie=cookie, sid=sid_int) + if response is None: + self._mark_task(worker_db, task, "error", "虎牙积分接口无响应") + return + + result = response.to_dict() + result["sid"] = sid_int + if response.status != 200: + self._mark_task( + worker_db, + task, + "failed", + response.msg or f"虎牙积分查询失败: {response.status}", + result, + ) + return + + points = response.available_score + account.points = points + account.status = "points_queried" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(worker_db, task, "success", f"积分: {points}", result) + + def _execute_one(self, task_id: int, account_info: dict, config_info: dict, total: int): + worker_db = SessionLocal() + try: + task = worker_db.query(HuyaTask).filter(HuyaTask.id == task_id).first() + account = worker_db.query(HuyaAccount).filter(HuyaAccount.id == account_info["account_id"]).first() + if not task or not account: + return + + if self._stop.is_set(): + self._mark_task(worker_db, task, "failed", "任务已停止") + return + + task.status = "running" + task.message = "执行中" + task.finished_at = None + worker_db.commit() + + with self._counter_lock: + self._started += 1 + current = self._started + + name = self._account_name(account_info) + self._push_log("info", f"[{current}/{total}] 开始虎牙任务: {name}") + + if self.task_type != "query_points": + self._mark_task(worker_db, task, "failed", "该虎牙任务执行器暂未实现") + self._push_log("warning", f"[{current}] {name} 暂未实现: {self.task_type}") + return + + try: + self._execute_query_points(worker_db, task, account, account_info, config_info) + worker_db.refresh(task) + if task.status == "success": + self._push_log("success", f"[{current}] {name} {task.message}") + else: + self._push_log("error", f"[{current}] {name} {task.message}") + except Exception as exc: + self._mark_task(worker_db, task, "error", f"执行异常: {exc}") + self._push_log("error", f"[{current}] {name} 执行异常: {exc}") + finally: + worker_db.close() + + def run(self): + """在线程中执行虎牙批次任务。""" + self._push_log( + "info", + f"虎牙批次 {self.batch_id} 开始,共执行 {self.task_type},并发数: {self.concurrency}", + ) + try: + config = self.db.query(HuyaConfig).first() + config_info = { + "sid": config.sid if config else "", + "outer_act_id": config.outer_act_id if config else "", + "bind_act_id": config.bind_act_id if config else "", + "pay_channel": config.pay_channel if config else "", + } + + tasks = ( + self.db.query(HuyaTask) + .options(joinedload(HuyaTask.account)) + .filter(HuyaTask.batch_id == self.batch_id) + .order_by(HuyaTask.id.asc()) + .all() + ) + + task_infos = [] + for task in tasks: + account = task.account + if not account: + task.status = "error" + task.message = "账号不存在" + task.finished_at = datetime.now(timezone.utc) + continue + task.status = "pending" + task.message = "等待执行" + task.finished_at = None + task_infos.append({ + "task_id": task.id, + "account_info": { + "account_id": account.id, + "uid": account.uid or "", + "yyuid": account.yyuid or "", + "username": account.username or "", + "nickname": account.nickname or "", + "cookie": account.cookie or "", + }, + }) + self.db.commit() + + total = len(task_infos) + if total == 0: + self._push_log("warning", "没有可执行的虎牙任务") + self._push_log("result", "") + return + + with ThreadPoolExecutor(max_workers=self.concurrency) as executor: + futures = [] + for item in task_infos: + if self._stop.is_set(): + self._push_log("warning", "任务已停止,跳过剩余账号") + break + futures.append(executor.submit( + self._execute_one, + item["task_id"], + item["account_info"], + config_info, + total, + )) + + for future in as_completed(futures): + try: + future.result() + except Exception as exc: + self._push_log("error", f"虎牙 Worker 异常: {exc}") + + self._push_log("info", f"虎牙批次 {self.batch_id} 完成") + self._push_log("result", "") + except Exception as exc: + self._push_log("error", f"虎牙批次执行异常: {exc}") + self._push_log("result", "") + finally: + self.db.close() + + +class HuyaBatchRegistry: + """管理运行中的虎牙批次。""" + + def __init__(self): + self._batches: dict[str, dict] = {} + + def register( + self, + batch_id: str, + log_queue: asyncio.Queue, + loop: asyncio.AbstractEventLoop, + runner: HuyaBatchRunner, + ): + self._batches[batch_id] = { + "log_queue": log_queue, + "loop": loop, + "runner": runner, + } + + 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) + + +huya_batch_registry = HuyaBatchRegistry() diff --git a/web/backend/services/huya_service.py b/web/backend/services/huya_service.py index d941544..0e77cf5 100644 --- a/web/backend/services/huya_service.py +++ b/web/backend/services/huya_service.py @@ -140,7 +140,7 @@ def create_planned_tasks( created_by: int, payload: dict | None = None, ) -> tuple[str, int]: - """创建虎牙任务记录,真实执行器后续接入。""" + """创建虎牙任务记录,等待后台执行器消费。""" if task_type not in SUPPORTED_TASK_TYPES: raise ValueError("不支持的任务类型") @@ -153,7 +153,7 @@ def create_planned_tasks( account_id=account.id, task_type=task_type, status="planned", - message="任务已创建,等待虎牙执行器接入", + message="任务已创建,等待执行", result={"payload": payload} if payload else None, created_by=created_by, ))