969 lines
34 KiB
Python
969 lines
34 KiB
Python
"""斗鱼活动任务路由。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import threading
|
||
from datetime import UTC, datetime
|
||
from typing import Any, cast
|
||
|
||
from fastapi import (
|
||
APIRouter,
|
||
Depends,
|
||
HTTPException,
|
||
Query,
|
||
Request,
|
||
WebSocket,
|
||
WebSocketDisconnect,
|
||
)
|
||
from loguru import logger
|
||
from sqlalchemy import case, func, or_, select
|
||
from sqlalchemy.orm import Session, joinedload
|
||
|
||
from core.douyu import (
|
||
FishFinRechargeClient,
|
||
FishFinRechargeConfig,
|
||
FishFinRechargeConfigError,
|
||
)
|
||
|
||
from ..database import SessionLocal, get_db
|
||
from ..deps import authenticate_websocket, get_current_user, require_permission
|
||
from ..models import (
|
||
Account,
|
||
AccountLease,
|
||
DouyuConfig,
|
||
DouyuEsportsGoodsSnapshot,
|
||
DouyuGoodsSnapshot,
|
||
DouyuTask,
|
||
DouyuWorkbench,
|
||
DouyuWorkbenchAccount,
|
||
DouyuXpdGoodsSnapshot,
|
||
User,
|
||
)
|
||
from ..permissions import user_has_permission
|
||
from ..schemas import (
|
||
DouyuConfigOut,
|
||
DouyuConfigUpdate,
|
||
DouyuGoodsOut,
|
||
DouyuTaskAccountOut,
|
||
DouyuTaskBatchRequest,
|
||
DouyuTaskOut,
|
||
DouyuWorkbenchAccountsUpdate,
|
||
DouyuXpdGoodsOut,
|
||
)
|
||
from ..services.account_binding import bind_payload
|
||
from ..services.audit_service import record_audit
|
||
from ..services.business_events import record_recharge
|
||
from ..services.douyu_runner import DouyuBatchRunner, douyu_batch_registry
|
||
from ..services.douyu_service import (
|
||
DOUYU_CONFIG_FIELDS,
|
||
DOUYU_HANDBOOK_TASK_TYPES,
|
||
SUPPORTED_DOUYU_TASK_TYPES,
|
||
apply_douyu_config_defaults,
|
||
cleanup_orphan_douyu_tasks,
|
||
cookie_holding_filter,
|
||
create_douyu_planned_tasks,
|
||
douyu_config_value,
|
||
douyu_task_payload,
|
||
ensure_douyu_config,
|
||
)
|
||
|
||
router = APIRouter(prefix="/api/douyu", tags=["斗鱼活动"])
|
||
|
||
|
||
def _supplier_response_value(payload: dict, *keys: str):
|
||
"""兼容供应商把订单字段放在根节点、data 或 result 中。"""
|
||
data_raw = payload.get("data")
|
||
data = cast(dict[str, Any], data_raw) if isinstance(data_raw, dict) else {}
|
||
result_raw = payload.get("result")
|
||
result = cast(dict[str, Any], result_raw) if isinstance(result_raw, dict) else {}
|
||
for source in (payload, data, result):
|
||
for key in keys:
|
||
if source.get(key) is not None:
|
||
return source[key]
|
||
return None
|
||
|
||
|
||
@router.post("/supplier-recharge/callback")
|
||
async def supplier_recharge_callback(request: Request, db: Session = Depends(get_db)):
|
||
"""接收供应商直充终态通知，验签后按 out_order_id 幂等更新任务。"""
|
||
try:
|
||
payload = await request.json()
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail="供应商回调不是 JSON") from exc
|
||
if not isinstance(payload, dict):
|
||
raise HTTPException(status_code=400, detail="供应商回调格式无效")
|
||
try:
|
||
client = FishFinRechargeClient(FishFinRechargeConfig.from_env())
|
||
except FishFinRechargeConfigError as exc:
|
||
logger.error("[douyu] 供应商直充回调配置无效: {}", exc)
|
||
raise HTTPException(status_code=503, detail="供应商直充回调未配置") from exc
|
||
if not client.verify_response_sign(payload, "POST"):
|
||
logger.warning("[douyu] 供应商直充回调验签失败: keys={}", sorted(payload))
|
||
raise HTTPException(status_code=401, detail="供应商回调签名无效")
|
||
|
||
out_order_id = str(_supplier_response_value(payload, "out_order_id") or "").strip()
|
||
if not out_order_id:
|
||
raise HTTPException(status_code=400, detail="供应商回调缺少 out_order_id")
|
||
task = (
|
||
db.query(DouyuTask)
|
||
.filter(
|
||
DouyuTask.supplier_out_order_id == out_order_id,
|
||
DouyuTask.task_type == "create_gold_qr",
|
||
)
|
||
.first()
|
||
)
|
||
if not task:
|
||
raise HTTPException(status_code=404, detail="供应商回调订单不存在")
|
||
|
||
status = DouyuBatchRunner._to_int(
|
||
_supplier_response_value(payload, "order_status", "orderStatus")
|
||
)
|
||
result = dict(task.result) if isinstance(task.result, dict) else {}
|
||
result.update(
|
||
{
|
||
"out_order_id": out_order_id,
|
||
"order_id": _supplier_response_value(payload, "order_id", "orderId")
|
||
or result.get("order_id"),
|
||
"supplier_order_status": status,
|
||
"supplier_order": DouyuBatchRunner._supplier_result(payload),
|
||
"supplier_callback_received": True,
|
||
}
|
||
)
|
||
# 这笔充值到没到账，用**回调自己说的终态**判断，而不是「任务状态有没有被改」。
|
||
# 否则会漏掉一个真实黑洞：runner 轮询超时先判了 failed（gold.py 末尾），
|
||
# 回调随后才到达成功 —— 守卫会跳过状态更新，这笔真金白银永远不进时间线。
|
||
recharge_confirmed = status == 2
|
||
if task.status not in {"success", "failed", "stopped"}:
|
||
if status == 2:
|
||
task.status = "success"
|
||
task.message = "供应商直充成功（异步通知）"
|
||
task.finished_at = datetime.now(UTC)
|
||
if task.account:
|
||
task.account.updated_at = datetime.now(UTC)
|
||
elif status in {3, 4}:
|
||
reason = str(
|
||
_supplier_response_value(payload, "fail_reason", "message", "msg")
|
||
or "供应商直充失败"
|
||
)
|
||
task.status = "failed"
|
||
task.message = reason[:512]
|
||
task.finished_at = datetime.now(UTC)
|
||
else:
|
||
task.status = "running"
|
||
task.message = (
|
||
f"供应商直充订单处理中（状态 {status if status is not None else '-'}）"
|
||
)
|
||
task.result = result
|
||
callback_success = status not in {3, 4}
|
||
# 回调说成功了就补一条充值事件。幂等键与 runner 那条**完全一致**
|
||
# （business_events.task_ref），所以「runner 也观测到了」时不会重复记账；
|
||
# 而「runner 已判超时/失败、回调才到」这个黑洞则能被补上。
|
||
if recharge_confirmed and task.account is not None:
|
||
record_recharge(
|
||
db,
|
||
task.account,
|
||
task_id=task.id,
|
||
amount=_to_int_or_none(result.get("buy_num")),
|
||
order_id=str(result.get("out_order_id") or ""),
|
||
supplier_order_id=str(result.get("order_id") or ""),
|
||
channel="supplier_api",
|
||
douyu_uid=str(result.get("douyu_uid") or ""),
|
||
balance_after=task.account.gold_balance,
|
||
confirmed=False,
|
||
)
|
||
record_audit(
|
||
db,
|
||
None,
|
||
action="recharge:douyu:callback",
|
||
target=f"supplier_order:{out_order_id}",
|
||
detail={
|
||
"out_order_id": out_order_id,
|
||
"task_id": task.id,
|
||
"order_id": result.get("order_id"),
|
||
"supplier_order_status": status,
|
||
"task_status": task.status,
|
||
"message": task.message,
|
||
},
|
||
success=callback_success,
|
||
)
|
||
db.commit()
|
||
logger.info(
|
||
"[douyu] 供应商直充回调已处理: out_order_id={} status={}", out_order_id, status
|
||
)
|
||
acknowledgement = {"code": 200, "message": "success"}
|
||
acknowledgement["sign"] = client.sign(acknowledgement, "POST")
|
||
return acknowledgement
|
||
|
||
|
||
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 就可见：排队中的普通登录（pending/running）和
|
||
重登失败（relogin_failed）都会保留旧 CK，不能让账号从工作台消失。
|
||
"""
|
||
cookie_ids = select(Account.id).where(cookie_holding_filter())
|
||
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 _visible_tasks_query(db: Session, current: User):
|
||
"""返回当前用户可查看的斗鱼任务查询。"""
|
||
# 任务列表只展示账号识别信息，避免 joinedload 把密码、邮箱等加密大字段带出。
|
||
account_loader = joinedload(DouyuTask.account).load_only(
|
||
Account.id,
|
||
Account.username,
|
||
Account.uid,
|
||
Account.nickname,
|
||
Account.assigned_to,
|
||
)
|
||
query = db.query(DouyuTask).options(account_loader)
|
||
if _can_view_all(current):
|
||
return query
|
||
if user_has_permission(current, "account:view_assigned"):
|
||
# 彻底消除全表 filesort：直接通过 douyu_tasks 上的 account_assigned_to 复合索引倒序定位
|
||
return query.filter(DouyuTask.account_assigned_to == current.id)
|
||
raise HTTPException(status_code=403, detail="无权查看斗鱼任务")
|
||
|
||
|
||
def _require_task_account_access(
|
||
db: Session, current: User, account_ids: list[int]
|
||
) -> None:
|
||
"""确保任务只会提交到当前用户可操作的账号。
|
||
|
||
可操作 = **我名下** ∪ **我当前持有有效租约**。
|
||
后半条是账号池加的（归属是管理员/运营的维护责任，客服只按需租用）——
|
||
旧路径「名下账号」一字未改，这里只做**并集**。
|
||
"""
|
||
requested_ids = set(account_ids)
|
||
query = db.query(Account.id).filter(Account.id.in_(requested_ids))
|
||
if not _can_view_all(current):
|
||
if not user_has_permission(current, "account:view_assigned"):
|
||
raise HTTPException(status_code=403, detail="无权操作斗鱼账号")
|
||
my_leases = select(AccountLease.account_id).where(
|
||
AccountLease.holder_user_id == current.id,
|
||
AccountLease.status == "active",
|
||
)
|
||
query = query.filter(
|
||
or_(Account.assigned_to == current.id, Account.id.in_(my_leases))
|
||
)
|
||
allowed_ids = {account_id for (account_id,) in query.all()}
|
||
if allowed_ids != requested_ids:
|
||
raise HTTPException(status_code=403, detail="包含无权操作的斗鱼账号")
|
||
|
||
|
||
def _require_batch_owner(db: Session, current: User, batch_id: str) -> None:
|
||
"""客服只能停止或订阅自己创建的任务批次。"""
|
||
if _can_view_all(current):
|
||
return
|
||
exists = (
|
||
db.query(DouyuTask.id)
|
||
.filter(DouyuTask.batch_id == batch_id, DouyuTask.created_by == current.id)
|
||
.first()
|
||
)
|
||
if not exists:
|
||
raise HTTPException(status_code=403, detail="无权操作该斗鱼任务批次")
|
||
|
||
|
||
def start_douyu_batch(
|
||
*,
|
||
batch_id: str,
|
||
task_type: str,
|
||
payload: dict,
|
||
concurrency: int,
|
||
) -> None:
|
||
"""在线程里启动一个批次的执行器（**必须在有事件循环的 async 路由里调用**）。
|
||
|
||
抽出来是因为账号池也要起批次：两条路径的执行方式必须完全一致 ——
|
||
复制一份的话，将来改了注册表或线程模型就会出现"老路径好了、池路径没好"。
|
||
|
||
⚠️ `DouyuBatchRunner` 按 `batch_id` 取**全部 planned 任务、不看 `task_type`**，
|
||
再按每条任务自己的类型分发。所以**一个批次里可以放不同类型的任务**；
|
||
但它们会被并发执行（`concurrency` 个 worker），顺序没有保证 ——
|
||
需要严格先后（充值 → 赠令 → 兑换）时只能把 `concurrency` 设成 1。
|
||
"""
|
||
log_queue = asyncio.Queue()
|
||
loop = asyncio.get_running_loop()
|
||
thread_db = SessionLocal()
|
||
runner = DouyuBatchRunner(
|
||
db=thread_db,
|
||
batch_id=batch_id,
|
||
task_type=task_type,
|
||
payload=payload,
|
||
log_queue=log_queue,
|
||
loop=loop,
|
||
concurrency=concurrency,
|
||
)
|
||
douyu_batch_registry.register(batch_id, log_queue, loop, runner)
|
||
thread = threading.Thread(target=runner.run, daemon=True)
|
||
thread.start()
|
||
|
||
|
||
def _parse_account_names(raw_names: str) -> list[str]:
|
||
"""解析前端粘贴的 Excel 账号名，每行一个并去重。"""
|
||
names: list[str] = []
|
||
seen: set[str] = set()
|
||
for value in (
|
||
(raw_names or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
||
):
|
||
name = value.strip()
|
||
if name and name not in seen:
|
||
names.append(name)
|
||
seen.add(name)
|
||
return names
|
||
|
||
|
||
def _apply_task_account_filters(
|
||
query,
|
||
*,
|
||
search: str = "",
|
||
tag: str = "",
|
||
account_names: str = "",
|
||
ids: str = "",
|
||
):
|
||
"""按工作台导入弹窗的搜索、标签、多行账号名筛选任务账号。"""
|
||
id_list = [int(item) for item in ids.split(",") if item.strip().isdigit()]
|
||
if id_list:
|
||
query = query.filter(Account.id.in_(id_list))
|
||
tag_text = (tag or "").strip()
|
||
if tag_text:
|
||
query = query.filter(Account.tag == tag_text)
|
||
selected_names = _parse_account_names(account_names)
|
||
if selected_names:
|
||
query = query.filter(Account.username.in_(selected_names))
|
||
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),
|
||
Account.esports_game_name.ilike(pattern),
|
||
Account.xpd_game_name.ilike(pattern),
|
||
)
|
||
)
|
||
return query, selected_names
|
||
|
||
|
||
def _order_task_accounts(query, selected_names: list[str]):
|
||
"""自定义账号名按粘贴顺序排序，未指定时保持最新账号优先。"""
|
||
if selected_names:
|
||
input_order = case(
|
||
{name: index for index, name in enumerate(selected_names)},
|
||
value=Account.username,
|
||
else_=len(selected_names),
|
||
)
|
||
return query.order_by(input_order, Account.id.desc())
|
||
return query.order_by(Account.id.desc())
|
||
|
||
|
||
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_payload(account),
|
||
change_role_wait_time=account.change_role_wait_time,
|
||
elite_opened=account.elite_opened,
|
||
elite_expire_time=account.elite_expire_time,
|
||
esports_points=account.esports_points,
|
||
esports_game_name=account.esports_game_name or "",
|
||
esports_game_channel=account.esports_game_channel or "",
|
||
esports_change_role_wait_time=account.esports_change_role_wait_time,
|
||
esports_can_change_time=account.esports_can_change_time,
|
||
xpd_game_name=account.xpd_game_name or "",
|
||
xpd_openid=account.xpd_openid or "",
|
||
xpd_role_id=account.xpd_role_id or "",
|
||
xpd_plat_id=account.xpd_plat_id,
|
||
xpd_area_id=account.xpd_area_id,
|
||
xpd_balance=account.xpd_balance,
|
||
xpd_fragments=account.xpd_fragments,
|
||
assigned_to=account.assigned_to,
|
||
assigned_username=account.assigned_user.username
|
||
if account.assigned_user
|
||
else None,
|
||
)
|
||
|
||
|
||
def _task_out(task: DouyuTask, *, include_detail: bool = False) -> DouyuTaskOut:
|
||
return DouyuTaskOut(**douyu_task_payload(task, include_detail=include_detail))
|
||
|
||
|
||
def _config_out(config: DouyuConfig) -> DouyuConfigOut:
|
||
text = lambda field: str(douyu_config_value(field, getattr(config, field)))
|
||
number = lambda field: int(douyu_config_value(field, getattr(config, field)))
|
||
boolean = lambda field: bool(douyu_config_value(field, getattr(config, field)))
|
||
return DouyuConfigOut(
|
||
manual_id=text("manual_id"),
|
||
rid=text("rid"),
|
||
bind_act_alias=text("bind_act_alias"),
|
||
confirm_act_alias=text("confirm_act_alias"),
|
||
legacy_act_alias=text("legacy_act_alias"),
|
||
room_id=text("room_id"),
|
||
elite_amount=number("elite_amount"),
|
||
esports_manual_id=text("esports_manual_id"),
|
||
esports_act_alias=text("esports_act_alias"),
|
||
esports_amount=number("esports_amount"),
|
||
esports_chicken_gift_id=text("esports_chicken_gift_id"),
|
||
esports_chicken_skin_id=text("esports_chicken_skin_id"),
|
||
esports_firework_gift_id=text("esports_firework_gift_id"),
|
||
esports_firework_skin_id=text("esports_firework_skin_id"),
|
||
peace_enabled=boolean("peace_enabled"),
|
||
xpd_act_alias=text("xpd_act_alias"),
|
||
xpd_act_id=text("xpd_act_id"),
|
||
xpd_rid=text("xpd_rid"),
|
||
gold_pay_type=number("gold_pay_type"),
|
||
gold_recharge_channel=text("gold_recharge_channel"),
|
||
gold_api_product_id=text("gold_api_product_id"),
|
||
gold_api_account_template_name=text("gold_api_account_template_name"),
|
||
gift_id=text("gift_id"),
|
||
skin_id=text("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")
|
||
def list_task_accounts(
|
||
search: str = Query(""),
|
||
tag: str = Query(""),
|
||
ids: str = Query(""),
|
||
account_names: str = Query(""),
|
||
page: int | None = Query(None, ge=1),
|
||
page_size: int = Query(50, ge=1, le=200),
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(get_current_user),
|
||
):
|
||
"""查看可执行斗鱼任务的账号（必须有成功 Cookie）。
|
||
|
||
tag 按账号标签精确筛选；ids 为逗号分隔的账号 ID，用于工作台按已导入账号过滤；
|
||
account_names 为多行斗鱼账户名（每行一个），按粘贴顺序精确匹配；为空时返回全部。
|
||
"""
|
||
query, selected_names = _apply_task_account_filters(
|
||
_visible_task_accounts_query(db, current),
|
||
search=search,
|
||
tag=tag,
|
||
account_names=account_names,
|
||
ids=ids,
|
||
)
|
||
total = None
|
||
if page is not None:
|
||
total = query.order_by(None).count()
|
||
query = _order_task_accounts(query, selected_names)
|
||
if page is not None:
|
||
query = query.offset((page - 1) * page_size).limit(page_size)
|
||
else:
|
||
query = query.limit(500)
|
||
result = [_account_out(account) for account in query.all()]
|
||
if page is not None:
|
||
return {
|
||
"items": result,
|
||
"total": total or 0,
|
||
"page": page,
|
||
"page_size": page_size,
|
||
}
|
||
return result
|
||
|
||
|
||
@router.get("/accounts/ids")
|
||
def list_task_account_ids(
|
||
search: str = Query(""),
|
||
tag: str = Query(""),
|
||
account_names: str = Query(""),
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(get_current_user),
|
||
):
|
||
"""返回当前用户可导入工作台的全部账号 ID，不受列表分页限制。"""
|
||
query, selected_names = _apply_task_account_filters(
|
||
_visible_task_accounts_query(db, current),
|
||
search=search,
|
||
tag=tag,
|
||
account_names=account_names,
|
||
)
|
||
account_ids = [
|
||
account_id
|
||
for (account_id,) in _order_task_accounts(
|
||
query.enable_eagerloads(False).with_entities(Account.id),
|
||
selected_names,
|
||
).all()
|
||
]
|
||
return {"account_ids": account_ids, "total": len(account_ids)}
|
||
|
||
|
||
@router.get("/workbench-accounts")
|
||
def list_workbench_accounts(
|
||
handbook_scope: str = Query(..., pattern="^(elite|esports|peace)$"),
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(get_current_user),
|
||
):
|
||
"""返回当前用户在指定工作台启用的账号，供不同浏览器同步。"""
|
||
rows = (
|
||
db.query(DouyuWorkbenchAccount.account_id)
|
||
.join(Account, Account.id == DouyuWorkbenchAccount.account_id)
|
||
.filter(
|
||
DouyuWorkbenchAccount.user_id == current.id,
|
||
DouyuWorkbenchAccount.handbook_scope == handbook_scope,
|
||
)
|
||
)
|
||
if not _can_view_all(current):
|
||
rows = rows.filter(Account.assigned_to == current.id)
|
||
account_ids = [
|
||
account_id
|
||
for (account_id,) in rows.order_by(DouyuWorkbenchAccount.id.asc()).all()
|
||
]
|
||
configured = (
|
||
db.query(DouyuWorkbench.id)
|
||
.filter(
|
||
DouyuWorkbench.user_id == current.id,
|
||
DouyuWorkbench.handbook_scope == handbook_scope,
|
||
)
|
||
.first()
|
||
is not None
|
||
)
|
||
return {"account_ids": account_ids, "configured": configured}
|
||
|
||
|
||
@router.put("/workbench-accounts")
|
||
def update_workbench_accounts(
|
||
req: DouyuWorkbenchAccountsUpdate,
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""用当前完整账号集合覆盖一个工作台，作为跨浏览器的同步状态。"""
|
||
account_ids = sorted(set(req.account_ids))
|
||
if any(account_id < 1 for account_id in account_ids):
|
||
raise HTTPException(status_code=400, detail="无效的账号ID")
|
||
if account_ids:
|
||
_require_task_account_access(db, current, account_ids)
|
||
workbench = (
|
||
db.query(DouyuWorkbench)
|
||
.filter(
|
||
DouyuWorkbench.user_id == current.id,
|
||
DouyuWorkbench.handbook_scope == req.handbook_scope,
|
||
)
|
||
.first()
|
||
)
|
||
if workbench is None:
|
||
db.add(DouyuWorkbench(user_id=current.id, handbook_scope=req.handbook_scope))
|
||
else:
|
||
workbench.updated_at = datetime.now(UTC)
|
||
db.query(DouyuWorkbenchAccount).filter(
|
||
DouyuWorkbenchAccount.user_id == current.id,
|
||
DouyuWorkbenchAccount.handbook_scope == req.handbook_scope,
|
||
).delete(synchronize_session=False)
|
||
db.add_all(
|
||
[
|
||
DouyuWorkbenchAccount(
|
||
user_id=current.id,
|
||
handbook_scope=req.handbook_scope,
|
||
account_id=account_id,
|
||
)
|
||
for account_id in account_ids
|
||
]
|
||
)
|
||
db.commit()
|
||
return {"account_ids": account_ids, "success": True}
|
||
|
||
|
||
@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.get("/recharge-channel")
|
||
def get_recharge_channel(
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""返回当前鱼翅充值渠道，不暴露供应商商品等配置细节。"""
|
||
config = ensure_douyu_config(db)
|
||
return {
|
||
"gold_recharge_channel": douyu_config_value(
|
||
"gold_recharge_channel",
|
||
config.gold_recharge_channel,
|
||
),
|
||
}
|
||
|
||
|
||
@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)
|
||
changed_fields = []
|
||
for field in DOUYU_CONFIG_FIELDS:
|
||
value = getattr(req, field, None)
|
||
if value is None:
|
||
continue
|
||
setattr(config, field, value.strip() if isinstance(value, str) else value)
|
||
changed_fields.append(field)
|
||
apply_douyu_config_defaults(config)
|
||
config.updated_at = datetime.now(UTC)
|
||
recharge_fields = [field for field in changed_fields if field.startswith("gold_")]
|
||
if recharge_fields:
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="recharge:douyu:config",
|
||
target="douyu_config",
|
||
detail={"changed_fields": recharge_fields},
|
||
)
|
||
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.get("/esports-goods", response_model=list[DouyuGoodsOut])
|
||
def list_esports_goods(
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""查看已缓存的电竞手册皮肤商城商品。"""
|
||
rows = (
|
||
db.query(DouyuEsportsGoodsSnapshot)
|
||
.order_by(DouyuEsportsGoodsSnapshot.id.asc())
|
||
.all()
|
||
)
|
||
return rows
|
||
|
||
|
||
@router.get("/xpd-goods", response_model=list[DouyuXpdGoodsOut])
|
||
def list_xpd_goods(
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""查看已缓存的和平小店商品。"""
|
||
rows = (
|
||
db.query(DouyuXpdGoodsSnapshot).order_by(DouyuXpdGoodsSnapshot.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="请选择斗鱼账号")
|
||
_require_task_account_access(db, current, req.account_ids)
|
||
|
||
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,
|
||
req.handbook_scope,
|
||
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"
|
||
)
|
||
|
||
if req.task_type == "create_gold_qr":
|
||
config = ensure_douyu_config(db)
|
||
recharge_channel = str(
|
||
douyu_config_value("gold_recharge_channel", config.gold_recharge_channel)
|
||
).strip()
|
||
payment_method = (
|
||
"API 直充支付" if recharge_channel == "supplier_api" else "微信扫码支付"
|
||
)
|
||
recharge_accounts = (
|
||
db.query(DouyuTask, Account)
|
||
.join(Account, Account.id == DouyuTask.account_id)
|
||
.filter(
|
||
DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr"
|
||
)
|
||
.order_by(DouyuTask.id.asc())
|
||
.limit(100)
|
||
.all()
|
||
)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="recharge:douyu:create",
|
||
target=f"douyu_batch:{batch_id}",
|
||
detail={
|
||
"batch_id": batch_id,
|
||
"task_type": req.task_type,
|
||
"count": count,
|
||
"account_count": len(req.account_ids),
|
||
"handbook_scope": req.handbook_scope,
|
||
"recharge_channel": recharge_channel,
|
||
"payment_method": payment_method,
|
||
# 仅记录充值身份，禁止在审计中写入 Cookie、密码等凭据。
|
||
"recharge_accounts": [
|
||
{
|
||
"task_id": task.id,
|
||
"username": account.username,
|
||
"douyu_uid": account.uid or "",
|
||
"douyu_nickname": account.nickname or "",
|
||
}
|
||
for task, account in recharge_accounts
|
||
],
|
||
"recharge_accounts_truncated": count > len(recharge_accounts),
|
||
},
|
||
)
|
||
db.commit()
|
||
|
||
start_douyu_batch(
|
||
batch_id=batch_id,
|
||
task_type=req.task_type,
|
||
payload=req.payload,
|
||
concurrency=req.concurrency,
|
||
)
|
||
|
||
return {"batch_id": batch_id, "count": count, "success": True}
|
||
|
||
|
||
@router.get("/tasks")
|
||
def list_tasks(
|
||
batch_id: str | None = None,
|
||
handbook_scope: str | None = Query(None, pattern="^(elite|esports|peace)$"),
|
||
include_detail: bool = Query(
|
||
False, description="是否返回完整任务结果（默认否，轮询请保持 false）"
|
||
),
|
||
page: int | None = Query(None, ge=1),
|
||
page_size: int = Query(100, ge=1, le=500),
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""查看斗鱼任务记录，默认只返回最近 100 条。"""
|
||
query = _visible_tasks_query(db, current)
|
||
if batch_id:
|
||
query = query.filter(DouyuTask.batch_id == batch_id)
|
||
if handbook_scope:
|
||
# 旧任务没有归属字段，按历史任务类型继续展示，但绝不会触发自动二维码弹窗。
|
||
query = query.filter(
|
||
or_(
|
||
DouyuTask.handbook_scope == handbook_scope,
|
||
(DouyuTask.handbook_scope == "legacy")
|
||
& DouyuTask.task_type.in_(DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]),
|
||
)
|
||
)
|
||
total = None
|
||
if page is not None:
|
||
total = (
|
||
query.enable_eagerloads(False)
|
||
.order_by(None)
|
||
.with_entities(func.count(DouyuTask.id))
|
||
.scalar()
|
||
or 0
|
||
)
|
||
query = query.order_by(DouyuTask.id.desc())
|
||
if page is not None:
|
||
query = query.offset((page - 1) * page_size).limit(page_size)
|
||
else:
|
||
query = query.limit(page_size)
|
||
result = [_task_out(task, include_detail=include_detail) for task in query.all()]
|
||
if page is not None:
|
||
return {
|
||
"items": result,
|
||
"total": total or 0,
|
||
"page": page,
|
||
"page_size": page_size,
|
||
}
|
||
return result
|
||
|
||
|
||
@router.post("/tasks/cleanup-orphans")
|
||
def cleanup_orphan_tasks(
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""手动清理没有内存执行器接管的斗鱼任务。"""
|
||
cleaned = cleanup_orphan_douyu_tasks(
|
||
db,
|
||
active_batch_ids=douyu_batch_registry.active_ids(),
|
||
statuses=("pending", "running"),
|
||
message="任务已中断（无执行器接管）",
|
||
)
|
||
return {
|
||
"message": f"已清理 {cleaned} 个残留任务",
|
||
"cleaned": cleaned,
|
||
"success": True,
|
||
}
|
||
|
||
|
||
@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 = _visible_tasks_query(db, current).filter(DouyuTask.id == task_id).first()
|
||
if not task:
|
||
raise HTTPException(status_code=404, detail="任务不存在")
|
||
return _task_out(task, include_detail=True)
|
||
|
||
|
||
@router.post("/stop/{batch_id}")
|
||
def stop_batch(
|
||
batch_id: str,
|
||
db: Session = Depends(get_db),
|
||
current: User = Depends(require_permission("douyu:task")),
|
||
):
|
||
"""停止正在运行的斗鱼批次。"""
|
||
_require_batch_owner(db, current, batch_id)
|
||
is_recharge_batch = (
|
||
db.query(DouyuTask.id)
|
||
.filter(DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr")
|
||
.first()
|
||
is not None
|
||
)
|
||
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()
|
||
if is_recharge_batch:
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="recharge:douyu:stop",
|
||
target=f"douyu_batch:{batch_id}",
|
||
detail={"batch_id": batch_id, "mode": "running"},
|
||
)
|
||
db.commit()
|
||
return {"message": "已发送停止信号", "success": True}
|
||
|
||
cleaned = cleanup_orphan_douyu_tasks(
|
||
db, batch_id=batch_id, message="任务已停止（批次不存在）"
|
||
)
|
||
if cleaned:
|
||
if is_recharge_batch:
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="recharge:douyu:stop",
|
||
target=f"douyu_batch:{batch_id}",
|
||
detail={
|
||
"batch_id": batch_id,
|
||
"mode": "orphan_cleanup",
|
||
"cleaned": cleaned,
|
||
},
|
||
)
|
||
db.commit()
|
||
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
|
||
if not user_has_permission(user, "douyu:task"):
|
||
await websocket.close(code=1008, reason="无权限")
|
||
return
|
||
db = SessionLocal()
|
||
try:
|
||
_require_batch_owner(db, user, batch_id)
|
||
except HTTPException:
|
||
await websocket.close(code=1008, reason="无权访问该任务批次")
|
||
return
|
||
finally:
|
||
db.close()
|
||
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 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)
|
||
|
||
|
||
def _to_int_or_none(value: object) -> int | None:
|
||
"""回调里的 `buy_num` 可能是字符串/Decimal，安全转 int。"""
|
||
try:
|
||
return int(float(str(value)))
|
||
except (TypeError, ValueError):
|
||
return None
|