yml2213
18bc7be4b1
在既有「管理员分配 → 客服导入工作台 → 手动选号」之外**新增**一条 「客服选皮肤 + 数量 → 系统分配账号 + 按需补积分 → 用完自动收回」的路径。 旧路径只做并集与可选参数扩展，行为不变（有用例守着）。 - 归属与使用权分离：新增 account_leases（生成列 + 唯一索引，并发抢号由数据库裁决）， assigned_to 收窄为管理员/运营的维护责任；客服可见范围扩为「名下 ∪ 租用中」 - account_pool 服务：候选筛选、两梯队排序（先花快过期的 / 先选充值少的）、 供给与成本预估、租约领取/释放/回收 - 补给链：查余额 → 充鱼翅（单次 ≤800 元，超限拆单）→ 赠精英令 → 兑换； 运行期两道闸门：_skip_when（余额已够就别再花钱）、_depends_on（前置未成功就跳过） - 转人工：鱼翅支付等 10 分钟未到账转 manual_required，租约先不还； account_lease_reaper 兜 TTL 到期 / 漏释放 / 孤儿批次 - 运营流水：业务事件补 actor_id = task.created_by；账号详情页展示操作人、渠道 与租赁记录 - 新工作台：/douyu/pool + 菜单「账号分配台」（新建，不改造旧工作台） - 修复 D1：官方补录的兑换事件漏写 subject_key（10,612 条主键为空）， 会让 365 天同皮肤规则漏判 87% 的历史；补代码 + 回填脚本 - 迁移 20261007_0052；权限 account:pool_use / account:pool_admin 测试：766 passed, 6 skipped（改动前基线 727）；ruff 与 tsc -b 干净。 设计稿与实施记录：docs/设计-账号自动分配与回收.md
494 lines
21 KiB
Python
494 lines
21 KiB
Python
"""斗鱼任务执行器：鱼翅充值（由 douyu_runner.py 按功能域拆分）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import time
|
||
from datetime import UTC, datetime
|
||
from decimal import Decimal
|
||
from typing import TYPE_CHECKING, Any, cast
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from core.douyu import (
|
||
DouyuActivityClient,
|
||
FishFinRechargeClient,
|
||
FishFinRechargeConfig,
|
||
FishFinRechargeError,
|
||
)
|
||
|
||
from ..models import Account, DouyuTask
|
||
from .account_pool import validate_recharge_amount
|
||
from .business_events import record_recharge, task_actor
|
||
from .douyu_runner_core import (
|
||
DOUYU_PAYMENT_POLL_INTERVAL,
|
||
DOUYU_PAYMENT_POLL_SECONDS,
|
||
MANUAL_REQUIRED_STATUS,
|
||
)
|
||
from .douyu_service import update_account_profile_from_cookie
|
||
|
||
|
||
class GoldMixin:
|
||
"""鱼翅充值域：扫码充值、供应商直充与到账轮询。"""
|
||
|
||
if TYPE_CHECKING:
|
||
_stop: Any
|
||
|
||
@staticmethod
|
||
def _to_int(value: Any) -> int | None: ...
|
||
|
||
def _client(self, cookie: str) -> DouyuActivityClient: ...
|
||
def _task_payload(self, task: DouyuTask) -> dict: ...
|
||
def _push_log(self, level: str, message: str) -> None: ...
|
||
def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
|
||
def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ...
|
||
|
||
def _refresh_account_gold_balance(
|
||
self, client: DouyuActivityClient, account: Account
|
||
) -> dict:
|
||
"""刷新鱼翅和钱包兑换余额并写回账号表。"""
|
||
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.updated_at = datetime.now(UTC)
|
||
return {
|
||
"gold_balance": account.gold_balance,
|
||
"exchange_balance": account.exchange_balance,
|
||
"gold": gold,
|
||
"exchange_balance_query": exchange,
|
||
}
|
||
|
||
def _wait_gold_balance_after_payment(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
client: DouyuActivityClient,
|
||
result: dict,
|
||
baseline_gold: int | None,
|
||
) -> bool:
|
||
"""等待鱼翅充值到账；余额变化后写回账号表。"""
|
||
deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
|
||
result["payment_polling"] = True
|
||
result["baseline_gold_balance"] = baseline_gold
|
||
poll_count = 0
|
||
last_gold = baseline_gold
|
||
baseline_ready = baseline_gold is not None
|
||
while not self._stop.is_set() and time.monotonic() <= deadline:
|
||
try:
|
||
balance_result = self._refresh_account_gold_balance(client, account)
|
||
db.commit()
|
||
poll_count += 1
|
||
last_gold = balance_result["gold_balance"]
|
||
result.update(balance_result)
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_polling"] = True
|
||
if not baseline_ready and last_gold is not None:
|
||
baseline_gold = last_gold
|
||
result["baseline_gold_balance"] = baseline_gold
|
||
baseline_ready = True
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"鱼翅支付码已生成，已记录当前余额 {last_gold}，等待到账",
|
||
result,
|
||
)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
continue
|
||
# ⚠️ 判据必须是 `>`（余额**增加**）而不是 `!=`。
|
||
# 原来用 `!=`，于是余额**下降**也算「到账」：批内并发（concurrency
|
||
# 最高 10）时同账号的送礼任务会消耗鱼翅，余额一变就被这个充值任务
|
||
# 读成「鱼翅已到账」，往台账里记一笔根本没发生的充值。
|
||
changed = last_gold is not None and (
|
||
baseline_gold is None or last_gold > baseline_gold
|
||
)
|
||
if changed:
|
||
result["payment_polling"] = False
|
||
result["gold_recharged"] = True
|
||
return True
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"鱼翅支付码已生成，等待到账（当前鱼翅 {last_gold if last_gold is not None else '-'}）",
|
||
result,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
poll_count += 1
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_poll_error"] = str(exc)
|
||
self._update_task_progress(
|
||
db, task, "running", f"等待鱼翅到账: {exc}", result
|
||
)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
result["payment_polling"] = False
|
||
result["gold_recharged"] = False
|
||
result["gold_balance"] = last_gold
|
||
return False
|
||
|
||
@staticmethod
|
||
def _supplier_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
|
||
|
||
@classmethod
|
||
def _supplier_order_status(cls, payload: dict) -> int | None:
|
||
"""提取供应商订单状态，文档约定 0-4。"""
|
||
return cls._to_int(
|
||
cls._supplier_value(
|
||
payload, "order_status", "orderStatus", "supplier_order_status"
|
||
)
|
||
)
|
||
|
||
@classmethod
|
||
def _supplier_message(cls, payload: dict) -> str:
|
||
"""提取供应商可展示的业务消息。"""
|
||
value = cls._supplier_value(payload, "msg", "message", "error_msg")
|
||
return str(value or "")[:256]
|
||
|
||
@staticmethod
|
||
def _supplier_result(payload: dict) -> dict:
|
||
"""保存必要订单状态，避免把完整供应商响应或签名暴露到任务结果。"""
|
||
data_raw = payload.get("data")
|
||
data = cast(dict[str, Any], data_raw) if isinstance(data_raw, dict) else {}
|
||
result_raw = payload.get("result")
|
||
response_result = (
|
||
cast(dict[str, Any], result_raw) if isinstance(result_raw, dict) else {}
|
||
)
|
||
result = {
|
||
key: value
|
||
for key, value in {**payload, **data, **response_result}.items()
|
||
if key not in {"sign", "cards", "card_no", "card_pwd", "recharge_arg"}
|
||
}
|
||
return result
|
||
|
||
@staticmethod
|
||
def _supplier_out_order_id(task: DouyuTask) -> str:
|
||
"""生成可追踪的供应商外部订单号；已有订单号必须在重试时复用。"""
|
||
existing = str(task.supplier_out_order_id or "").strip()
|
||
if existing:
|
||
return existing
|
||
batch_token = (
|
||
re.sub(r"[^A-Za-z0-9]", "", str(task.batch_id or "")).upper()[:16]
|
||
or "LOCAL"
|
||
)
|
||
return f"DYGF{batch_token}T{task.id}"
|
||
|
||
def _wait_supplier_gold_order(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
client: FishFinRechargeClient,
|
||
result: dict,
|
||
) -> int | None:
|
||
"""轮询供应商直充订单至结束状态。"""
|
||
order_no = str(result["out_order_id"])
|
||
deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
|
||
poll_count = 0
|
||
result["payment_polling"] = True
|
||
while not self._stop.is_set() and time.monotonic() <= deadline:
|
||
try:
|
||
# 回调可能已在另一个数据库会话中结束订单，刷新后直接使用其结果。
|
||
db.refresh(task)
|
||
if task.status in {"success", "failed"}:
|
||
callback_result = (
|
||
task.result if isinstance(task.result, dict) else result
|
||
)
|
||
result.update(callback_result)
|
||
result["payment_polling"] = False
|
||
return self._supplier_order_status(callback_result)
|
||
payload = client.query_order(order_no)
|
||
code = self._to_int(self._supplier_value(payload, "code"))
|
||
status = self._supplier_order_status(payload)
|
||
poll_count += 1
|
||
result.update(
|
||
{
|
||
"payment_poll_count": poll_count,
|
||
"supplier_code": code,
|
||
"supplier_order_status": status,
|
||
"supplier_order": self._supplier_result(payload),
|
||
}
|
||
)
|
||
if code != 200:
|
||
result["payment_polling"] = False
|
||
return status if status in {2, 3, 4} else 4
|
||
if status in {2, 3, 4}:
|
||
result["payment_polling"] = False
|
||
return status
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"供应商直充订单处理中（状态 {status if status is not None else '-'}）",
|
||
result,
|
||
)
|
||
except FishFinRechargeError as exc:
|
||
poll_count += 1
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_poll_error"] = str(exc)
|
||
self._update_task_progress(
|
||
db, task, "running", f"查询供应商订单失败: {exc}", result
|
||
)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
result["payment_polling"] = False
|
||
return None
|
||
|
||
def _execute_create_gold_qr(
|
||
self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
|
||
):
|
||
payload = self._task_payload(task)
|
||
# 单次充值上限（元）。**必须服务端校验**：前端限制只是体验，
|
||
# 任务可以绕过界面直接建（`POST /api/douyu/tasks/batch` 的 payload 是自由字典）。
|
||
try:
|
||
amount = validate_recharge_amount(
|
||
int(payload.get("amount") or payload.get("gold_amount") or 1)
|
||
)
|
||
except ValueError as exc:
|
||
self._mark_task(db, task, "failed", str(exc))
|
||
return
|
||
channel = str(config.get("gold_recharge_channel") or "wechat_qr")
|
||
if channel == "supplier_api":
|
||
try:
|
||
self._execute_create_gold_supplier_order(
|
||
db, task, account, cookie, config, amount
|
||
)
|
||
except FishFinRechargeError as exc:
|
||
self._mark_task(
|
||
db, task, "failed", str(exc), {"recharge_channel": "supplier_api"}
|
||
)
|
||
return
|
||
client = self._client(cookie)
|
||
baseline_gold = account.gold_balance
|
||
try:
|
||
baseline = self._refresh_account_gold_balance(client, account)
|
||
baseline_gold = baseline["gold_balance"]
|
||
db.commit()
|
||
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
self._push_log("warning", f"生成鱼翅码前刷新余额失败: {exc}")
|
||
result = client.create_gold_qr(
|
||
amount=amount, pay_type=int(config["gold_pay_type"])
|
||
)
|
||
account.updated_at = datetime.now(UTC)
|
||
self._update_task_progress(
|
||
db, task, "running", f"鱼翅 {amount} 元支付码已生成，等待到账", result
|
||
)
|
||
recharged = self._wait_gold_balance_after_payment(
|
||
db, task, account, client, result, baseline_gold
|
||
)
|
||
if self._stop.is_set():
|
||
self._mark_task(db, task, "stopped", "任务已停止", result)
|
||
return
|
||
if recharged:
|
||
account.updated_at = datetime.now(UTC)
|
||
# 金额用**本地变量 amount**：二维码链路的 result 里没有金额
|
||
# （create_gold_qr 只返回 pay_url/token/raw），旧收集器在这里恒写
|
||
# amount=None，导致 accounts.recharge_total 系统性少计。
|
||
record_recharge(
|
||
db,
|
||
account,
|
||
task_id=task.id,
|
||
**task_actor(task),
|
||
amount=amount,
|
||
channel=channel,
|
||
balance_after=account.gold_balance,
|
||
confirmed=True,
|
||
)
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"success",
|
||
f"鱼翅已到账，当前余额: {account.gold_balance if account.gold_balance is not None else '-'}",
|
||
result,
|
||
)
|
||
return
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
MANUAL_REQUIRED_STATUS,
|
||
f"等待 {DOUYU_PAYMENT_POLL_SECONDS // 60} 分钟未检测到鱼翅到账，已转人工处理"
|
||
f"（当前余额: {account.gold_balance if account.gold_balance is not None else '-'}）",
|
||
result,
|
||
)
|
||
|
||
def _execute_create_gold_supplier_order(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
cookie: str,
|
||
config: dict,
|
||
amount: int,
|
||
) -> None:
|
||
"""创建供应商鱼翅直充订单并轮询订单状态。"""
|
||
product_id = str(config.get("gold_api_product_id") or "").strip()
|
||
template_name = str(
|
||
config.get("gold_api_account_template_name") or "斗鱼昵称"
|
||
).strip()
|
||
if not product_id:
|
||
raise FishFinRechargeError("请先在配置中填写供应商直充商品 ID")
|
||
# 充值商品按斗鱼昵称识别账号，UID 只能作为审计信息，不能作为充值值。
|
||
update_account_profile_from_cookie(account, cookie)
|
||
recharge_account = str(account.nickname or "").strip()
|
||
if not recharge_account:
|
||
raise FishFinRechargeError("账号缺少斗鱼昵称，无法发起供应商直充")
|
||
|
||
# 首次生成后持久化，网络重试或进程重启都继续查询同一笔订单。
|
||
order_no = self._supplier_out_order_id(task)
|
||
task.supplier_out_order_id = order_no
|
||
db.commit()
|
||
# pay_amount 是用户选择的充值面值；goodsFaceValue=0.993 是供货成本，不能作为支付金额。
|
||
pay_amount = Decimal(amount)
|
||
|
||
def trace(event: dict) -> None:
|
||
"""将脱敏供应商协议信息输出到任务日志，便于线上联调。"""
|
||
stage = event.get("stage")
|
||
if stage == "request":
|
||
params = event.get("params") or {}
|
||
self._push_log(
|
||
"info",
|
||
"供应商直充 | 下单 "
|
||
f"| 外部单号={params.get('out_order_id') or '-'} "
|
||
f"| 数量={params.get('buy_num') or '-'} "
|
||
f"| 金额={params.get('pay_amount') or '-'} "
|
||
f"| 商品={params.get('product_id') or '-'}",
|
||
)
|
||
if event.get("json_body"):
|
||
self._push_log(
|
||
"debug",
|
||
"供应商协议 | 请求 "
|
||
f"| {event.get('method')} {event.get('path')} "
|
||
f"| 签名摘要={event.get('sign_digest')} "
|
||
f"| 参数={FishFinRechargeClient._json_text(event['json_body'])}",
|
||
)
|
||
elif stage == "response":
|
||
status = self._to_int(event.get("order_status"))
|
||
status_labels = {
|
||
0: "待处理",
|
||
1: "处理中",
|
||
2: "成功",
|
||
3: "失败",
|
||
4: "异常",
|
||
}
|
||
status_text = status_labels.get(
|
||
status if status is not None else -1, "-"
|
||
)
|
||
reason = str(event.get("fail_reason") or event.get("message") or "-")
|
||
self._push_log(
|
||
"info",
|
||
"供应商直充 | 响应 "
|
||
f"| HTTP={event.get('http_status') or '-'} "
|
||
f"| 业务码={event.get('code') or '-'} "
|
||
f"| 外部单号={event.get('out_order_id') or '-'} "
|
||
f"| 供应商单号={event.get('order_id') or '-'} "
|
||
f"| 状态={status_text} "
|
||
f"| 提示={reason}",
|
||
)
|
||
if event.get("response_body"):
|
||
self._push_log(
|
||
"debug",
|
||
"供应商协议 | 响应 "
|
||
f"| HTTP={event.get('http_status')} "
|
||
f"| 内容={FishFinRechargeClient._json_text(event['response_body'])}",
|
||
)
|
||
|
||
client = FishFinRechargeClient(FishFinRechargeConfig.from_env(), trace=trace)
|
||
order_payload = client.create_order(
|
||
buy_num=amount,
|
||
pay_amount=pay_amount,
|
||
out_order_id=order_no,
|
||
product_id=product_id,
|
||
recharge_arg=[
|
||
{"templateName": template_name, "templateVal": recharge_account}
|
||
],
|
||
order_type=0,
|
||
notify_url=client.config.notify_url,
|
||
)
|
||
code = self._to_int(self._supplier_value(order_payload, "code"))
|
||
status = self._supplier_order_status(order_payload)
|
||
result = {
|
||
"recharge_channel": "supplier_api",
|
||
"out_order_id": order_no,
|
||
"order_id": self._supplier_value(order_payload, "order_id", "orderId"),
|
||
"recharge_account": recharge_account,
|
||
"douyu_uid": str(account.uid or "").strip(),
|
||
"buy_num": amount,
|
||
"product_id": product_id,
|
||
"pay_amount": format(pay_amount.normalize(), "f"),
|
||
"order_type": 0,
|
||
"supplier_code": code,
|
||
"supplier_order_status": status,
|
||
"supplier_order": self._supplier_result(order_payload),
|
||
}
|
||
if code != 200:
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"failed",
|
||
self._supplier_message(order_payload) or "供应商创建直充订单失败",
|
||
result,
|
||
)
|
||
return
|
||
account.updated_at = datetime.now(UTC)
|
||
self._update_task_progress(
|
||
db, task, "running", "供应商直充订单已创建，等待到账", result
|
||
)
|
||
if status not in {2, 3, 4}:
|
||
status = self._wait_supplier_gold_order(db, task, client, result)
|
||
if self._stop.is_set():
|
||
self._mark_task(db, task, "stopped", "任务已停止", result)
|
||
return
|
||
if status == 2:
|
||
account.updated_at = datetime.now(UTC)
|
||
# 幂等键与**供应商回调**（routers/douyu.py）完全一致 —— 两条路径都可能
|
||
# 观测到同一笔成功（_wait_supplier_gold_order 会采纳回调结果再 mark 一次），
|
||
# 同一个 source_ref 天然去重。
|
||
record_recharge(
|
||
db,
|
||
account,
|
||
task_id=task.id,
|
||
**task_actor(task),
|
||
amount=amount,
|
||
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=account.gold_balance,
|
||
# 供应商终态成功 ≠ 斗鱼余额已变，如实标注
|
||
confirmed=False,
|
||
)
|
||
self._mark_task(db, task, "success", "供应商直充成功", result)
|
||
return
|
||
if status in {3, 4}:
|
||
self._mark_task(db, task, "failed", "供应商直充失败", result)
|
||
return
|
||
self._mark_task(db, task, "failed", "供应商直充订单查询超时", result)
|
||
|
||
def _execute_query_gold_balance(
|
||
self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
|
||
):
|
||
client = self._client(cookie)
|
||
result = self._refresh_account_gold_balance(client, account)
|
||
account.updated_at = datetime.now(UTC)
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"success",
|
||
f"鱼翅余额: {account.gold_balance if account.gold_balance is not None else '-'}",
|
||
result,
|
||
)
|