yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复虎牙任务台卡死与轮询过重，并调整默认端口

Browse Source 
默认端口改为后端 8800 / 前端 5174，避免多项目冲突。
清理无执行器的残留 running 任务，支持停止无效批次；绑定二维码结果页不再被假活跃批次锁死。
任务列表默认剥离 base64 小程序码，按需拉取详情，空闲轮询降频，显著降低 tasks 请求体积。
This commit is contained in:
yml2213
2026-07-24 11:38:07 +08:00
parent 69b6120ede
commit 9ec4bba0a9
13 changed files with 430 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/backend/main.py
+16 -2
View File
@@ -25,6 +25,19 @@ async def lifespan(app: FastAPI):
setup_logger(level=_log_level, log_dir=str(_log_dir))
init_db()
# 进程重启后内存批次丢失，清理历史 pending/running，避免前端被假活跃批次锁死。
from loguru import logger
from .database import SessionLocal
from .services.huya_service import cleanup_orphan_huya_tasks
db = SessionLocal()
try:
cleaned = cleanup_orphan_huya_tasks(db, message="任务已中断（服务重启）")
if cleaned:
logger.info(f"启动清理虎牙残留任务: {cleaned} 条")
finally:
db.close()
yield
@@ -39,7 +52,7 @@ _cors_env = os.getenv("CORS_ORIGINS", "")
if _cors_env:
_cors_origins = [o.strip() for o in _cors_env.split(",") if o.strip()]
else:
_cors_origins = ["http://localhost:5173", "http://localhost:3000"]
_cors_origins = ["http://localhost:5174", "http://localhost:5173", "http://localhost:3000"]
app.add_middleware(
CORSMiddleware,
@@ -115,7 +128,8 @@ if _INDEX_HTML.exists():
def run():
_reload = os.getenv("UVICORN_RELOAD", "false").lower() == "true"
uvicorn.run("web.backend.main:app", host="0.0.0.0", port=8000, reload=_reload)
_port = int(os.getenv("BACKEND_PORT", "8800"))
uvicorn.run("web.backend.main:app", host="0.0.0.0", port=_port, reload=_reload)
if __name__ == "__main__":
web/backend/routers/huya.py
+77 -4
View File
@@ -51,6 +51,7 @@ from ..services.huya_service import (
HUYA_CONFIG_FIELDS,
SUPPORTED_TASK_TYPES,
apply_huya_config_defaults,
cleanup_orphan_huya_tasks,
create_planned_tasks,
ensure_huya_config,
huya_config_value,
@@ -133,7 +134,23 @@ def _account_out(account: HuyaAccount, include_cookie: bool = True) -> HuyaAccou
)
def _task_out(task: HuyaTask) -> HuyaTaskOut:
def _sanitize_task_result(result: dict | None, *, include_images: bool = False) -> dict | None:
"""列表接口默认剥离 base64 图片，避免轮询每次传 1MB+ 数据。"""
if not isinstance(result, dict):
return result
data = dict(result)
image = data.get("mini_qrcode_image")
if isinstance(image, str) and image:
data["has_mini_qrcode"] = True
if not include_images:
data.pop("mini_qrcode_image", None)
elif data.get("has_mini_qrcode"):
data["has_mini_qrcode"] = True
return data
def _task_out(task: HuyaTask, *, include_images: bool = False) -> HuyaTaskOut:
account = task.account
return HuyaTaskOut(
id=task.id,
@@ -144,7 +161,7 @@ def _task_out(task: HuyaTask) -> HuyaTaskOut:
task_type=task.task_type,
status=task.status or "",
message=task.message or "",
result=task.result,
result=_sanitize_task_result(task.result if isinstance(task.result, dict) else None, include_images=include_images),
created_by=task.created_by,
created_at=task.created_at,
finished_at=task.finished_at,
@@ -952,6 +969,16 @@ async def create_task_batch(
"""创建虎牙任务记录并启动后台执行器。"""
if not req.account_ids:
raise HTTPException(status_code=400, detail="请选择虎牙账号")
# 先清掉没有执行器的历史 running/pending，避免前端被假活跃批次锁死。
# 不清理 planned：避免与刚创建的新任务产生竞态。
cleanup_orphan_huya_tasks(
db,
active_batch_ids=huya_batch_registry.active_ids(),
statuses=("pending", "running"),
message="任务已中断（无执行器接管）",
)
try:
batch_id, count = create_planned_tasks(
db,
@@ -988,27 +1015,71 @@ async def create_task_batch(
@router.get("/tasks", response_model=list[HuyaTaskOut])
def list_tasks(
batch_id: str | None = None,
include_images: bool = Query(False, description="是否返回 base64 小程序码（默认否，轮询请保持 false）"),
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:task")),
):
"""查看虎牙任务记录。"""
cleanup_orphan_huya_tasks(
db,
active_batch_ids=huya_batch_registry.active_ids(),
statuses=("pending", "running"),
message="任务已中断（无执行器接管）",
)
query = db.query(HuyaTask).options(joinedload(HuyaTask.account))
if batch_id:
query = query.filter(HuyaTask.batch_id == batch_id)
tasks = query.order_by(HuyaTask.id.desc()).limit(300).all()
return [_task_out(task) for task in tasks]
return [_task_out(task, include_images=include_images) for task in tasks]
@router.get("/tasks/{task_id}", response_model=HuyaTaskOut)
def get_task(
task_id: int,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:task")),
):
"""获取单条虎牙任务详情（含二维码图片）。"""
task = (
db.query(HuyaTask)
.options(joinedload(HuyaTask.account))
.filter(HuyaTask.id == task_id)
.first()
)
if not task:
raise HTTPException(status_code=404, detail="任务不存在")
return _task_out(task, include_images=True)
@router.post("/stop/{batch_id}")
def stop_batch(
batch_id: str,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:task")),
):
"""停止正在运行的虎牙批次。"""
batch = huya_batch_registry.get(batch_id)
if batch:
if batch.get("finished"):
huya_batch_registry.pop(batch_id)
cleaned = cleanup_orphan_huya_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_huya_tasks(
db,
batch_id=batch_id,
message="任务已停止（批次不存在）",
)
if cleaned:
return {"message": f"已清理 {cleaned} 个残留任务", "success": True}
raise HTTPException(status_code=404, detail="批次不存在或已结束")
@@ -1042,4 +1113,6 @@ async def ws_huya_logs(websocket: WebSocket, batch_id: str):
except WebSocketDisconnect:
pass
finally:
huya_batch_registry.pop(batch_id)
latest = huya_batch_registry.get(batch_id)
if latest and latest.get("finished"):
huya_batch_registry.pop(batch_id)
web/backend/services/huya_runner.py
+57 -114
View File
@@ -22,8 +22,6 @@ HUYA_RECHARGE_SOURCE_ID = "yellowcarlist"
HUYA_RECHARGE_SCENE = 4
HUYA_PAYMENT_POLL_SECONDS = 180
HUYA_PAYMENT_POLL_INTERVAL = 3
HUYA_BIND_ROLE_POLL_SECONDS = 180
HUYA_BIND_ROLE_POLL_INTERVAL = 3
HUYA_BIND_ZT_UUID = "b02faae1"
HUYA_BIND_ROOM_ID = "30596253"
HUYA_RECHARGE_EXTRA_PRODUCTS = [
@@ -65,6 +63,11 @@ class HuyaBatchRunner:
self._stop.set()
def _push_log(self, level: str, message: str):
if level == "result":
try:
huya_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"[huya] {message}")
@@ -252,86 +255,6 @@ class HuyaBatchRunner:
f"&roomid={HUYA_BIND_ROOM_ID}"
)
def _wait_bind_role_result(
self,
client: HuyaHttpClient,
worker_db: Session,
task: HuyaTask,
account: HuyaAccount,
uid: int,
cookie: str,
b_act_id_int: int,
result: dict,
) -> tuple[str, dict]:
deadline = time.monotonic() + HUYA_BIND_ROLE_POLL_SECONDS
qrcode_token = str(result.get("qrcode_token") or "")
qrcode_finished = not qrcode_token
while not self._stop.is_set() and time.monotonic() < deadline:
if self._stop.wait(HUYA_BIND_ROLE_POLL_INTERVAL):
break
if qrcode_token and not qrcode_finished:
qrcode_status = client.get_livelink_qrcode_status(qrcode_token, timeout=10.0)
if qrcode_status is not None:
result["qrcode_status"] = qrcode_status
if qrcode_status["is_expired"] or qrcode_status["is_failure"]:
result.update({
"bind_phase": "qrcode_expired",
"bind_ready_for_confirm": False,
})
self._update_task_progress(worker_db, task, "running", "绑定小程序码已失效，请重新获取", result)
return "", result
if qrcode_status["is_completed"]:
qrcode_finished = True
result["bind_phase"] = "qrcode_completed"
elif qrcode_status["is_scan"]:
result["bind_phase"] = "qrcode_scanned"
else:
result["bind_phase"] = "waiting_scan"
bind_status, bind_query_result = self._resolve_bind_status(
client=client,
uid=uid,
cookie=cookie,
b_act_id_int=b_act_id_int,
)
if bind_status is None:
continue
result.update(bind_query_result)
if bind_status.status != 200:
result.update({
"bind_phase": "role_check_failed",
"bind_status": bind_status.to_dict(),
})
self._update_task_progress(worker_db, task, "running", bind_status.msg or "等待绑定角色同步", result)
continue
previous_phase = result.get("bind_phase")
ready = self._bind_ready_result(bind_status)
if not ready["role_name"] and previous_phase in {"waiting_scan", "qrcode_scanned", "qrcode_completed"}:
ready["bind_phase"] = previous_phase
result.update(ready)
role_name = ready["role_name"]
if role_name:
self._apply_role_to_account(account, bind_status, "game_queried")
self._update_task_progress(worker_db, task, "running", f"已识别角色: {role_name}，待确认绑定", result)
return role_name, result
if result.get("bind_phase") == "qrcode_completed":
message = "小程序绑定已完成，等待角色同步"
elif result.get("bind_phase") == "qrcode_scanned":
message = "已扫码，等待小程序绑定完成"
else:
message = "已生成绑定小程序码，等待扫码绑定"
self._update_task_progress(worker_db, task, "running", message, result)
result.update({
"bind_phase": "role_timeout" if not self._stop.is_set() else "stopped",
"bind_ready_for_confirm": False,
})
return "", result
def _mark_task(
self,
worker_db: Session,
@@ -671,7 +594,7 @@ class HuyaBatchRunner:
return
if scheduled_at and scheduled_at.timestamp() > time.time():
if not self._wait_until(scheduled_at, uid):
self._mark_task(worker_db, task, "failed", "兑换任务已停止")
self._mark_task(worker_db, task, "stopped", "兑换任务已停止")
return
client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}"))
@@ -1037,6 +960,10 @@ class HuyaBatchRunner:
paid_message += f"，支付时间 {self._format_local_time(int(payment_order['pay_time']) // 1000)}"
self._mark_task(worker_db, task, "success", paid_message, result)
return
if payment_status == "stopped":
account.status = "recharge_order_created"
self._mark_task(worker_db, task, "stopped", f"{message}，已停止监听支付", result)
return
account.status = "recharge_order_created"
timeout_message = f"{message}，{result['payment_status_label']}"
@@ -1167,6 +1094,7 @@ class HuyaBatchRunner:
**bind_query_result,
"bind_phase": "waiting_scan" if mini_qrcode.get("qrcode_token") else "waiting_role",
"bind_ready_for_confirm": False,
"bind_polling": False,
"bind_redirect_url": bind_redirect_url,
**role_info,
**change_state,
@@ -1181,29 +1109,7 @@ class HuyaBatchRunner:
account.game_channel = self._role_channel(bind_status) or account.game_channel
account.nickname = profile_nick or account.nickname
account.updated_at = datetime.now(timezone.utc)
self._update_task_progress(worker_db, task, "running", "已生成绑定小程序码，等待绑定角色", result)
role_name, result = self._wait_bind_role_result(
client=client,
worker_db=worker_db,
task=task,
account=account,
uid=uid,
cookie=cookie,
b_act_id_int=b_act_id_int,
result=result,
)
if role_name:
self._mark_task(worker_db, task, "success", f"已识别角色: {role_name}，待确认绑定", result)
return
if result.get("bind_phase") == "stopped":
self._mark_task(worker_db, task, "failed", "任务已停止", result)
return
if result.get("bind_phase") == "qrcode_expired":
self._mark_task(worker_db, task, "failed", "绑定小程序码已失效，请重新获取", result)
return
self._mark_task(worker_db, task, "success", "已生成绑定小程序码，未检测到绑定角色", result)
self._mark_task(worker_db, task, "success", "已生成绑定小程序码", result)
def _execute_query_game_name(
self,
@@ -1260,6 +1166,8 @@ class HuyaBatchRunner:
**role_info,
**change_state,
**bind_query_result,
"bind_ready_for_confirm": bool(role_info["role_name"]),
"bind_phase": "role_ready" if role_info["role_name"] else "waiting_role",
}
role_name = role_info["role_name"]
if role_name:
@@ -1417,7 +1325,7 @@ class HuyaBatchRunner:
return
if self._stop.is_set():
self._mark_task(worker_db, task, "failed", "任务已停止")
self._mark_task(worker_db, task, "stopped", "任务已停止")
return
task.status = "running"
@@ -1562,6 +1470,17 @@ class HuyaBatchRegistry:
def __init__(self):
self._batches: dict[str, dict] = {}
self._lock = threading.Lock()
def _cleanup_locked(self, ttl_seconds: int = 300):
now = time.time()
expired = [
batch_id
for batch_id, batch in self._batches.items()
if batch.get("finished") and now - float(batch.get("finished_at") or now) > ttl_seconds
]
for batch_id in expired:
self._batches.pop(batch_id, None)
def register(
self,
@@ -1570,17 +1489,41 @@ class HuyaBatchRegistry:
loop: asyncio.AbstractEventLoop,
runner: HuyaBatchRunner,
):
self._batches[batch_id] = {
"log_queue": log_queue,
"loop": loop,
"runner": runner,
}
with self._lock:
self._cleanup_locked()
self._batches[batch_id] = {
"log_queue": log_queue,
"loop": loop,
"runner": runner,
"finished": False,
"finished_at": None,
}
def get(self, batch_id: str):
return self._batches.get(batch_id)
with self._lock:
self._cleanup_locked()
return self._batches.get(batch_id)
def active_ids(self) -> set[str]:
with self._lock:
self._cleanup_locked()
return {
batch_id
for batch_id, batch in self._batches.items()
if not batch.get("finished")
}
def mark_finished(self, batch_id: str):
with self._lock:
batch = self._batches.get(batch_id)
if not batch:
return
batch["finished"] = True
batch["finished_at"] = time.time()
def pop(self, batch_id: str):
return self._batches.pop(batch_id, None)
with self._lock:
return self._batches.pop(batch_id, None)
huya_batch_registry = HuyaBatchRegistry()
web/backend/services/huya_service.py
+34
View File
@@ -313,6 +313,40 @@ def ensure_huya_config(db: Session) -> HuyaConfig:
return config
HUYA_ACTIVE_TASK_STATUSES = ("planned", "pending", "running")
HUYA_STALE_TASK_STATUSES = ("pending", "running")
def cleanup_orphan_huya_tasks(
db: Session,
*,
active_batch_ids: set[str] | None = None,
batch_id: str | None = None,
statuses: tuple[str, ...] = HUYA_ACTIVE_TASK_STATUSES,
message: str = "任务已中断（服务重启或批次丢失）",
) -> int:
"""清理没有执行器接管的虎牙任务，避免前端被历史 running 状态锁死。"""
query = db.query(HuyaTask).filter(HuyaTask.status.in_(statuses))
if batch_id:
query = query.filter(HuyaTask.batch_id == batch_id)
elif active_batch_ids is not None:
if active_batch_ids:
query = query.filter(~HuyaTask.batch_id.in_(list(active_batch_ids)))
# active_batch_ids is None 且未指定 batch_id：清理全部匹配状态
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)
def create_planned_tasks(
db: Session,
account_ids: list[int],