2834 lines
101 KiB
Python
2834 lines
101 KiB
Python
"""虎牙外部 CK 协作台账的服务层（模型：``models.HuyaExternalOrder``）。
|
||
|
||
设计见 ``docs/设计-虎牙外部CK协作台账.md``。改这里之前先读文档 §2.3 与 §10.2，
|
||
两条边界不是 bug：
|
||
|
||
1. **不写 ``huya_accounts``、不做外键**：CK 独立存本表；
|
||
2. **不 import 任何任务执行器**（``huya_runner*`` / ``huya_tasks``）：
|
||
绑定/充值/上脚本都在别的工具里做，本表只记录结果。
|
||
S5 的探针只 import ``core.huya.http_client``（只读 RPC），同样不碰 runner。
|
||
|
||
本模块是**两套状态机的唯一写入口**（设计文档 §4.3 防漂移）：
|
||
|
||
- ``set_status()`` —— 业务进度 ``status``
|
||
- ``set_ck_status()`` / ``set_ck()`` / ``rebuild_ck()`` —— 凭据状态 ``ck_status``
|
||
|
||
接口层不允许直接改这两个列，否则状态与时间戳会各自漂移。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import dataclasses
|
||
import hashlib
|
||
import re
|
||
import time
|
||
from collections.abc import Mapping, Sequence
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from loguru import logger
|
||
from sqlalchemy import func, or_, tuple_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..huya_defaults import (
|
||
HUYA_DEFAULT_BIND_ACT_ID,
|
||
HUYA_DEFAULT_OUTER_ACT_ID,
|
||
HUYA_DEFAULT_SID,
|
||
)
|
||
from ..models import (
|
||
HuyaCollabShop,
|
||
HuyaConfig,
|
||
HuyaExternalOrder,
|
||
HuyaExternalOrderMetric,
|
||
User,
|
||
)
|
||
from .audit_service import record_audit
|
||
from .file_service import build_file_url
|
||
from .file_storage import normalize_object_key
|
||
from .huya_service import build_web_cookie_from_raw, parse_huya_cookie_line
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 业务状态机（一单卡在哪一步）
|
||
# --------------------------------------------------------------------------
|
||
|
||
# 「待补资料」是流程链的起点：只登记订单号 + 店铺，CK / 礼包 / 主页截图等客户资料
|
||
# 到了再补（2026-10-17 用户需求）。资料齐全后才能进「待处理」。
|
||
STATUS_INCOMPLETE = "incomplete"
|
||
STATUS_PENDING = "pending"
|
||
STATUS_BOUND = "bound"
|
||
STATUS_RECHARGED = "recharged"
|
||
STATUS_SCRIPTED = "scripted"
|
||
STATUS_DONE = "done"
|
||
STATUS_FAILED = "failed"
|
||
STATUS_VOID = "void"
|
||
|
||
ORDER_STATUSES: tuple[str, ...] = (
|
||
STATUS_INCOMPLETE,
|
||
STATUS_PENDING,
|
||
STATUS_BOUND,
|
||
STATUS_RECHARGED,
|
||
STATUS_SCRIPTED,
|
||
STATUS_DONE,
|
||
STATUS_FAILED,
|
||
STATUS_VOID,
|
||
)
|
||
|
||
ORDER_STATUS_LABELS = {
|
||
STATUS_INCOMPLETE: "待补资料",
|
||
STATUS_PENDING: "待处理",
|
||
# 人工核对过「探针读到的绑定角色」与「主页截图」一致，才允许进这一步（见 _assert_can_enter_bound）
|
||
STATUS_BOUND: "已确认绑定",
|
||
STATUS_RECHARGED: "已充值",
|
||
STATUS_SCRIPTED: "已上脚本",
|
||
# 抢购跑到结果：查兑换记录能确认（自动改）也可以人工切（2026-10-16 用户把「已完成」改叫这个）
|
||
STATUS_DONE: "抢购成功",
|
||
# 旁路终态：抢购后查兑换记录，没查到这一单的礼包 → 判失败（人工确认后可作废或回退重试）
|
||
STATUS_FAILED: "抢购失败",
|
||
STATUS_VOID: "已作废",
|
||
}
|
||
|
||
# 流程链（时间戳同步与「能不能向前走」都只看这条链）；failed / void 是**旁路**终态。
|
||
STATUS_FLOW: tuple[str, ...] = (
|
||
STATUS_INCOMPLETE,
|
||
STATUS_PENDING,
|
||
STATUS_BOUND,
|
||
STATUS_RECHARGED,
|
||
STATUS_SCRIPTED,
|
||
STATUS_DONE,
|
||
)
|
||
|
||
# 「向前一步」的下一站；done 是终点，failed / void 是任意态可达的旁路，都不在这条链上。
|
||
NEXT_STATUS = {
|
||
STATUS_INCOMPLETE: STATUS_PENDING,
|
||
STATUS_PENDING: STATUS_BOUND,
|
||
STATUS_BOUND: STATUS_RECHARGED,
|
||
STATUS_RECHARGED: STATUS_SCRIPTED,
|
||
STATUS_SCRIPTED: STATUS_DONE,
|
||
}
|
||
|
||
# 阶段序号：用来同步时间戳（回退时要清掉后面阶段的时间戳，否则状态与时间戳互相矛盾）。
|
||
STAGE_INDEX = {
|
||
STATUS_INCOMPLETE: 0,
|
||
STATUS_PENDING: 1,
|
||
STATUS_BOUND: 2,
|
||
STATUS_RECHARGED: 3,
|
||
STATUS_SCRIPTED: 4,
|
||
STATUS_DONE: 5,
|
||
}
|
||
|
||
STATUS_TIMESTAMP_FIELDS = {
|
||
STATUS_BOUND: "bound_at",
|
||
STATUS_RECHARGED: "recharged_at",
|
||
STATUS_SCRIPTED: "scripted_at",
|
||
STATUS_DONE: "done_at",
|
||
}
|
||
|
||
|
||
def is_forward_status(from_status: str, target: str) -> bool:
|
||
"""这一步算不算「向前」（人工切换允许往前跳步，回退必须显式声明）。
|
||
|
||
2026-10-16 用户拍板：以前只允许一步一步推进，但人工实际会跳（比如没记充值就直接上脚本）。
|
||
现在规则：
|
||
|
||
- 流程链内：目标阶段序号 **≥ 当前** 就是向前（含原地重复设置，幂等）；
|
||
- `void`（作废）：任意流程内状态都能进；
|
||
- `failed`（抢购失败）：流程内任意状态都能进（兑换记录查出来没有就是没有）；
|
||
- 从 `void` / `failed` 回到流程内：**必须** rollback（这两步是"要人工再确认一次"的）。
|
||
"""
|
||
if target == STATUS_VOID or target == STATUS_FAILED:
|
||
return from_status not in (STATUS_VOID, STATUS_FAILED)
|
||
if from_status in (STATUS_VOID, STATUS_FAILED):
|
||
return False
|
||
return STAGE_INDEX[target] >= STAGE_INDEX[from_status]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# CK 状态机（凭据本身）
|
||
# --------------------------------------------------------------------------
|
||
|
||
CK_UNVERIFIED = "unverified"
|
||
CK_VALID = "valid"
|
||
CK_INVALID = "invalid"
|
||
CK_DISABLED = "disabled"
|
||
|
||
CK_STATUSES: tuple[str, ...] = (CK_UNVERIFIED, CK_VALID, CK_INVALID, CK_DISABLED)
|
||
|
||
CK_STATUS_LABELS = {
|
||
CK_UNVERIFIED: "未验证",
|
||
CK_VALID: "有效",
|
||
CK_INVALID: "已失效",
|
||
CK_DISABLED: "已停用",
|
||
}
|
||
|
||
# 只有这两种状态代表「刚跑过探针」，需要刷新 ck_checked_at。
|
||
_CK_CHECKED_STATUSES = (CK_VALID, CK_INVALID)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 字段与图片槽位
|
||
# --------------------------------------------------------------------------
|
||
|
||
EDITABLE_FIELDS: tuple[str, ...] = (
|
||
"order_no",
|
||
"shop",
|
||
"owner_id",
|
||
"bind_douyu_account",
|
||
"bind_douyu_uid",
|
||
"gift_name",
|
||
"remark",
|
||
)
|
||
|
||
FIELD_LIMITS = {
|
||
"order_no": 64,
|
||
"shop": 64,
|
||
"shop_remark": 255,
|
||
"bind_douyu_account": 128,
|
||
"bind_douyu_uid": 32,
|
||
"gift_name": 128,
|
||
"remark": 512,
|
||
"void_reason": 256,
|
||
}
|
||
|
||
FIELD_LABELS = {
|
||
"order_no": "订单编号",
|
||
"shop": "店铺",
|
||
"shop_remark": "店铺备注",
|
||
"owner_id": "负责人",
|
||
"bind_douyu_account": "绑定斗鱼号",
|
||
"bind_douyu_uid": "绑定斗鱼号 uid",
|
||
"gift_name": "礼包名",
|
||
"remark": "备注",
|
||
"void_reason": "作废原因",
|
||
}
|
||
|
||
IMAGE_SLOTS: tuple[str, ...] = ("bind_images", "home_images")
|
||
|
||
IMAGE_SLOT_LABELS = {
|
||
"bind_images": "绑定图",
|
||
"home_images": "主页截图",
|
||
}
|
||
|
||
MAX_IMAGES_PER_SLOT = 20
|
||
|
||
# 网页运行 CK 才有这两类字段；没有 → 粘贴进来的其实是 App 原始凭据，
|
||
# 顺手存进 raw_cookie，以后就能用「重建」续命（不用重新找人要 CK）。
|
||
_WEB_COOKIE_HINTS = ("guid=", "_qimei_uuid42")
|
||
|
||
|
||
class CollabError(RuntimeError):
|
||
"""协作台账的业务错误，路由层据此转成对应 HTTP 状态码。"""
|
||
|
||
def __init__(self, detail: str, *, status_code: int = 400, code: str = ""):
|
||
super().__init__(detail)
|
||
self.detail = detail
|
||
self.status_code = status_code
|
||
self.code = code
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 内部工具
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(UTC)
|
||
|
||
|
||
def clean_text(
|
||
value: Any, *, field: str, required: bool = False, max_length: int | None = None
|
||
) -> str:
|
||
"""统一收口字段清洗：去空白、限长、必填校验。
|
||
|
||
**不静默截断**：超长直接报错，否则用户填了 100 字备注、库里只剩 512 字节
|
||
之前的片段，事后没人知道被切过。
|
||
"""
|
||
text = str(value if value is not None else "").strip()
|
||
limit = max_length or FIELD_LIMITS.get(field, 255)
|
||
if len(text) > limit:
|
||
raise CollabError(
|
||
f"{FIELD_LABELS.get(field, field)}不能超过 {limit} 个字符",
|
||
code="field_too_long",
|
||
)
|
||
if required and not text:
|
||
raise CollabError(
|
||
f"{FIELD_LABELS.get(field, field)}不能为空", code="field_required"
|
||
)
|
||
return text
|
||
|
||
|
||
def normalize_images(value: Any, *, field: str) -> list[str]:
|
||
"""把图片字段统一成 object_key 数组（去空、去重、挡路径穿越、限张数）。"""
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, str):
|
||
items: Sequence[Any] = [part for part in re.split(r"[,\s]+", value) if part]
|
||
elif isinstance(value, (list, tuple)):
|
||
items = list(value)
|
||
else:
|
||
raise CollabError(
|
||
f"{IMAGE_SLOT_LABELS.get(field, field)}格式不正确", code="images_invalid"
|
||
)
|
||
|
||
result: list[str] = []
|
||
for item in items:
|
||
text = str(item or "").strip()
|
||
if not text:
|
||
continue
|
||
try:
|
||
key = normalize_object_key(text)
|
||
except ValueError as exc:
|
||
raise CollabError(
|
||
f"{IMAGE_SLOT_LABELS.get(field, field)}的文件 key 不正确",
|
||
code="image_key_invalid",
|
||
) from exc
|
||
if key not in result:
|
||
result.append(key)
|
||
|
||
if len(result) > MAX_IMAGES_PER_SLOT:
|
||
raise CollabError(
|
||
f"{IMAGE_SLOT_LABELS.get(field, field)}最多 {MAX_IMAGES_PER_SLOT} 张",
|
||
code="images_too_many",
|
||
)
|
||
return result
|
||
|
||
|
||
def images_of(order: HuyaExternalOrder, slot: str) -> list[str]:
|
||
"""安全读取图片数组（JSON 列历史上可能是 None）。"""
|
||
value = getattr(order, slot, None)
|
||
return [str(item) for item in value] if isinstance(value, list) else []
|
||
|
||
|
||
def look_like_web_cookie(cookie: str) -> bool:
|
||
"""判断一份 CK 是否已是「网页运行态」（含 guid/_qimei_uuid42 设备字段）。"""
|
||
text = cookie or ""
|
||
return any(hint in text for hint in _WEB_COOKIE_HINTS)
|
||
|
||
|
||
def parse_ck(text: str) -> dict:
|
||
"""解析粘贴进来的 CK（复用 huya_service 的纯函数，不依赖账号表）。"""
|
||
parsed = parse_huya_cookie_line(text or "")
|
||
if not parsed:
|
||
raise CollabError(
|
||
"CK 解析失败：没识别到虎牙 uid（支持「纯 CK」或「账号----密码----CK」）",
|
||
code="ck_invalid",
|
||
)
|
||
return parsed
|
||
|
||
|
||
def _touch(order: HuyaExternalOrder, current: User | None) -> None:
|
||
"""每次写入统一推进版本号与最后修改人（乐观锁靠 version，不用 updated_at）。"""
|
||
order.updated_by = current.id if current else None
|
||
order.version = int(order.version or 0) + 1
|
||
order.updated_at = _now()
|
||
|
||
|
||
def _audit_target(order: HuyaExternalOrder) -> str:
|
||
return f"huya_external_order:{order.id}"
|
||
|
||
|
||
def assert_version(order: HuyaExternalOrder, expected_version: int | None) -> None:
|
||
"""乐观锁：客户端提交时带的版本必须与库里一致，否则说明别人先改了。"""
|
||
if expected_version is None:
|
||
return
|
||
if int(expected_version) != int(order.version or 0):
|
||
raise CollabError(
|
||
f"这行已被 {_last_editor_name(order) or '其他人'} 改过（版本 {order.version}），"
|
||
"请刷新后重试；直接覆盖会丢掉对方的修改",
|
||
status_code=409,
|
||
code="version_conflict",
|
||
)
|
||
|
||
|
||
def _last_editor_name(order: HuyaExternalOrder) -> str:
|
||
editor = getattr(order, "updated_by_user", None)
|
||
return str(getattr(editor, "username", "") or "")
|
||
|
||
|
||
def _assert_order_no_available(
|
||
db: Session, order_no: str, *, exclude_id: int | None = None
|
||
) -> None:
|
||
query = db.query(HuyaExternalOrder).filter(HuyaExternalOrder.order_no == order_no)
|
||
if exclude_id is not None:
|
||
query = query.filter(HuyaExternalOrder.id != exclude_id)
|
||
other = query.first()
|
||
if other is not None:
|
||
raise CollabError(
|
||
f"订单编号 {order_no} 已被第 {other.id} 行占用",
|
||
status_code=409,
|
||
code="order_no_conflict",
|
||
)
|
||
|
||
|
||
def uid_used_orders(
|
||
db: Session,
|
||
uid: str | None,
|
||
*,
|
||
exclude_id: int | None = None,
|
||
limit: int = 10,
|
||
) -> list[HuyaExternalOrder]:
|
||
"""这个虎牙号名下**其它**订单（按创建时间升序）。
|
||
|
||
2026-10-16 用户拍板**放开**「一个虎牙号只能挂一单」：同一个买家第二单常常继续用同一份
|
||
CK。所以这里**不再抛 409**（旧行为见设计文档 §1 的变更记录），只用来在界面上提示
|
||
「这份 CK 已经挂在哪些单上」、以及在审计里留一笔复用痕迹。
|
||
"""
|
||
text = str(uid or "").strip()
|
||
if not text:
|
||
return []
|
||
query = (
|
||
db.query(HuyaExternalOrder)
|
||
.filter(HuyaExternalOrder.uid == text)
|
||
.order_by(HuyaExternalOrder.created_at.asc(), HuyaExternalOrder.id.asc())
|
||
)
|
||
if exclude_id is not None:
|
||
query = query.filter(HuyaExternalOrder.id != exclude_id)
|
||
return query.limit(limit).all()
|
||
|
||
|
||
def brief_order(order: HuyaExternalOrder) -> dict[str, Any]:
|
||
"""一行的最小视图（列表/详情里"另一个单"那种展示用）。"""
|
||
return {
|
||
"id": int(order.id),
|
||
"order_no": order.order_no or "",
|
||
"shop": order.shop or "",
|
||
"gift_name": order.gift_name or "",
|
||
"uid": order.uid or "",
|
||
"status": order.status,
|
||
"status_label": ORDER_STATUS_LABELS.get(order.status, order.status),
|
||
"created_at": order.created_at,
|
||
}
|
||
|
||
|
||
def uids_in_use(
|
||
db: Session, uid: str | None, *, exclude_id: int | None = None
|
||
) -> list[dict]:
|
||
"""给预览用的「这个虎牙号已经在哪些单上」（没有就空列表）。"""
|
||
return [
|
||
brief_order(order) for order in uid_used_orders(db, uid, exclude_id=exclude_id)
|
||
]
|
||
|
||
|
||
def _apply_ck(order: HuyaExternalOrder, cookie_text: str) -> dict:
|
||
"""把一份 CK 写进行对象（不含冲突检查与审计，两层调用方共用）。"""
|
||
parsed = parse_ck(cookie_text)
|
||
cookie = str(parsed.get("cookie") or "")
|
||
order.uid = str(parsed.get("uid") or "")
|
||
order.yyuid = str(parsed.get("yyuid") or "")
|
||
order.hy_username = str(parsed.get("username") or "")[:128]
|
||
order.cookie = cookie
|
||
# 粘贴的若是 App 原始凭据（没有网页设备字段），顺手留一份 raw_cookie，
|
||
# 以后被风控标记时能直接「重建」而不必重新要 CK。
|
||
if not look_like_web_cookie(cookie):
|
||
order.raw_cookie = cookie
|
||
order.ck_status = CK_UNVERIFIED
|
||
order.ck_checked_at = None
|
||
order.ck_check_message = ""
|
||
return {
|
||
"uid": order.uid,
|
||
"cookie_kind": "web" if look_like_web_cookie(cookie) else "raw",
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 业务状态机
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def _assert_can_enter_bound(order: HuyaExternalOrder) -> None:
|
||
"""进入「已确认绑定」的前置校验（服务层做，前端校验只是体验）。
|
||
|
||
2026-10-16 用户口径：**人工核对过「探针读到的绑定角色」与「主页截图」一致**，才允许
|
||
切到这一步。所以这里要两样东西齐全：
|
||
|
||
- **绑定角色**（`bind_game_role`，探针写）—— 没读到角色就没人知道该跟哪张图对；
|
||
- **主页截图**（`home_images`，人工上传）—— 没有图就没有"核对"这件事的依据。
|
||
|
||
绑定图已停用（绑定关系改由探针读），不再要求；`bind_douyu_account` 也不再算数
|
||
（那是人工自由文本，且已从界面下线）。
|
||
"""
|
||
missing: list[str] = []
|
||
if not str(order.cookie or "").strip():
|
||
missing.append("CK")
|
||
if not str(order.bind_game_role or "").strip():
|
||
missing.append("绑定角色（先贴 CK 让它自动探针读一次）")
|
||
if not images_of(order, "home_images"):
|
||
missing.append("主页截图（至少 1 张）")
|
||
if missing:
|
||
raise CollabError(
|
||
f"还不能标记为「{ORDER_STATUS_LABELS[STATUS_BOUND]}」，缺少：{'、'.join(missing)}",
|
||
code="bound_precondition",
|
||
)
|
||
|
||
|
||
def _assert_can_leave_incomplete(order: HuyaExternalOrder) -> None:
|
||
"""离开「待补资料」的前置校验：客户资料必须齐全。
|
||
|
||
只在 ``incomplete`` 这个起点用：CK / 礼包名 / 主页截图至少各一份，
|
||
否则「待处理」只是个空壳（后面谁都处理不了）。作废 / 判失败是旁路，不拦。
|
||
"""
|
||
missing: list[str] = []
|
||
if not str(order.cookie or "").strip():
|
||
missing.append("外部 CK")
|
||
if not str(order.gift_name or "").strip():
|
||
missing.append("礼包名")
|
||
if not images_of(order, "home_images"):
|
||
missing.append("主页截图（至少 1 张）")
|
||
if missing:
|
||
raise CollabError(
|
||
f"「{ORDER_STATUS_LABELS[STATUS_INCOMPLETE]}」还不能往前改，缺少：{'、'.join(missing)}",
|
||
code="incomplete_precondition",
|
||
)
|
||
|
||
|
||
def _sync_stage_timestamps(
|
||
order: HuyaExternalOrder, target: str, now: datetime
|
||
) -> None:
|
||
"""让各阶段时间戳与目标状态保持一致。
|
||
|
||
- 目标状态**之前**（含自身）缺失的时间戳补上：状态说「已上脚本」，
|
||
时间戳就不能是空的，否则前端那格显示空白、没人知道什么时候做的；
|
||
- 目标状态**之后**的时间戳清空：回退之后还留着「已上脚本时间」，
|
||
就是状态与时间戳互相矛盾；
|
||
- `void` / `failed` 是旁路终态：不动流程链上的时间戳（从"已上脚本"判失败时，
|
||
"什么时候上的脚本"这条信息必须留着）。
|
||
"""
|
||
if target in (STATUS_VOID, STATUS_FAILED):
|
||
return
|
||
target_index = STAGE_INDEX[target]
|
||
for stage, field in STATUS_TIMESTAMP_FIELDS.items():
|
||
if STAGE_INDEX[stage] <= target_index:
|
||
if getattr(order, field) is None:
|
||
setattr(order, field, now)
|
||
else:
|
||
setattr(order, field, None)
|
||
|
||
|
||
def set_status(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
target: str,
|
||
*,
|
||
reason: str = "",
|
||
rollback: bool = False,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""推进/回退业务状态（本模块唯一的 ``status`` 写入口）。
|
||
|
||
人工切换**允许往前跳步**（如 待处理 → 已上脚本），回退必须显式 ``rollback=True``；
|
||
规则见 :func:`is_forward_status`。
|
||
|
||
Args:
|
||
target: 目标状态，必须在 ``ORDER_STATUSES`` 里。
|
||
reason: 作废原因（``target=void`` 时必填）。
|
||
rollback: 目标不是「向前」时必须显式置 True 才允许改。
|
||
"""
|
||
if target not in ORDER_STATUSES:
|
||
raise CollabError(f"未知状态：{target}", code="status_unknown")
|
||
if target == order.status:
|
||
return {}
|
||
|
||
from_status = order.status
|
||
forward = is_forward_status(from_status, target)
|
||
if not forward:
|
||
if not rollback:
|
||
next_status = NEXT_STATUS.get(from_status)
|
||
hint = (
|
||
f"要往前的话下一步是「{ORDER_STATUS_LABELS[next_status]}」（也可以直接跳到更后面的状态）"
|
||
if next_status
|
||
else "这单已经走到终点"
|
||
)
|
||
raise CollabError(
|
||
f"不能把「{ORDER_STATUS_LABELS.get(from_status, from_status)}」改回"
|
||
f"「{ORDER_STATUS_LABELS[target]}」（{hint}）；确实要回退请显式勾选「回退」",
|
||
code="status_not_forward",
|
||
)
|
||
if from_status == STATUS_VOID:
|
||
raise CollabError(
|
||
"已作废的单子不能改回流程内状态，请编辑或新建",
|
||
code="status_from_void",
|
||
)
|
||
|
||
if from_status == STATUS_INCOMPLETE and target not in (
|
||
STATUS_VOID,
|
||
STATUS_FAILED,
|
||
):
|
||
# 「待补资料」是起点：客户资料没齐之前不许往前，否则下一个环节无从下手。
|
||
_assert_can_leave_incomplete(order)
|
||
|
||
if target == STATUS_BOUND:
|
||
_assert_can_enter_bound(order)
|
||
|
||
now = _now()
|
||
if target == STATUS_VOID:
|
||
order.void_reason = clean_text(reason, field="void_reason", required=True)
|
||
else:
|
||
# 从作废改回流程内状态时，作废原因要清掉，不然列表上会自相矛盾。
|
||
order.void_reason = ""
|
||
|
||
order.status = target
|
||
_sync_stage_timestamps(order, target, now)
|
||
_touch(order, current)
|
||
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:status",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"from": from_status,
|
||
"to": target,
|
||
"rollback": bool(not forward),
|
||
"reason": order.void_reason,
|
||
"order_no": order.order_no,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return {"from": from_status, "to": target}
|
||
|
||
|
||
def set_status_batch(
|
||
db: Session,
|
||
current: User | None,
|
||
orders: Sequence[HuyaExternalOrder],
|
||
target: str,
|
||
*,
|
||
reason: str = "",
|
||
rollback: bool = False,
|
||
) -> dict:
|
||
"""批量改状态：勾选多行一次切到同一个状态（部分失败不影响其它行）。
|
||
|
||
逐行用 SAVEPOINT 隔离（同 :func:`import_orders` 的做法）：某一行因为前置校验没过
|
||
（例如没主页截图不能进「已确认绑定」）只记在这一条的 ``message`` 里，整体不整体失败。
|
||
上限与批量验证一致（``PROBE_BATCH_MAX``）—— 一次别点太多，人会看不过来。
|
||
"""
|
||
rows = list(orders)
|
||
if not rows:
|
||
raise CollabError("请先勾选要改状态的行", code="status_batch_empty")
|
||
if len(rows) > PROBE_BATCH_MAX:
|
||
raise CollabError(
|
||
f"一次最多改 {PROBE_BATCH_MAX} 行，请分批操作",
|
||
code="status_batch_too_large",
|
||
)
|
||
|
||
results: list[dict] = []
|
||
updated = 0
|
||
for order in rows:
|
||
try:
|
||
with db.begin_nested():
|
||
change = set_status(
|
||
db,
|
||
current,
|
||
order,
|
||
target,
|
||
reason=reason,
|
||
rollback=rollback,
|
||
commit=False,
|
||
)
|
||
except CollabError as exc:
|
||
results.append(
|
||
{
|
||
"id": int(order.id),
|
||
"order_no": order.order_no,
|
||
"ok": False,
|
||
"message": exc.detail,
|
||
}
|
||
)
|
||
continue
|
||
if change:
|
||
updated += 1
|
||
results.append(
|
||
{
|
||
"id": int(order.id),
|
||
"order_no": order.order_no,
|
||
"ok": True,
|
||
"message": (
|
||
f"{ORDER_STATUS_LABELS.get(change['from'], change['from'])}"
|
||
f" → {ORDER_STATUS_LABELS.get(change['to'], change['to'])}"
|
||
if change
|
||
else "已经是这个状态"
|
||
),
|
||
}
|
||
)
|
||
db.commit()
|
||
for order in rows:
|
||
db.refresh(order)
|
||
return {
|
||
"total": len(rows),
|
||
"updated": updated,
|
||
"failed": len(rows) - updated,
|
||
"target": target,
|
||
"results": results,
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# CK 写入口
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def set_ck(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
cookie_text: str,
|
||
*,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""贴入 / 换一份 CK，状态回到「未验证」。
|
||
|
||
同一个虎牙号**允许**挂在多张单上（2026-10-16 用户拍板放开 uid 唯一性）：
|
||
复用到的那些单号只记一笔审计，不挡操作 —— 挡了就变成"同一个买家的第二单建不了"。
|
||
"""
|
||
before = str(order.cookie or "")
|
||
detail = _apply_ck(order, cookie_text)
|
||
reused = [
|
||
item.order_no for item in uid_used_orders(db, order.uid, exclude_id=order.id)
|
||
]
|
||
_touch(order, current)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:ck",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"kind": "paste",
|
||
"uid": order.uid,
|
||
"order_no": order.order_no,
|
||
# 刻意**不**把 cookie 交给审计层：即便 record_audit 会脱敏，
|
||
# 凭据也不该多经一手（脱敏规则一改就是凭据泄漏）。
|
||
# 只留一个「内容是否变了」的布尔，够用来追溯换没换过。
|
||
"credential_changed": before != str(order.cookie or ""),
|
||
# 这个虎牙号还挂在哪些单上（同一份 CK 服务多单是允许的，留个痕）
|
||
"uid_reused_in": reused,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return detail
|
||
|
||
|
||
def set_ck_status(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
status: str,
|
||
*,
|
||
message: str = "",
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""改写凭据状态（本模块唯一的 ``ck_status`` 写入口）。
|
||
|
||
``valid`` / ``invalid`` 视为「刚跑过探针」，会刷新 ``ck_checked_at``；
|
||
``disabled`` 是人工停用，不动探针时间。
|
||
"""
|
||
if status not in CK_STATUSES:
|
||
raise CollabError(f"未知 CK 状态：{status}", code="ck_status_unknown")
|
||
previous = order.ck_status
|
||
order.ck_status = status
|
||
order.ck_check_message = clean_text(
|
||
message, field="ck_check_message", max_length=256
|
||
)
|
||
if status in _CK_CHECKED_STATUSES:
|
||
order.ck_checked_at = _now()
|
||
_touch(order, current)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:ck_status",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"from": previous,
|
||
"to": status,
|
||
"message": order.ck_check_message,
|
||
"order_no": order.order_no,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return {"from": previous, "to": status}
|
||
|
||
|
||
def rebuild_ck(
|
||
db: Session, current: User | None, order: HuyaExternalOrder, *, commit: bool = True
|
||
) -> dict:
|
||
"""用 ``raw_cookie`` 重建网页运行 CK（复用 huya_service 的纯函数）。"""
|
||
if not str(order.raw_cookie or "").strip():
|
||
raise CollabError(
|
||
"这行没有 App 原始凭据，无法重建网页 CK（请重新贴一份）",
|
||
code="no_raw_cookie",
|
||
)
|
||
try:
|
||
rebuilt = build_web_cookie_from_raw(
|
||
uid=str(order.uid or ""),
|
||
yyuid=str(order.yyuid or ""),
|
||
username=str(order.hy_username or ""),
|
||
raw_cookie=str(order.raw_cookie or ""),
|
||
)
|
||
except ValueError as exc:
|
||
raise CollabError(f"重建失败：{exc}", code="rebuild_failed") from exc
|
||
|
||
order.cookie = rebuilt
|
||
order.ck_status = CK_UNVERIFIED
|
||
order.ck_checked_at = None
|
||
order.ck_check_message = "已从 App 原始凭据重建，待验证"
|
||
_touch(order, current)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:ck",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"kind": "rebuild",
|
||
"uid": order.uid,
|
||
"order_no": order.order_no,
|
||
"credential_changed": True,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return {"uid": order.uid, "cookie_kind": "web"}
|
||
|
||
|
||
# 人工能改的 CK 状态：`valid` / `invalid` 只能由探针给（人工标「有效」等于绕过验证）。
|
||
MANUAL_CK_STATUSES: tuple[str, ...] = (CK_UNVERIFIED, CK_DISABLED)
|
||
|
||
|
||
def set_manual_ck_status(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
status: str,
|
||
*,
|
||
message: str = "",
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""人工改凭据状态（只允许「已停用」/「未验证」），仍然走 :func:`set_ck_status` 落库。"""
|
||
if status not in MANUAL_CK_STATUSES:
|
||
raise CollabError(
|
||
"人工只能把 CK 标成「已停用」或「未验证」；有效/失效由探针判定",
|
||
code="ck_status_manual_only",
|
||
)
|
||
return set_ck_status(db, current, order, status, message=message, commit=commit)
|
||
|
||
|
||
def set_images(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
slot: str,
|
||
value: Any,
|
||
*,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""整槽替换图片（前端上传完拿到 object_key 后一次性提交）。"""
|
||
if slot not in IMAGE_SLOTS:
|
||
raise CollabError(f"未知图片槽位：{slot}", code="image_slot_unknown")
|
||
before = images_of(order, slot)
|
||
after = normalize_images(value, field=slot)
|
||
if before == after:
|
||
return {"slot": slot, "images": after}
|
||
setattr(order, slot, after)
|
||
_touch(order, current)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:image",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"slot": slot,
|
||
"slot_label": IMAGE_SLOT_LABELS[slot],
|
||
"added": [key for key in after if key not in before],
|
||
"removed": [key for key in before if key not in after],
|
||
"order_no": order.order_no,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return {"slot": slot, "images": after}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# CK 探针（S5，只读验证）
|
||
# --------------------------------------------------------------------------
|
||
|
||
# 探针参数与 web/backend/huya_defaults.py 的默认活动一致（2026-10-16 真机实测）。
|
||
PROBE_BIND_ACT_ID = int(HUYA_DEFAULT_BIND_ACT_ID)
|
||
PROBE_OUTER_ACT_ID = HUYA_DEFAULT_OUTER_ACT_ID
|
||
PROBE_TIMEOUT_SECONDS = 15.0
|
||
|
||
# 探针独占的绑定快照列（每次验证**覆盖写入**，人工不编辑）：
|
||
# 「这单现在到底绑了哪个游戏账号、哪个角色」只认这几列。
|
||
PROBE_SNAPSHOT_FIELDS: tuple[str, ...] = (
|
||
"bind_game_account",
|
||
"bind_game_role",
|
||
"bind_game_channel",
|
||
)
|
||
|
||
# 活动礼包（get_act_prize_list）用的 SID **兜底值**：真正该用的是工作台配置里的那个
|
||
# （``huya_config.sid``，见 :func:`resolve_gifts_sid`）。
|
||
# 2026-10-16 真机定论：**SID 就是活动 ID，会随活动轮换** —— 库里 9 月的活动是 2203，
|
||
# 今天是 2277。写死默认值时接口照样 ``status=200`` 但 **0 个商品**，界面上就是
|
||
# 「礼包列表空白」（换 SID 立刻回到 30 个礼包，跟 CK、UA、解析都无关）。
|
||
GIFTS_SID = int(HUYA_DEFAULT_SID)
|
||
# 一次最多回多少个礼包（避免前端下拉塞几百项）。
|
||
GIFTS_LIMIT = 200
|
||
# 同一份 CK 的探针结论缓存时长（"预览一次 + 提交校验一次" 只打一次接口）。
|
||
PROBE_CACHE_TTL_SECONDS = 90.0
|
||
_probe_cache: dict[str, tuple[float, CkProbeResult]] = {}
|
||
|
||
# 批量验证**串行 + 间隔**：不做定时轮询，一次也别勾太多，避免把账号送进风控名单。
|
||
PROBE_BATCH_INTERVAL_SECONDS = 1.0
|
||
PROBE_BATCH_MAX = 50
|
||
|
||
_CK_MESSAGE_LIMIT = 256 # ck_check_message 列宽
|
||
_PROBE_PIECE_LIMIT = 40 # 拼消息时单段文本的上限
|
||
|
||
|
||
def _clip(value: Any, limit: int = _PROBE_PIECE_LIMIT) -> str:
|
||
"""把探针文本裁到可读长度（``ck_check_message`` 只有 256 字符）。"""
|
||
text = str(value if value is not None else "").strip()
|
||
return text if len(text) <= limit else text[: limit - 3] + "..."
|
||
|
||
|
||
def _probe_cache_key(uid: int, cookie: str) -> str:
|
||
"""探针缓存键：uid + CK 的指纹（只存哈希，不留凭据原文）。"""
|
||
digest = hashlib.sha256(f"{uid}|{cookie}".encode()).hexdigest()
|
||
return digest
|
||
|
||
|
||
@dataclasses.dataclass(frozen=True)
|
||
class CkProbeResult:
|
||
"""一次只读探针的结论（不落库；落库由 :func:`verify_ck` 负责）。"""
|
||
|
||
ok: bool
|
||
message: str
|
||
snapshot: dict[str, str] = dataclasses.field(default_factory=dict)
|
||
|
||
|
||
def _build_probe_client() -> Any:
|
||
"""探针客户端（HTTP cdnws 只读通道）。
|
||
|
||
只做只读 RPC：**不建 ``HuyaAccount``、不 import 任何 runner**（设计文档 §2.3）。
|
||
测试替换这个工厂就能离线跑，不真连虎牙。
|
||
"""
|
||
from core.huya.http_client import HuyaHttpClient
|
||
|
||
return HuyaHttpClient(
|
||
logger=lambda message: logger.debug(f"[huya-collab] {message}")
|
||
)
|
||
|
||
|
||
def _probe_snapshot(resp: Any) -> dict[str, str]:
|
||
"""探针读到的**绑定对象**：绑定在该虎牙号上的游戏账号 + 角色。
|
||
|
||
2026-10-16 真机确认：``check_user_bind_game_account`` 返回的是**游戏账号**
|
||
（账号昵称 / 角色名 / 平台·区服），**不是虎牙号自己的昵称** —— 虎牙昵称要
|
||
`get_grow_level_user_info`，可那个接口不认凭据、当不了探针，所以 ``nickname``
|
||
一个字段都不碰（那是「虎牙号昵称」，硬写就是假数据）。
|
||
|
||
这几个列由探针**独占并覆盖写入**（``bind_snapshot_at`` 记读取时间）：
|
||
"这单现在到底绑了哪个号、哪个角色"只有这里可信（人工填的是自由文本）。
|
||
"""
|
||
account_data = getattr(resp, "accountData", None)
|
||
game_account = getattr(account_data, "gameAccount", None)
|
||
game_role = getattr(account_data, "gameRole", None)
|
||
|
||
snapshot: dict[str, str] = dict.fromkeys(PROBE_SNAPSHOT_FIELDS, "")
|
||
account_name = _clip(getattr(game_account, "nick", "")) or _clip(
|
||
getattr(game_role, "roleName", "")
|
||
)
|
||
if account_name:
|
||
snapshot["bind_game_account"] = account_name
|
||
role_name = _clip(getattr(game_role, "roleName", ""))
|
||
if role_name:
|
||
snapshot["bind_game_role"] = role_name
|
||
channel = " · ".join(
|
||
piece
|
||
for piece in (
|
||
_clip(getattr(game_role, "platName", "")),
|
||
_clip(getattr(game_role, "areaName", "")),
|
||
)
|
||
if piece
|
||
)
|
||
if channel:
|
||
snapshot["bind_game_channel"] = channel
|
||
return snapshot
|
||
|
||
|
||
def _probe_success_message(resp: Any) -> str:
|
||
"""成功时的可读结论（列表上直接能看到探针读到的绑定对象）。"""
|
||
account_data = getattr(resp, "accountData", None)
|
||
game_account = getattr(account_data, "gameAccount", None)
|
||
game_role = getattr(account_data, "gameRole", None)
|
||
|
||
parts: list[str] = []
|
||
nick = _clip(getattr(game_account, "nick", ""))
|
||
role_name = _clip(getattr(game_role, "roleName", ""))
|
||
if nick:
|
||
parts.append(f"账号 {nick}")
|
||
if role_name:
|
||
parts.append(f"角色 {role_name}")
|
||
channel = " · ".join(
|
||
piece
|
||
for piece in (
|
||
_clip(getattr(game_role, "platName", "")),
|
||
_clip(getattr(game_role, "areaName", "")),
|
||
)
|
||
if piece
|
||
)
|
||
if channel:
|
||
parts.append(channel)
|
||
detail = " / ".join(parts)
|
||
return (f"有效：{detail}" if detail else "有效")[:_CK_MESSAGE_LIMIT]
|
||
|
||
|
||
def probe_ck(order: HuyaExternalOrder, *, client: Any | None = None) -> CkProbeResult:
|
||
"""只读探针：这份 CK 还能不能过（不写库、不发敏感动作）。
|
||
|
||
判据（2026-10-16 真机对照确定，**别改成「有数据就算成功」**）：
|
||
响应**非 None 且 ``status == 200``**。
|
||
|
||
实测支持这个判据：裸 cookie（无任何凭据）+ 真 uid → 返回 ``None``；
|
||
真 CK + 别人的 uid → 也返回 ``None``；只有「真 CK + 真 uid」才拿到 200 + 真实数据。
|
||
|
||
同一个 CK 在 ``PROBE_CACHE_TTL_SECONDS`` 秒内复用上次结论：新建弹窗先"预览"、
|
||
提交时再校验，不该真的打两次虎牙接口。**注入 client 时不走缓存**（测试要真结论）。
|
||
"""
|
||
uid_text = str(order.uid or "").strip()
|
||
cookie = str(order.cookie or "").strip()
|
||
if not uid_text or not cookie:
|
||
return CkProbeResult(False, "这行还没有可验证的 CK（uid 或凭据为空）")
|
||
try:
|
||
uid = int(uid_text)
|
||
except ValueError:
|
||
return CkProbeResult(False, f"CK 里的 uid 不是数字：{_clip(uid_text)}")
|
||
|
||
cache_key = _probe_cache_key(uid, cookie) if client is None else ""
|
||
if cache_key:
|
||
cached = _probe_cache.get(cache_key)
|
||
if cached and time.time() - cached[0] < PROBE_CACHE_TTL_SECONDS:
|
||
return cached[1]
|
||
|
||
try:
|
||
probe_client = client or _build_probe_client()
|
||
resp = probe_client.check_user_bind_game_account(
|
||
uid=uid,
|
||
cookie=cookie,
|
||
b_act_id=PROBE_BIND_ACT_ID,
|
||
is_use_outer_act_id=1,
|
||
outer_act_id=PROBE_OUTER_ACT_ID,
|
||
timeout=PROBE_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 探针异常=验证不通过，不能让接口 500
|
||
logger.debug(f"[huya-collab] CK 探针异常：{type(exc).__name__}: {exc}")
|
||
return CkProbeResult(False, f"探针调用失败：{type(exc).__name__}: {_clip(exc)}")
|
||
|
||
if resp is None:
|
||
return CkProbeResult(False, "探针没有返回数据（凭据无效或已被风控）")
|
||
status = int(getattr(resp, "status", 0) or 0)
|
||
if status != 200:
|
||
msg = _clip(getattr(resp, "msg", ""))
|
||
suffix = f"：{msg}" if msg else ""
|
||
return CkProbeResult(False, f"探针返回 status={status}{suffix}")
|
||
result = CkProbeResult(True, _probe_success_message(resp), _probe_snapshot(resp))
|
||
if cache_key:
|
||
_probe_cache[cache_key] = (time.time(), result)
|
||
return result
|
||
|
||
|
||
def resolve_gifts_sid(db: Session) -> int:
|
||
"""当前活动 SID：**优先用工作台配置**（``huya_config.sid``），读不到再回落到默认值。
|
||
|
||
2026-10-16 真机定位：礼包接口的 SID 是「活动 ID」，会随活动轮换（9 月 2203 → 今天 2277），
|
||
对旧 SID 打过去是 ``status=200`` + 0 个商品 —— 界面上就表现为「礼包列表空白」。
|
||
工作台「虎牙配置」里改过 SID 之后，这里跟着变，不用改代码。
|
||
|
||
只读：配置行不存在时**不会**顺手建一行（预览是只读动作，设计文档 §2.3）。
|
||
"""
|
||
try:
|
||
row = db.query(HuyaConfig).first()
|
||
except Exception as exc: # noqa: BLE001 配置读失败不该让预览 500
|
||
logger.debug(
|
||
f"[huya-collab] 读取虎牙活动 SID 失败：{type(exc).__name__}: {exc}"
|
||
)
|
||
return GIFTS_SID
|
||
try:
|
||
value = int(str(getattr(row, "sid", "") or "").strip())
|
||
except (TypeError, ValueError):
|
||
return GIFTS_SID
|
||
return value or GIFTS_SID
|
||
|
||
|
||
def fetch_activity_gifts(
|
||
uid: int,
|
||
cookie: str,
|
||
*,
|
||
sid: int | None = None,
|
||
client: Any | None = None,
|
||
) -> tuple[list[dict], str]:
|
||
"""用这份 CK 拉**活动可兑换礼包**列表（只读，不落库）。
|
||
|
||
``sid`` 不传就用 :data:`GIFTS_SID` 兜底；接口层会传工作台配置里的活动 ID
|
||
（:func:`resolve_gifts_sid`）。
|
||
|
||
Returns:
|
||
``(礼包列表, 可读结论)`` —— 每项是 ``{product_id, name, price, ...}``；
|
||
拉不到时返回 ``([], "…")``，不影响 CK 本身的有效性结论。
|
||
"""
|
||
sid_value = int(sid or GIFTS_SID)
|
||
try:
|
||
gift_client = client or _build_probe_client()
|
||
resp = gift_client.get_act_prize_list(
|
||
uid=int(uid), cookie=cookie, sid=sid_value
|
||
)
|
||
except Exception as exc: # noqa: BLE001 拉礼包失败不该让整个预览 500
|
||
logger.debug(f"[huya-collab] 拉取活动礼包异常：{type(exc).__name__}: {exc}")
|
||
return [], f"拉取礼包失败：{type(exc).__name__}: {_clip(exc)}"
|
||
|
||
if resp is None:
|
||
return [], "拉取礼包失败：接口没有返回数据"
|
||
status = int(getattr(resp, "status", 0) or 0)
|
||
if status != 200:
|
||
msg = _clip(getattr(resp, "msg", ""))
|
||
return [], f"拉取礼包失败：status={status}{('：' + msg) if msg else ''}"
|
||
|
||
raw_goods = (resp.to_dict() or {}).get("goods") or []
|
||
gifts: list[dict] = []
|
||
for item in raw_goods:
|
||
product_id = str(item.get("product_id") or "").strip()
|
||
name = str(item.get("name") or "").strip()
|
||
if not product_id or not name:
|
||
continue
|
||
gifts.append(
|
||
{
|
||
"product_id": product_id,
|
||
"name": name[:256],
|
||
"price": item.get("price"),
|
||
"remain_text": str(item.get("remain_text") or "")[:64],
|
||
"stock": item.get("stock"),
|
||
"score": item.get("score"),
|
||
"category_name": str(item.get("category_name") or "")[:64],
|
||
"icon": str(item.get("icon") or "")[:512],
|
||
}
|
||
)
|
||
gifts = gifts[:GIFTS_LIMIT]
|
||
if not gifts:
|
||
# 200 + 空列表 = 这个 SID 不是当前活动（见 resolve_gifts_sid 的真机定位）。
|
||
return [], (
|
||
f"活动 SID={sid_value} 没返回商品：接口通了但列表是空的，"
|
||
"多半是活动换 ID 了 —— 去工作台「虎牙配置」把活动 SID 改成当前的再试"
|
||
)
|
||
return gifts, f"已拉取 {len(gifts)} 个礼包"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 账号全生命周期（积分 / 兑换记录：抢购之后靠它判断到底兑换成功没有）
|
||
# --------------------------------------------------------------------------
|
||
|
||
ACTIVITY_QUERY_TIMEOUT_SECONDS = 15.0
|
||
|
||
VERDICT_SUCCESS = "success"
|
||
VERDICT_FAILED = "failed"
|
||
VERDICT_UNKNOWN = "unknown"
|
||
|
||
VERDICT_LABELS = {
|
||
VERDICT_SUCCESS: "兑换成功",
|
||
VERDICT_FAILED: "兑换失败",
|
||
VERDICT_UNKNOWN: "无法判定",
|
||
}
|
||
|
||
# metric.raw 里最多留多少条兑换记录（避免 JSON 列被几百条记录撑爆）
|
||
_METRIC_RECORD_LIMIT = 50
|
||
# 兑换记录里跟订单礼包名比对时，只取有意义的长度（太短的名字容易误匹配）
|
||
_MIN_GIFT_MATCH_LENGTH = 2
|
||
|
||
|
||
def _metric_message(text: Any, limit: int = 480) -> str:
|
||
piece = str(text or "").strip()
|
||
return piece if len(piece) <= limit else piece[: limit - 3] + "..."
|
||
|
||
|
||
def gift_matches_records(
|
||
records: Sequence[Mapping[str, Any]], gift_name: str
|
||
) -> list[dict[str, Any]]:
|
||
"""在兑换记录里找这一单的礼包（名字完全相等或互相包含都算）。
|
||
|
||
虎牙记录里的 ``prize_name`` 与活动列表里的名字基本一字不差，但偶尔带后缀/前缀，
|
||
所以用「互相包含」宽松匹配；太短的名字（<2 字）不参与，免得 "礼包" 两个字匹配一堆。
|
||
"""
|
||
wanted = str(gift_name or "").strip()
|
||
if len(wanted) < _MIN_GIFT_MATCH_LENGTH:
|
||
return []
|
||
matched: list[dict[str, Any]] = []
|
||
for record in records:
|
||
name = str(record.get("prize_name") or "").strip()
|
||
if not name:
|
||
continue
|
||
if name == wanted or wanted in name or name in wanted:
|
||
matched.append(dict(record))
|
||
return matched
|
||
|
||
|
||
def _extract_score(resp: Any) -> tuple[int | None, int | None, int | None, str]:
|
||
"""从 ``getUserScore`` 响应里取「可用 / 累计获得 / 已消耗」+ 可读结论。"""
|
||
if resp is None:
|
||
return None, None, None, "积分接口没有返回数据"
|
||
status = int(getattr(resp, "status", 0) or 0)
|
||
if status != 200:
|
||
msg = _clip(getattr(resp, "msg", ""))
|
||
return (
|
||
None,
|
||
None,
|
||
None,
|
||
f"积分接口 status={status}{('：' + msg) if msg else ''}",
|
||
)
|
||
available = getattr(resp, "available_score", None)
|
||
if available is None:
|
||
available = getattr(resp, "score", 0)
|
||
gained = int(getattr(resp, "newGainScore", 0) or 0) or int(
|
||
getattr(resp, "gainScore", 0) or 0
|
||
)
|
||
used = int(getattr(resp, "newUsedScore", 0) or 0) or int(
|
||
getattr(resp, "usedScore", 0) or 0
|
||
)
|
||
return int(available or 0), gained, used, ""
|
||
|
||
|
||
def _extract_records(resp: Any) -> tuple[list[dict[str, Any]], str]:
|
||
"""从 ``getUserPrizeRecords`` 响应里取兑换记录列表 + 可读结论。"""
|
||
if resp is None:
|
||
return [], "兑换记录接口没有返回数据"
|
||
status = int(getattr(resp, "status", 0) or 0)
|
||
if status != 200:
|
||
msg = _clip(getattr(resp, "msg", ""))
|
||
return [], f"兑换记录接口 status={status}{('：' + msg) if msg else ''}"
|
||
data = resp.to_dict() or {}
|
||
return list(data.get("records") or []), ""
|
||
|
||
|
||
def serialize_metric(
|
||
metric: HuyaExternalOrderMetric, *, users: Mapping[int, str] | None = None
|
||
) -> dict:
|
||
"""一条生命周期快照的输出（时间线用）。"""
|
||
names = users or {}
|
||
raw = metric.raw if isinstance(metric.raw, dict) else {}
|
||
matched = raw.get("matched") if isinstance(raw.get("matched"), list) else []
|
||
return {
|
||
"id": int(metric.id),
|
||
"order_id": int(metric.order_id),
|
||
"order_no": metric.order_no or "",
|
||
"uid": metric.uid or "",
|
||
"sid": metric.sid,
|
||
"score": metric.score,
|
||
"gained_score": metric.gained_score,
|
||
"used_score": metric.used_score,
|
||
"records_total": int(metric.records_total or 0),
|
||
"records_matched": int(metric.records_matched or 0),
|
||
"verdict": metric.verdict or VERDICT_UNKNOWN,
|
||
"verdict_label": VERDICT_LABELS.get(metric.verdict, metric.verdict or ""),
|
||
"message": metric.message or "",
|
||
"source": metric.source or "manual",
|
||
"matched_preview": [
|
||
{
|
||
"prize_name": str(item.get("prize_name") or ""),
|
||
"score": item.get("score"),
|
||
"status_label": str(item.get("status_label") or ""),
|
||
"order_id": str(item.get("order_id") or ""),
|
||
"exchange_time": item.get("exchange_time"),
|
||
}
|
||
for item in matched[:5]
|
||
],
|
||
"created_by": metric.created_by,
|
||
"created_by_name": names.get(int(metric.created_by or 0), ""),
|
||
"created_at": metric.created_at,
|
||
}
|
||
|
||
|
||
def list_metrics(
|
||
db: Session, order: HuyaExternalOrder, *, limit: int = 100
|
||
) -> list[HuyaExternalOrderMetric]:
|
||
"""某个订单的生命周期快照（**倒序**：最近一次在前，界面直接铺时间线）。"""
|
||
return (
|
||
db.query(HuyaExternalOrderMetric)
|
||
.filter(HuyaExternalOrderMetric.order_id == int(order.id))
|
||
.order_by(HuyaExternalOrderMetric.id.desc())
|
||
.limit(max(1, min(500, int(limit))))
|
||
.all()
|
||
)
|
||
|
||
|
||
def latest_metrics(
|
||
db: Session, orders: Sequence[HuyaExternalOrder]
|
||
) -> dict[int, HuyaExternalOrderMetric]:
|
||
"""一次查出这批行各自「最近一条」快照（列表要显示最新积分，别按行发 SQL）。"""
|
||
ids = [int(order.id) for order in orders]
|
||
if not ids:
|
||
return {}
|
||
newest = (
|
||
db.query(func.max(HuyaExternalOrderMetric.id).label("metric_id"))
|
||
.filter(HuyaExternalOrderMetric.order_id.in_(ids))
|
||
.group_by(HuyaExternalOrderMetric.order_id)
|
||
.subquery()
|
||
)
|
||
metric_ids = [row[0] for row in db.query(newest.c.metric_id).all()]
|
||
if not metric_ids:
|
||
return {}
|
||
rows = (
|
||
db.query(HuyaExternalOrderMetric)
|
||
.filter(HuyaExternalOrderMetric.id.in_(metric_ids))
|
||
.all()
|
||
)
|
||
return {int(row.order_id): row for row in rows}
|
||
|
||
|
||
def query_order_activity(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
*,
|
||
client: Any | None = None,
|
||
source: str = "manual",
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""查一次这个账号的**积分 + 兑换记录**，落一条生命周期快照，必要时自动改状态。
|
||
|
||
判定规则（2026-10-16 用户口径）：
|
||
|
||
| 情况 | verdict | 业务状态 |
|
||
| --- | --- | --- |
|
||
| 兑换记录里有这一单的礼包 | ``success`` | 「已上脚本」的单 → 自动「抢购成功」 |
|
||
| 接口通了，但记录里没有这个礼包 | ``failed`` | 「已上脚本」的单 → 自动「抢购失败」 |
|
||
| 接口没通 / 这单没填礼包名 | ``unknown`` | **不动**（数据不够，不许瞎判） |
|
||
|
||
只有 ``已上脚本`` 的单会被自动改状态 —— 抢购是上脚本之后的事，其它阶段去改属于误判；
|
||
被跳过时结论里会写清楚原因。每次调用都写一条 metric（哪怕接口失败），
|
||
「积分变化 + 兑换记录」两条线索合起来才看得清有没有真的抢到。
|
||
"""
|
||
if order.status == STATUS_VOID:
|
||
raise CollabError("已作废的单子不查积分/兑换记录", code="void_order")
|
||
uid = str(order.uid or "").strip()
|
||
cookie = str(order.cookie or "").strip()
|
||
if not uid or not cookie:
|
||
raise CollabError("这行还没有 CK，先贴一份再查积分/兑换记录", code="no_cookie")
|
||
|
||
sid = resolve_gifts_sid(db)
|
||
activity_client = client or _build_probe_client()
|
||
score_resp: Any = None
|
||
records_resp: Any = None
|
||
errors: list[str] = []
|
||
try:
|
||
score_resp = activity_client.query_user_score(
|
||
uid=int(uid), cookie=cookie, sid=sid, timeout=ACTIVITY_QUERY_TIMEOUT_SECONDS
|
||
)
|
||
except Exception as exc: # noqa: BLE001 一个接口炸了不影响另一个
|
||
errors.append(f"积分接口异常：{type(exc).__name__}: {_clip(exc)}")
|
||
try:
|
||
records_resp = activity_client.get_user_prize_records(
|
||
uid=int(uid), cookie=cookie, sid=sid, timeout=ACTIVITY_QUERY_TIMEOUT_SECONDS
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
errors.append(f"兑换记录接口异常：{type(exc).__name__}: {_clip(exc)}")
|
||
|
||
score, gained, used, score_msg = _extract_score(score_resp)
|
||
records, records_msg = _extract_records(records_resp)
|
||
matched = gift_matches_records(records, order.gift_name)
|
||
|
||
if not str(order.gift_name or "").strip():
|
||
verdict = VERDICT_UNKNOWN
|
||
message = "这单没填礼包名，没法跟兑换记录比对"
|
||
elif records_msg or errors:
|
||
verdict = VERDICT_UNKNOWN
|
||
message = (
|
||
"；".join([*errors, records_msg]).strip("；") or "兑换记录没取到，无法判定"
|
||
)
|
||
elif matched:
|
||
verdict = VERDICT_SUCCESS
|
||
latest = matched[0]
|
||
message = (
|
||
f"兑换记录里查到 {len(matched)} 条匹配「{order.gift_name}」"
|
||
f"（最近一条：{latest.get('status_label') or '状态未知'}"
|
||
f"{('，单号 ' + str(latest.get('order_id'))) if latest.get('order_id') else ''}）"
|
||
)
|
||
else:
|
||
verdict = VERDICT_FAILED
|
||
message = (
|
||
f"兑换记录里没有「{order.gift_name}」"
|
||
f"（当前共 {len(records)} 条记录，都没有匹配上）"
|
||
)
|
||
|
||
auto_status = ""
|
||
if verdict == VERDICT_SUCCESS:
|
||
target = STATUS_DONE
|
||
elif verdict == VERDICT_FAILED:
|
||
target = STATUS_FAILED
|
||
else:
|
||
target = ""
|
||
if target and order.status == STATUS_SCRIPTED:
|
||
set_status(db, current, order, target, commit=False)
|
||
auto_status = target
|
||
elif target:
|
||
message = (
|
||
f"{message}（当前状态是「{ORDER_STATUS_LABELS.get(order.status, order.status)}」，"
|
||
f"只有「{ORDER_STATUS_LABELS[STATUS_SCRIPTED]}」的单才会按查询结果自动改状态）"
|
||
)
|
||
|
||
raw: dict[str, Any] = {
|
||
"verdict": verdict,
|
||
"score_available": score,
|
||
"score_gained": gained,
|
||
"score_used": used,
|
||
"records_count": len(records),
|
||
"matched": matched[:5],
|
||
"records": records[:_METRIC_RECORD_LIMIT],
|
||
}
|
||
if score_resp is not None:
|
||
try:
|
||
raw["score_response"] = score_resp.to_dict()
|
||
except Exception as exc: # noqa: BLE001 没有 to_dict 的假客户端（测试）就不存原始体
|
||
logger.debug(f"[huya-collab] 积分原始体不入库：{type(exc).__name__}: {exc}")
|
||
|
||
metric = HuyaExternalOrderMetric(
|
||
order_id=int(order.id),
|
||
order_no=order.order_no or "",
|
||
uid=uid,
|
||
sid=sid,
|
||
score=score,
|
||
gained_score=gained,
|
||
used_score=used,
|
||
records_total=len(records),
|
||
records_matched=len(matched),
|
||
verdict=verdict,
|
||
message=_metric_message(
|
||
"；".join(piece for piece in (score_msg, message) if piece)
|
||
),
|
||
source=source,
|
||
raw=raw,
|
||
created_by=current.id if current else None,
|
||
)
|
||
db.add(metric)
|
||
db.flush()
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:activity_query",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"order_no": order.order_no,
|
||
"uid": uid,
|
||
"sid": sid,
|
||
"source": source,
|
||
"score": score,
|
||
"records_total": len(records),
|
||
"records_matched": len(matched),
|
||
"verdict": verdict,
|
||
"auto_status": auto_status,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
db.refresh(metric)
|
||
return {
|
||
"ok": verdict != VERDICT_UNKNOWN,
|
||
"order_id": int(order.id),
|
||
"order_no": order.order_no,
|
||
"uid": uid,
|
||
"sid": sid,
|
||
"verdict": verdict,
|
||
"verdict_label": VERDICT_LABELS.get(verdict, verdict),
|
||
"score": score,
|
||
"gained_score": gained,
|
||
"used_score": used,
|
||
"records_total": len(records),
|
||
"records_matched": len(matched),
|
||
"message": metric.message,
|
||
"auto_status": auto_status,
|
||
"auto_status_label": ORDER_STATUS_LABELS.get(auto_status, auto_status),
|
||
"status": order.status,
|
||
"status_label": ORDER_STATUS_LABELS.get(order.status, order.status),
|
||
"metric": serialize_metric(metric),
|
||
}
|
||
|
||
|
||
def query_activity_batch(
|
||
db: Session,
|
||
current: User | None,
|
||
orders: Sequence[HuyaExternalOrder],
|
||
*,
|
||
client: Any | None = None,
|
||
interval_seconds: float = PROBE_BATCH_INTERVAL_SECONDS,
|
||
) -> dict:
|
||
"""勾选批量查询积分 / 兑换记录：**串行 + 间隔**（防风控），单条失败不影响其它。"""
|
||
rows = list(orders)
|
||
if not rows:
|
||
raise CollabError("请先勾选要查询的行", code="activity_batch_empty")
|
||
if len(rows) > PROBE_BATCH_MAX:
|
||
raise CollabError(
|
||
f"一次最多查 {PROBE_BATCH_MAX} 行，请分批查询",
|
||
code="activity_batch_too_large",
|
||
)
|
||
|
||
results: list[dict] = []
|
||
for index, order in enumerate(rows):
|
||
if index and interval_seconds > 0:
|
||
time.sleep(interval_seconds)
|
||
try:
|
||
with db.begin_nested():
|
||
results.append(
|
||
query_order_activity(
|
||
db, current, order, client=client, source="batch", commit=False
|
||
)
|
||
)
|
||
except CollabError as exc:
|
||
results.append(
|
||
{
|
||
"ok": False,
|
||
"order_id": int(order.id),
|
||
"order_no": order.order_no,
|
||
"verdict": VERDICT_UNKNOWN,
|
||
"verdict_label": VERDICT_LABELS[VERDICT_UNKNOWN],
|
||
"message": exc.detail,
|
||
"auto_status": "",
|
||
"status": order.status,
|
||
"status_label": ORDER_STATUS_LABELS.get(order.status, order.status),
|
||
}
|
||
)
|
||
db.commit()
|
||
for order in rows:
|
||
db.refresh(order)
|
||
valid = [item for item in results if item.get("verdict") == VERDICT_SUCCESS]
|
||
failed = [item for item in results if item.get("verdict") == VERDICT_FAILED]
|
||
unknown = [item for item in results if item.get("verdict") == VERDICT_UNKNOWN]
|
||
auto_changed = [item for item in results if item.get("auto_status")]
|
||
return {
|
||
"total": len(results),
|
||
"success": len(valid),
|
||
"failed": len(failed),
|
||
"unknown": len(unknown),
|
||
"auto_changed": len(auto_changed),
|
||
"results": results,
|
||
}
|
||
|
||
|
||
def preview_ck(
|
||
ck_text: Any,
|
||
*,
|
||
with_gifts: bool = True,
|
||
sid: int | None = None,
|
||
client: Any | None = None,
|
||
) -> dict:
|
||
"""**只读预览**一份待提交的 CK：解析 + 探针（+ 顺带拉活动礼包），一行都不写库。
|
||
|
||
新建协作单时用它给出「CK 有效 + 绑定角色」的即时反馈；提交时 :func:`create_order`
|
||
还会再校验一次（同一份 CK 命中探针缓存，不会重复打接口）。
|
||
|
||
``sid`` = 活动 ID（接口层传工作台配置里的那个，见 :func:`resolve_gifts_sid`）：
|
||
它决定礼包列表能不能拉到，**不影响 CK 有效性结论**。
|
||
"""
|
||
parsed = parse_ck(str(ck_text or ""))
|
||
uid_text = str(parsed.get("uid") or "")
|
||
cookie = str(parsed.get("cookie") or "")
|
||
|
||
probe_target = HuyaExternalOrder(order_no="preview")
|
||
probe_target.uid = uid_text
|
||
probe_target.cookie = cookie
|
||
result = probe_ck(probe_target, client=client)
|
||
|
||
gifts: list[dict] = []
|
||
gifts_message = ""
|
||
if with_gifts and result.ok and uid_text:
|
||
gifts, gifts_message = fetch_activity_gifts(
|
||
int(uid_text), cookie, sid=sid, client=client
|
||
)
|
||
return {
|
||
"ok": result.ok,
|
||
"message": result.message,
|
||
"uid": uid_text,
|
||
"yyuid": str(parsed.get("yyuid") or ""),
|
||
"hy_username": str(parsed.get("username") or ""),
|
||
"ck_status": CK_VALID if result.ok else CK_INVALID,
|
||
"bind_game_account": result.snapshot.get("bind_game_account", ""),
|
||
"bind_game_role": result.snapshot.get("bind_game_role", ""),
|
||
"bind_game_channel": result.snapshot.get("bind_game_channel", ""),
|
||
"has_role": bool(result.snapshot.get("bind_game_role")),
|
||
"gifts": gifts,
|
||
"gifts_message": gifts_message,
|
||
}
|
||
|
||
|
||
def verify_ck(
|
||
db: Session,
|
||
current: User | None,
|
||
order: HuyaExternalOrder,
|
||
*,
|
||
client: Any | None = None,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""跑一次只读探针并把结论落库。
|
||
|
||
- 结论**只**通过 :func:`set_ck_status` 写（``ck_status`` 的唯一写入口，审计不会丢）；
|
||
- 成功时**覆盖写入**绑定快照（``bind_game_account`` / ``bind_game_role`` /
|
||
``bind_game_channel`` + ``bind_snapshot_at``）：这几列是探针独占的"现在绑了谁"，
|
||
覆盖前会把旧值记进 ``huya:collab:probe_snapshot`` 审计，出问题能查回来；
|
||
- 人工填的那些列（含 ``bind_douyu_account``）**一律不碰**；
|
||
- CK 失效**不改业务状态**（设计文档 §4.2：已绑定/已充值的单子业务上可能仍然有效）。
|
||
"""
|
||
if not str(order.cookie or "").strip():
|
||
raise CollabError("这行还没有 CK，先贴一份再验证", code="no_cookie")
|
||
if order.status == STATUS_VOID:
|
||
raise CollabError(
|
||
"已作废的行不验证 CK（要重新用请先改回流程内状态）", code="void_order"
|
||
)
|
||
|
||
result = probe_ck(order, client=client)
|
||
binding: dict[str, str] = {
|
||
field_name: str(getattr(order, field_name) or "")
|
||
for field_name in PROBE_SNAPSHOT_FIELDS
|
||
}
|
||
if result.ok:
|
||
before = dict(binding)
|
||
for field_name in PROBE_SNAPSHOT_FIELDS:
|
||
setattr(order, field_name, result.snapshot.get(field_name, ""))
|
||
order.bind_snapshot_at = _now()
|
||
binding = {
|
||
field_name: str(getattr(order, field_name) or "")
|
||
for field_name in PROBE_SNAPSHOT_FIELDS
|
||
}
|
||
if binding != before:
|
||
# 快照变了（或第一次读到）才记一条，别每次验证都刷两行审计。
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:probe_snapshot",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"order_no": order.order_no,
|
||
"before": before,
|
||
"after": binding,
|
||
},
|
||
)
|
||
|
||
target = CK_VALID if result.ok else CK_INVALID
|
||
set_ck_status(db, current, order, target, message=result.message, commit=False)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return {
|
||
"id": order.id,
|
||
"order_no": order.order_no,
|
||
"ok": result.ok,
|
||
"ck_status": target,
|
||
"message": result.message,
|
||
"binding": binding,
|
||
}
|
||
|
||
|
||
def _probe_row_error(order: HuyaExternalOrder, message: str) -> dict:
|
||
"""批量验证里「这一行没跑成」的统一形状（状态保持原样，不谎报 invalid）。"""
|
||
return {
|
||
"id": order.id,
|
||
"order_no": order.order_no,
|
||
"ok": False,
|
||
"ck_status": order.ck_status,
|
||
"message": _clip(message, _CK_MESSAGE_LIMIT),
|
||
"binding": {
|
||
field_name: str(getattr(order, field_name) or "")
|
||
for field_name in PROBE_SNAPSHOT_FIELDS
|
||
},
|
||
}
|
||
|
||
|
||
def verify_ck_batch(
|
||
db: Session,
|
||
current: User | None,
|
||
orders: Sequence[HuyaExternalOrder],
|
||
*,
|
||
client: Any | None = None,
|
||
interval_seconds: float | None = None,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""批量验证：**串行 + 间隔**（人工触发，不做定时轮询），单条失败不影响其他条。
|
||
|
||
Returns:
|
||
``{total, valid, failed, results: [{id, order_no, ok, ck_status, message, binding}]}``
|
||
"""
|
||
rows = list(orders)
|
||
if len(rows) > PROBE_BATCH_MAX:
|
||
raise CollabError(
|
||
f"一次最多验证 {PROBE_BATCH_MAX} 条，请分批勾选",
|
||
code="verify_batch_too_large",
|
||
)
|
||
interval = (
|
||
PROBE_BATCH_INTERVAL_SECONDS
|
||
if interval_seconds is None
|
||
else max(0.0, float(interval_seconds))
|
||
)
|
||
|
||
results: list[dict] = []
|
||
for index, order in enumerate(rows):
|
||
if index and interval:
|
||
time.sleep(interval)
|
||
try:
|
||
# 每行一个 SAVEPOINT：某行在写库阶段炸了也只脏这一行（同 import_orders）。
|
||
with db.begin_nested():
|
||
results.append(
|
||
verify_ck(db, current, order, client=client, commit=False)
|
||
)
|
||
db.flush()
|
||
except CollabError as exc:
|
||
results.append(_probe_row_error(order, exc.detail))
|
||
except Exception as exc: # noqa: BLE001 单条失败不该带崩整批
|
||
logger.debug(f"[huya-collab] 批量验证单行失败：{type(exc).__name__}: {exc}")
|
||
results.append(_probe_row_error(order, f"{type(exc).__name__}: {exc}"))
|
||
|
||
if commit:
|
||
db.commit()
|
||
valid = sum(1 for row in results if row["ok"])
|
||
return {
|
||
"total": len(results),
|
||
"valid": valid,
|
||
"failed": len(results) - valid,
|
||
"results": results,
|
||
}
|
||
|
||
|
||
def list_orders_by_ids(db: Session, ids: Sequence[int]) -> list[HuyaExternalOrder]:
|
||
"""按 id 取行（保持勾选顺序；已被别人删掉的 id 直接忽略）。"""
|
||
wanted = list(dict.fromkeys(int(value) for value in ids))
|
||
if not wanted:
|
||
raise CollabError("请先选择要验证的行", code="verify_empty")
|
||
if len(wanted) > PROBE_BATCH_MAX:
|
||
raise CollabError(
|
||
f"一次最多验证 {PROBE_BATCH_MAX} 条，请分批勾选",
|
||
code="verify_batch_too_large",
|
||
)
|
||
rows = db.query(HuyaExternalOrder).filter(HuyaExternalOrder.id.in_(wanted)).all()
|
||
by_id = {int(row.id): row for row in rows}
|
||
return [by_id[value] for value in wanted if value in by_id]
|
||
|
||
|
||
# 导出 CK 的审计动作名（导出=把凭据带出系统，必须留痕）
|
||
CK_EXPORT_ACTION = "huya:collab:ck_export"
|
||
|
||
|
||
def export_cks(
|
||
db: Session,
|
||
current: User | None,
|
||
orders: Sequence[HuyaExternalOrder],
|
||
*,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""把勾选行的 **CK 明文**取出来（前端写剪贴板用）。
|
||
|
||
这是本模块唯一一个把凭据**往外送**的读接口，所以三条约束写死在这里：
|
||
|
||
1. **权限单独一个点**（``huya:collab:ck_export``，默认只给运营）—— 贴 CK 是「录入」，
|
||
导出是「把凭据带走」，两件事不该共用一个权限（口径同 ``huya:cookie:export``）；
|
||
2. **逐行写审计**（``huya:collab:ck_export``）：只记「谁在什么时候导出了哪一单」，
|
||
**绝不把 CK 本体写进审计/日志**（列表接口也照旧不返回 cookie）；
|
||
3. **没 CK 的行显式跳过并回原因**，不静默丢；也不替用户猜 ——
|
||
作废/停用的单只要还有 CK 照样导出，界面把状态标出来让人自己判断。
|
||
|
||
Returns:
|
||
``{total, exported, skipped: [{id, order_no, reason}], items: [{..., ck_text}]}``
|
||
——只有这个接口会返回 ``ck_text``。
|
||
"""
|
||
rows = list(orders)
|
||
if not rows:
|
||
raise CollabError("请先勾选要导出的行", code="ck_export_empty")
|
||
if len(rows) > PROBE_BATCH_MAX:
|
||
raise CollabError(
|
||
f"一次最多导出 {PROBE_BATCH_MAX} 行，请分批操作", code="ck_export_too_large"
|
||
)
|
||
|
||
items: list[dict] = []
|
||
skipped: list[dict] = []
|
||
for order in rows:
|
||
ck_text = str(order.cookie or "").strip()
|
||
if not ck_text:
|
||
skipped.append(
|
||
{
|
||
"id": int(order.id),
|
||
"order_no": order.order_no,
|
||
"reason": "这行还没贴 CK",
|
||
}
|
||
)
|
||
continue
|
||
items.append(
|
||
{
|
||
"id": int(order.id),
|
||
"order_no": order.order_no,
|
||
"uid": str(order.uid or ""),
|
||
"nickname": order.nickname or "",
|
||
"status": order.status,
|
||
"status_label": ORDER_STATUS_LABELS.get(order.status, order.status),
|
||
"ck_status": order.ck_status,
|
||
"ck_status_label": CK_STATUS_LABELS.get(
|
||
order.ck_status, order.ck_status
|
||
),
|
||
"ck_text": ck_text,
|
||
}
|
||
)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action=CK_EXPORT_ACTION,
|
||
target=_audit_target(order),
|
||
detail={
|
||
"order_no": order.order_no,
|
||
"uid": str(order.uid or ""),
|
||
"ck_status": order.ck_status,
|
||
"batch_size": len(rows),
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
return {
|
||
"total": len(rows),
|
||
"exported": len(items),
|
||
"skipped": skipped,
|
||
"items": items,
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 店铺字典（新建协作单时点一下选；也支持自定义输入）
|
||
# --------------------------------------------------------------------------
|
||
|
||
SHOP_NAME_LIMIT = 64
|
||
|
||
|
||
def normalize_shop_name(value: Any) -> str:
|
||
"""店铺名统一收口：去空白 + 限长（空串 = 不填店铺）。"""
|
||
return clean_text(value, field="shop", max_length=SHOP_NAME_LIMIT)
|
||
|
||
|
||
def _shop_audit_target(shop: HuyaCollabShop) -> str:
|
||
return f"huya_collab_shop:{shop.id}"
|
||
|
||
|
||
def _touch_shop(shop: HuyaCollabShop, current: User | None) -> None:
|
||
shop.updated_by = current.id if current else None
|
||
shop.updated_at = _now()
|
||
|
||
|
||
def serialize_shop(
|
||
shop: HuyaCollabShop, *, users: Mapping[int, str] | None = None
|
||
) -> dict:
|
||
names = users or {}
|
||
|
||
def name_of(value: int | None) -> str:
|
||
return names.get(int(value), "") if value else ""
|
||
|
||
return {
|
||
"id": shop.id,
|
||
"name": shop.name or "",
|
||
"sort_order": int(shop.sort_order or 0),
|
||
"is_active": bool(shop.is_active),
|
||
"remark": shop.remark or "",
|
||
"created_by_name": name_of(shop.created_by),
|
||
"updated_by_name": name_of(shop.updated_by),
|
||
"created_at": shop.created_at,
|
||
"updated_at": shop.updated_at,
|
||
}
|
||
|
||
|
||
def list_shop_records(db: Session) -> list[HuyaCollabShop]:
|
||
"""全部店铺（**含停用的**：管理弹窗要能看到；新建下拉由前端按 `is_active` 过滤）。"""
|
||
return (
|
||
db.query(HuyaCollabShop)
|
||
.order_by(HuyaCollabShop.sort_order.asc(), HuyaCollabShop.name.asc())
|
||
.all()
|
||
)
|
||
|
||
|
||
def get_shop(db: Session, shop_id: int) -> HuyaCollabShop:
|
||
shop = db.query(HuyaCollabShop).filter(HuyaCollabShop.id == shop_id).first()
|
||
if shop is None:
|
||
raise CollabError(
|
||
"这个店铺不存在（可能已被别人删除）", status_code=404, code="shop_not_found"
|
||
)
|
||
return shop
|
||
|
||
|
||
def _assert_shop_name_available(
|
||
db: Session, name: str, *, exclude_id: int | None = None
|
||
) -> None:
|
||
query = db.query(HuyaCollabShop).filter(HuyaCollabShop.name == name)
|
||
if exclude_id is not None:
|
||
query = query.filter(HuyaCollabShop.id != exclude_id)
|
||
other = query.first()
|
||
if other is not None:
|
||
raise CollabError(
|
||
f"店铺「{name}」已经在字典里了（第 {other.id} 行）",
|
||
status_code=409,
|
||
code="shop_name_conflict",
|
||
)
|
||
|
||
|
||
def create_shop(
|
||
db: Session,
|
||
current: User | None,
|
||
*,
|
||
name: Any,
|
||
sort_order: int = 0,
|
||
remark: Any = "",
|
||
commit: bool = True,
|
||
) -> HuyaCollabShop:
|
||
"""新增一个店铺（重名 409）。"""
|
||
cleaned = normalize_shop_name(name)
|
||
if not cleaned:
|
||
raise CollabError("店铺名不能为空", code="shop_name_required")
|
||
_assert_shop_name_available(db, cleaned)
|
||
shop = HuyaCollabShop(
|
||
name=cleaned,
|
||
sort_order=int(sort_order or 0),
|
||
is_active=True,
|
||
remark=clean_text(remark, field="shop_remark", max_length=255),
|
||
created_by=current.id if current else None,
|
||
updated_by=current.id if current else None,
|
||
)
|
||
db.add(shop)
|
||
db.flush()
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:shop_create",
|
||
target=_shop_audit_target(shop),
|
||
detail={"name": shop.name, "sort_order": shop.sort_order},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(shop)
|
||
return shop
|
||
|
||
|
||
def update_shop(
|
||
db: Session,
|
||
current: User | None,
|
||
shop: HuyaCollabShop,
|
||
payload: Mapping[str, Any],
|
||
*,
|
||
rename_orders: bool = True,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""改店铺（名字 / 排序 / 停用 / 备注），返回字段级 diff。
|
||
|
||
``rename_orders=True``（默认）时，改名会**级联**把已有订单里的旧名字一起改掉，
|
||
否则筛选会出现"改名前的老名字"这一堆孤儿值；不想要级联就显式传 False。
|
||
"""
|
||
changes: dict[str, dict[str, Any]] = {}
|
||
old_name = shop.name or ""
|
||
renamed_orders = 0
|
||
|
||
if "name" in payload:
|
||
new_name = normalize_shop_name(payload["name"])
|
||
if not new_name:
|
||
raise CollabError("店铺名不能为空", code="shop_name_required")
|
||
if new_name != old_name:
|
||
_assert_shop_name_available(db, new_name, exclude_id=shop.id)
|
||
changes["name"] = {"old": old_name, "new": new_name}
|
||
shop.name = new_name
|
||
|
||
if "sort_order" in payload:
|
||
value = int(payload["sort_order"] or 0)
|
||
if value != int(shop.sort_order or 0):
|
||
changes["sort_order"] = {"old": int(shop.sort_order or 0), "new": value}
|
||
shop.sort_order = value
|
||
|
||
if "is_active" in payload:
|
||
value = bool(payload["is_active"])
|
||
if value != bool(shop.is_active):
|
||
changes["is_active"] = {"old": bool(shop.is_active), "new": value}
|
||
shop.is_active = value
|
||
|
||
if "remark" in payload:
|
||
value = clean_text(payload["remark"], field="shop_remark", max_length=255)
|
||
if value != (shop.remark or ""):
|
||
changes["remark"] = {"old": shop.remark or "", "new": value}
|
||
shop.remark = value
|
||
|
||
if not changes:
|
||
return {}
|
||
|
||
if "name" in changes and rename_orders:
|
||
# 批量改名：订单里存的是名字字符串（见 models.HuyaCollabShop 的取舍说明）。
|
||
renamed_orders = (
|
||
db.query(HuyaExternalOrder)
|
||
.filter(HuyaExternalOrder.shop == old_name)
|
||
.update(
|
||
{HuyaExternalOrder.shop: changes["name"]["new"]},
|
||
synchronize_session=False,
|
||
)
|
||
)
|
||
|
||
_touch_shop(shop, current)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:shop_update",
|
||
target=_shop_audit_target(shop),
|
||
detail={"changes": changes, "renamed_orders": renamed_orders},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(shop)
|
||
return changes
|
||
|
||
|
||
def delete_shop(
|
||
db: Session, current: User | None, shop: HuyaCollabShop, *, commit: bool = True
|
||
) -> dict:
|
||
"""从字典里删掉（需 ``huya:collab:manage``）。
|
||
|
||
**不动订单**：订单里存的是名字字符串，历史单子照旧；想让以后不再用它，
|
||
用「停用」更稳。
|
||
"""
|
||
snapshot = {"id": shop.id, "name": shop.name or ""}
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:shop_delete",
|
||
target=_shop_audit_target(shop),
|
||
detail=snapshot,
|
||
)
|
||
db.delete(shop)
|
||
if commit:
|
||
db.commit()
|
||
return snapshot
|
||
|
||
|
||
def ensure_shops_registered(
|
||
db: Session, current: User | None, names: Sequence[Any]
|
||
) -> list[str]:
|
||
"""把订单里用到的店铺名补进字典（新建 / 编辑 / 粘贴导入共用），返回新登记的名字。
|
||
|
||
这样"自定义输入一个新店铺"就能直接出现在下次的下拉里，不用先去管理弹窗加一遍。
|
||
"""
|
||
wanted = [str(name or "").strip() for name in names]
|
||
wanted = [name for name in dict.fromkeys(wanted) if name]
|
||
if not wanted:
|
||
return []
|
||
known = {
|
||
row[0]
|
||
for row in db.query(HuyaCollabShop.name)
|
||
.filter(HuyaCollabShop.name.in_(wanted))
|
||
.all()
|
||
}
|
||
created: list[str] = []
|
||
for name in wanted:
|
||
if name in known or len(name) > SHOP_NAME_LIMIT:
|
||
continue
|
||
db.add(
|
||
HuyaCollabShop(
|
||
name=name,
|
||
sort_order=0,
|
||
is_active=True,
|
||
remark="",
|
||
created_by=current.id if current else None,
|
||
updated_by=current.id if current else None,
|
||
)
|
||
)
|
||
created.append(name)
|
||
if created:
|
||
# 先 flush 让同一批里的重名在唯一索引上炸出来（而不是拖到 commit）。
|
||
db.flush()
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:shop_create",
|
||
target="huya_collab_shops",
|
||
detail={"auto_registered": created},
|
||
)
|
||
return created
|
||
|
||
|
||
def list_shops(db: Session) -> list[str]:
|
||
"""字典里的店铺名（筛选/下拉用；停用的也在，前端自己决定展示哪些）。"""
|
||
return sorted(shop.name for shop in list_shop_records(db))
|
||
|
||
|
||
def list_gift_names(db: Session) -> list[str]:
|
||
"""台账里出现过的礼包/皮肤名（去重、去空；筛选下拉用）。"""
|
||
rows = db.query(HuyaExternalOrder.gift_name).distinct().all()
|
||
return sorted(
|
||
{str(row[0]).strip() for row in rows if row[0] and str(row[0]).strip()}
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# CRUD
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def create_order(
|
||
db: Session,
|
||
current: User,
|
||
*,
|
||
order_no: str,
|
||
shop: str = "",
|
||
owner_id: int | None = None,
|
||
ck_text: str = "",
|
||
bind_douyu_account: str = "",
|
||
gift_name: str = "",
|
||
home_images: Any = None,
|
||
remark: str = "",
|
||
require_verified: bool = False,
|
||
draft: bool = False,
|
||
client: Any | None = None,
|
||
commit: bool = True,
|
||
) -> HuyaExternalOrder:
|
||
"""新建一行（可同时贴 CK）。订单号唯一，重复直接报冲突。
|
||
|
||
**负责人不填就是当前登录用户**（创建人）—— 新建时不用选，谁建的单归谁；
|
||
要转给别人在列表里点一下改负责人即可（负责人不是权限边界，见设计文档 §7.1）。
|
||
|
||
``draft=True``（**先建单后补资料**）：只要求订单号，店铺也允许先空着；
|
||
状态直接进「待补资料」，CK / 礼包名 / 主页截图都可以等客户资料到了再补。
|
||
这条路径不跑探针、不做严格校验（2026-10-17 用户需求）。
|
||
|
||
``require_verified=True``（**人工新建表单用**，粘贴导入不传）时按"人工必填项"严格校验：
|
||
**订单编号 / 店铺 / CK / 礼包名 / 主页截图**都不许空，且 CK 必须探针通、
|
||
**读到绑定角色**才让建；读不到角色直接 400 —— 这条在服务层做，前端只是体验（AGENTS §2）。
|
||
校验全部在落库**之前**完成，失败不会留下半行。
|
||
|
||
图片只收**主页截图**（``home_images``）：绑定图不再人工上传 —— 绑了谁由探针自动读
|
||
（``bind_game_*``），人工截图没有意义（2026-10-16 用户拍板）。
|
||
"""
|
||
order_no = clean_text(order_no, field="order_no", required=True)
|
||
_assert_order_no_available(db, order_no)
|
||
home_images_keys = normalize_images(home_images, field="home_images")
|
||
|
||
if draft:
|
||
# 草稿单不验 CK、不跑探针：先落一行占住订单号，资料到了在详情里补。
|
||
require_verified = False
|
||
|
||
order = HuyaExternalOrder(
|
||
order_no=order_no,
|
||
shop=clean_text(shop, field="shop"),
|
||
owner_id=_clean_owner_id(owner_id) if owner_id is not None else current.id,
|
||
bind_douyu_account=clean_text(bind_douyu_account, field="bind_douyu_account"),
|
||
gift_name=clean_text(gift_name, field="gift_name"),
|
||
remark=clean_text(remark, field="remark"),
|
||
bind_images=[],
|
||
home_images=home_images_keys,
|
||
status=STATUS_INCOMPLETE if draft else STATUS_PENDING,
|
||
ck_status=CK_UNVERIFIED,
|
||
created_by=current.id,
|
||
updated_by=current.id,
|
||
version=0,
|
||
)
|
||
|
||
uid_reused: list[str] = []
|
||
ck_provided = bool(str(ck_text or "").strip())
|
||
if ck_provided:
|
||
_apply_ck(order, ck_text)
|
||
# 同一个虎牙号允许挂多单（2026-10-16 放开 uid 唯一性）：只记复用痕迹，不拦
|
||
uid_reused = [item.order_no for item in uid_used_orders(db, order.uid)]
|
||
|
||
probe_result: CkProbeResult | None = None
|
||
if require_verified:
|
||
missing: list[str] = []
|
||
if not order.shop:
|
||
missing.append("店铺")
|
||
if not ck_provided:
|
||
missing.append("外部 CK")
|
||
if not order.gift_name:
|
||
missing.append("礼包名")
|
||
if not order.home_images:
|
||
missing.append("主页截图")
|
||
if missing:
|
||
raise CollabError(
|
||
f"下面这几项必须填：{'、'.join(missing)}", code="create_field_required"
|
||
)
|
||
# 先探针再落库：读不到绑定角色就不许建（临时对象，不会留下半行数据）。
|
||
probe_result = probe_ck(order, client=client)
|
||
role = probe_result.snapshot.get("bind_game_role", "")
|
||
if not probe_result.ok or not role:
|
||
raise CollabError(
|
||
f"这份 CK 不能用：{probe_result.message}"
|
||
+ ("（探针通了但读不到绑定角色）" if probe_result.ok else ""),
|
||
code="ck_not_verified",
|
||
)
|
||
for field_name in PROBE_SNAPSHOT_FIELDS:
|
||
setattr(order, field_name, probe_result.snapshot.get(field_name, ""))
|
||
order.bind_snapshot_at = _now()
|
||
|
||
if order.shop:
|
||
# 自定义输入的店铺名顺手登记进字典 → 下次下拉里就能直接点选。
|
||
ensure_shops_registered(db, current, [order.shop])
|
||
|
||
db.add(order)
|
||
db.flush() # 先拿到 id，审计里要带 target
|
||
if probe_result is not None and probe_result.ok:
|
||
# 结论仍然走 set_ck_status 这个唯一写入口（审计与 ck_checked_at 不会漂）。
|
||
set_ck_status(
|
||
db, current, order, CK_VALID, message=probe_result.message, commit=False
|
||
)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:create",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"order_no": order.order_no,
|
||
"shop": order.shop,
|
||
"uid": order.uid or "",
|
||
"ck_status": order.ck_status,
|
||
"gift_name": order.gift_name,
|
||
"bind_game_role": order.bind_game_role or "",
|
||
"home_images": len(order.home_images or []),
|
||
"draft": bool(draft),
|
||
# 这个虎牙号还挂在哪些单上（允许同一份 CK 服务多单，留个痕）
|
||
"uid_reused_in": uid_reused,
|
||
},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return order
|
||
|
||
|
||
def update_order(
|
||
db: Session,
|
||
current: User,
|
||
order: HuyaExternalOrder,
|
||
payload: Mapping[str, Any],
|
||
*,
|
||
expected_version: int | None = None,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""编辑可改字段，返回**字段级 diff**（``{字段: {old, new}}``）。
|
||
|
||
乐观锁：``expected_version`` 与库里不一致 → 409，绝不静默覆盖
|
||
（这正是 Excel 协同最痛的地方）。
|
||
"""
|
||
assert_version(order, expected_version)
|
||
|
||
changes: dict[str, dict[str, Any]] = {}
|
||
for field in EDITABLE_FIELDS:
|
||
if field not in payload:
|
||
continue
|
||
if field == "owner_id":
|
||
new_value: Any = _clean_owner_id(payload[field])
|
||
else:
|
||
new_value = clean_text(
|
||
payload[field],
|
||
field=field,
|
||
required=(field == "order_no"),
|
||
)
|
||
old_value = getattr(order, field)
|
||
if old_value == new_value:
|
||
continue
|
||
if field == "order_no":
|
||
_assert_order_no_available(db, new_value, exclude_id=order.id)
|
||
changes[field] = {"old": old_value, "new": new_value}
|
||
setattr(order, field, new_value)
|
||
|
||
if not changes:
|
||
return {}
|
||
|
||
if "shop" in changes:
|
||
# 改成字典里没有的名字也允许（自定义），顺手登记，筛选下拉不会漏。
|
||
ensure_shops_registered(db, current, [changes["shop"]["new"]])
|
||
|
||
_touch(order, current)
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:update",
|
||
target=_audit_target(order),
|
||
detail={"order_no": order.order_no, "changes": changes},
|
||
)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(order)
|
||
return changes
|
||
|
||
|
||
def delete_order(
|
||
db: Session, current: User, order: HuyaExternalOrder, *, commit: bool = True
|
||
) -> dict:
|
||
"""物理删除（需 ``huya:collab:manage``）。
|
||
|
||
只删行，**不删对象存储里的图**：撤图不可逆，交给孤儿清理脚本按保留期处理
|
||
（设计文档 §6.4）。
|
||
"""
|
||
snapshot = {
|
||
"order_no": order.order_no,
|
||
"uid": order.uid or "",
|
||
"images": images_of(order, "bind_images") + images_of(order, "home_images"),
|
||
}
|
||
record_audit(
|
||
db,
|
||
current,
|
||
action="huya:collab:delete",
|
||
target=_audit_target(order),
|
||
detail={
|
||
"order_no": snapshot["order_no"],
|
||
"uid": snapshot["uid"],
|
||
"image_count": len(snapshot["images"]),
|
||
},
|
||
)
|
||
db.delete(order)
|
||
if commit:
|
||
db.commit()
|
||
return snapshot
|
||
|
||
|
||
def _clean_owner_id(value: Any) -> int | None:
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
owner_id = int(value)
|
||
except (TypeError, ValueError) as exc:
|
||
raise CollabError("负责人必须是用户 ID", code="owner_invalid") from exc
|
||
if owner_id <= 0:
|
||
return None
|
||
return owner_id
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 查询
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def get_order(db: Session, order_id: int) -> HuyaExternalOrder:
|
||
order = db.query(HuyaExternalOrder).filter(HuyaExternalOrder.id == order_id).first()
|
||
if order is None:
|
||
raise CollabError(
|
||
"这一行不存在（可能已被别人删除）", status_code=404, code="order_not_found"
|
||
)
|
||
return order
|
||
|
||
|
||
ORDER_SORT_KEYS: tuple[str, ...] = (
|
||
"seq",
|
||
"order_no",
|
||
"shop",
|
||
"owner",
|
||
"bind",
|
||
"gift_name",
|
||
"status",
|
||
"uid",
|
||
"ck_status",
|
||
"updated_at",
|
||
"created_at",
|
||
)
|
||
|
||
|
||
def list_orders(
|
||
db: Session,
|
||
*,
|
||
shop: str = "",
|
||
owner_id: int | None = None,
|
||
status: str = "",
|
||
ck_status: str = "",
|
||
gift_name: str = "",
|
||
keyword: str = "",
|
||
include_void: bool = False,
|
||
sort_by: str = "seq",
|
||
sort_order: str = "asc",
|
||
page: int = 1,
|
||
page_size: int = 50,
|
||
) -> tuple[list[HuyaExternalOrder], int]:
|
||
"""共享视图：默认**不按人过滤**（负责人只是筛选维度，不是权限边界）。
|
||
|
||
默认按 ``id`` 升序（即创建顺序），新建的行永远追加在最后；表头排序走白名单，
|
||
不允许前端把任意列名拼进 SQL。"""
|
||
query = db.query(HuyaExternalOrder)
|
||
if shop:
|
||
query = query.filter(HuyaExternalOrder.shop == shop)
|
||
if owner_id:
|
||
query = query.filter(HuyaExternalOrder.owner_id == owner_id)
|
||
if status:
|
||
query = query.filter(HuyaExternalOrder.status == status)
|
||
if ck_status:
|
||
query = query.filter(HuyaExternalOrder.ck_status == ck_status)
|
||
skin = (gift_name or "").strip()
|
||
if skin:
|
||
query = query.filter(HuyaExternalOrder.gift_name == skin)
|
||
if not include_void:
|
||
query = query.filter(HuyaExternalOrder.status != STATUS_VOID)
|
||
text = (keyword or "").strip()
|
||
if text:
|
||
like = f"%{text}%"
|
||
query = query.filter(
|
||
or_(
|
||
HuyaExternalOrder.order_no.like(like),
|
||
HuyaExternalOrder.shop.like(like),
|
||
HuyaExternalOrder.bind_douyu_account.like(like),
|
||
HuyaExternalOrder.bind_game_account.like(like),
|
||
HuyaExternalOrder.bind_game_role.like(like),
|
||
HuyaExternalOrder.bind_game_channel.like(like),
|
||
HuyaExternalOrder.uid.like(like),
|
||
HuyaExternalOrder.nickname.like(like),
|
||
HuyaExternalOrder.hy_username.like(like),
|
||
HuyaExternalOrder.gift_name.like(like),
|
||
HuyaExternalOrder.remark.like(like),
|
||
)
|
||
)
|
||
|
||
total = query.count()
|
||
page = max(1, int(page or 1))
|
||
page_size = min(200, max(1, int(page_size or 50)))
|
||
|
||
key = (sort_by or "seq").strip().lower()
|
||
if key not in ORDER_SORT_KEYS:
|
||
key = "seq"
|
||
descending = (sort_order or "asc").strip().lower() == "desc"
|
||
|
||
# 负责人按显示的用户名排序（不是 id），需要一次 outerjoin；其它列都直接落在订单表上。
|
||
if key == "owner":
|
||
query = query.outerjoin(User, HuyaExternalOrder.owner_id == User.id)
|
||
sort_column = User.username
|
||
elif key == "status":
|
||
# 与页头状态顺序保持一致，避免状态列按字母序排得莫名其妙。
|
||
sort_column = func.field(HuyaExternalOrder.status, *ORDER_STATUSES)
|
||
elif key == "bind":
|
||
sort_column = HuyaExternalOrder.bind_game_account
|
||
elif key == "seq":
|
||
sort_column = HuyaExternalOrder.id
|
||
else:
|
||
sort_column = getattr(HuyaExternalOrder, key)
|
||
|
||
ordered = sort_column.desc() if descending else sort_column.asc()
|
||
rows = (
|
||
query.order_by(ordered, HuyaExternalOrder.id.asc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return rows, total
|
||
|
||
|
||
def summary(db: Session) -> dict:
|
||
"""页头统计条：各状态计数 + CK 失效数。"""
|
||
status_rows = (
|
||
db.query(HuyaExternalOrder.status, func.count(HuyaExternalOrder.id))
|
||
.group_by(HuyaExternalOrder.status)
|
||
.all()
|
||
)
|
||
by_status = {str(status): int(count) for status, count in status_rows}
|
||
ck_invalid = (
|
||
db.query(func.count(HuyaExternalOrder.id))
|
||
.filter(HuyaExternalOrder.ck_status == CK_INVALID)
|
||
.scalar()
|
||
)
|
||
return {
|
||
"total": sum(by_status.values()),
|
||
"by_status": {code: by_status.get(code, 0) for code in ORDER_STATUSES},
|
||
"ck_invalid": int(ck_invalid or 0),
|
||
}
|
||
|
||
|
||
def user_names(db: Session, ids: Sequence[Any]) -> dict[int, str]:
|
||
"""一批用户 id → 用户名（避免每行懒加载，N+1）。"""
|
||
clean = {int(value) for value in ids if value}
|
||
if not clean:
|
||
return {}
|
||
rows = db.query(User.id, User.username).filter(User.id.in_(clean)).all()
|
||
return {int(row[0]): str(row[1]) for row in rows}
|
||
|
||
|
||
def users_by_id(db: Session, orders: Sequence[HuyaExternalOrder]) -> dict[int, str]:
|
||
"""一次性把涉及的用户名查出来，避免每行都懒加载（N+1）。"""
|
||
return user_names(
|
||
db,
|
||
[
|
||
value
|
||
for order in orders
|
||
for value in (order.owner_id, order.created_by, order.updated_by)
|
||
],
|
||
)
|
||
|
||
|
||
def serialize_order(
|
||
order: HuyaExternalOrder,
|
||
*,
|
||
users: Mapping[int, str] | None = None,
|
||
buyer_cluster: Mapping[str, Any] | None = None,
|
||
last_metric: HuyaExternalOrderMetric | None = None,
|
||
) -> dict:
|
||
"""把行对象转成接口输出。
|
||
|
||
图片存的是 object_key，URL 在这里现拼 —— 库里存 URL 会在换存储/改路径时
|
||
全部失效（order_site 踩过「存下来的签名链接过期后到处裂图」）。
|
||
|
||
``buyer_cluster`` = 同一买家的其它单（见 :func:`buyer_clusters`），
|
||
``last_metric`` = 最近一次「积分 / 兑换记录」快照（见 :func:`latest_metrics`），
|
||
都由调用方传进来（列表是一次查询算出整页的，别在这里按行查）。
|
||
"""
|
||
names = users or {}
|
||
bind_images = images_of(order, "bind_images")
|
||
home_images = images_of(order, "home_images")
|
||
|
||
def urls(keys: list[str], variant: str = "") -> list[str]:
|
||
return [build_file_url(key, variant) for key in keys]
|
||
|
||
def name_of(value: int | None) -> str:
|
||
return names.get(int(value), "") if value else ""
|
||
|
||
return {
|
||
"id": order.id,
|
||
"order_no": order.order_no,
|
||
"shop": order.shop or "",
|
||
"owner_id": order.owner_id,
|
||
"owner_name": name_of(order.owner_id),
|
||
"uid": order.uid or "",
|
||
"yyuid": order.yyuid or "",
|
||
"hy_username": order.hy_username or "",
|
||
"nickname": order.nickname or "",
|
||
# 只暴露「有没有」，CK 本体绝不出接口。
|
||
"has_cookie": bool(order.cookie),
|
||
"has_raw_cookie": bool(order.raw_cookie),
|
||
"ck_status": order.ck_status,
|
||
"ck_status_label": CK_STATUS_LABELS.get(order.ck_status, order.ck_status),
|
||
"ck_checked_at": order.ck_checked_at,
|
||
"ck_check_message": order.ck_check_message or "",
|
||
"bind_douyu_account": order.bind_douyu_account or "",
|
||
"bind_douyu_uid": order.bind_douyu_uid or "",
|
||
# 探针读到的绑定对象（只读快照，人工不可编辑）
|
||
"bind_game_account": order.bind_game_account or "",
|
||
"bind_game_role": order.bind_game_role or "",
|
||
"bind_game_channel": order.bind_game_channel or "",
|
||
"bind_snapshot_at": order.bind_snapshot_at,
|
||
"bind_images": bind_images,
|
||
"home_images": home_images,
|
||
"bind_image_urls": urls(bind_images),
|
||
"home_image_urls": urls(home_images),
|
||
"bind_image_thumb_urls": urls(bind_images, "thumb"),
|
||
"home_image_thumb_urls": urls(home_images, "thumb"),
|
||
"gift_name": order.gift_name or "",
|
||
"remark": order.remark or "",
|
||
"status": order.status,
|
||
"status_label": ORDER_STATUS_LABELS.get(order.status, order.status),
|
||
"next_status": NEXT_STATUS.get(order.status),
|
||
"bound_at": order.bound_at,
|
||
"recharged_at": order.recharged_at,
|
||
"scripted_at": order.scripted_at,
|
||
"done_at": order.done_at,
|
||
"void_reason": order.void_reason or "",
|
||
"points": order.points,
|
||
"elite_opened": order.elite_opened,
|
||
"snapshot_at": order.snapshot_at,
|
||
"created_by": order.created_by,
|
||
"created_by_name": name_of(order.created_by),
|
||
"updated_by": order.updated_by,
|
||
"updated_by_name": name_of(order.updated_by),
|
||
"version": int(order.version or 0),
|
||
"created_at": order.created_at,
|
||
"updated_at": order.updated_at,
|
||
# 同一买家的其它单（软识别，只读派生；没有就是 null）
|
||
"buyer_cluster": dict(buyer_cluster) if buyer_cluster else None,
|
||
# 最近一次「积分 / 兑换记录」快照（生命周期时间线在详情里，列表只带最新一条）
|
||
"last_score": last_metric.score if last_metric else None,
|
||
"last_metric_at": last_metric.created_at if last_metric else None,
|
||
"last_verdict": (last_metric.verdict or "") if last_metric else "",
|
||
"last_verdict_label": (
|
||
VERDICT_LABELS.get(last_metric.verdict, last_metric.verdict or "")
|
||
if last_metric
|
||
else ""
|
||
),
|
||
"last_metric_message": (last_metric.message or "") if last_metric else "",
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 同一买家的多单（软识别：只读派生，不落库、不改 uid 唯一性规则）
|
||
# --------------------------------------------------------------------------
|
||
|
||
# 买家指纹 = 探针读到的「绑定账号 + 绑定角色」（同一个游戏角色 = 同一个买家）。
|
||
# **不拿 uid 当主键**：一单一份外部 CK，同一个买家买两件往往是两份不同的 CK（uid 不同），
|
||
# 拿 uid 去比反而永远比不中；uid 只作为明细里的辅助信息展示。
|
||
BUYER_KEY_METHOD = "bind_game_account+bind_game_role"
|
||
|
||
|
||
def buyer_key_of(order: HuyaExternalOrder) -> tuple[str, str] | None:
|
||
"""一行的买家指纹；探针还没读到「绑定账号 + 绑定角色」时返回 None（不参与归组）。"""
|
||
account = str(order.bind_game_account or "").strip()
|
||
role = str(order.bind_game_role or "").strip()
|
||
if not account or not role:
|
||
return None
|
||
return (account, role)
|
||
|
||
|
||
def buyer_clusters(
|
||
db: Session, orders: Sequence[HuyaExternalOrder]
|
||
) -> dict[int, dict[str, Any]]:
|
||
"""算出这批行各自「同一买家的其它单」（不是按行发 SQL：整批一次查完）。
|
||
|
||
只有真的 ≥2 单才给结果 —— 单独一单不必在界面上标「同买家 1 单」。
|
||
|
||
返回 ``{order_id: {key, method, total, others: [...]}}``，``others`` 按创建时间升序。
|
||
"""
|
||
keys: dict[tuple[str, str], None] = {}
|
||
for order in orders:
|
||
key = buyer_key_of(order)
|
||
if key is not None:
|
||
keys[key] = None
|
||
if not keys:
|
||
return {}
|
||
|
||
pairs = list(keys)
|
||
rows = (
|
||
db.query(HuyaExternalOrder)
|
||
.filter(
|
||
tuple_(
|
||
HuyaExternalOrder.bind_game_account, HuyaExternalOrder.bind_game_role
|
||
).in_(pairs)
|
||
)
|
||
.order_by(HuyaExternalOrder.created_at.asc(), HuyaExternalOrder.id.asc())
|
||
.all()
|
||
)
|
||
|
||
grouped: dict[tuple[str, str], list[HuyaExternalOrder]] = {}
|
||
for row in rows:
|
||
key = buyer_key_of(row)
|
||
if key is not None and key in keys:
|
||
grouped.setdefault(key, []).append(row)
|
||
|
||
clusters: dict[int, dict[str, Any]] = {}
|
||
for key, siblings in grouped.items():
|
||
if len(siblings) < 2:
|
||
continue
|
||
for row in siblings:
|
||
others = [
|
||
{
|
||
"id": int(item.id),
|
||
"order_no": item.order_no or "",
|
||
"shop": item.shop or "",
|
||
"gift_name": item.gift_name or "",
|
||
"uid": item.uid or "",
|
||
"status": item.status,
|
||
"status_label": ORDER_STATUS_LABELS.get(item.status, item.status),
|
||
"created_at": item.created_at,
|
||
}
|
||
for item in siblings
|
||
if int(item.id) != int(row.id)
|
||
]
|
||
clusters[int(row.id)] = {
|
||
"key": f"{key[0]} · {key[1]}",
|
||
"method": BUYER_KEY_METHOD,
|
||
"total": len(siblings),
|
||
"others": others,
|
||
}
|
||
return clusters
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 批量导入（Excel 粘贴）
|
||
# --------------------------------------------------------------------------
|
||
|
||
IMPORT_COLUMNS: tuple[str, ...] = (
|
||
"shop",
|
||
"order_no",
|
||
"owner",
|
||
"bind_douyu_account",
|
||
"gift_name",
|
||
"remark",
|
||
)
|
||
|
||
# 表头识别：中文（Excel 原表头）+ 英文别名。按位置兜底要求 6 列顺序一致。
|
||
_HEADER_ALIASES: dict[str, str] = {
|
||
"店铺": "shop",
|
||
"shop": "shop",
|
||
"订单编号": "order_no",
|
||
"订单号": "order_no",
|
||
"order_no": "order_no",
|
||
"orderno": "order_no",
|
||
"负责人": "owner",
|
||
"owner": "owner",
|
||
"绑定斗鱼号": "bind_douyu_account",
|
||
"绑定斗鱼": "bind_douyu_account",
|
||
"斗鱼号": "bind_douyu_account",
|
||
"bind_douyu_account": "bind_douyu_account",
|
||
"礼包": "gift_name",
|
||
"礼包名": "gift_name",
|
||
"商品": "gift_name",
|
||
"gift_name": "gift_name",
|
||
"备注": "remark",
|
||
"remark": "remark",
|
||
}
|
||
|
||
|
||
def parse_import_text(text: str) -> tuple[list[str], list[tuple[int, list[str]]]]:
|
||
"""把粘贴的文本切成 (列顺序, [(行号, 单元格)])。
|
||
|
||
支持 Excel 直接复制出来的 TSV（``\\t``），也兼容逗号/分号分隔。
|
||
第一行若能被识别成表头，则按表头映射列；否则按固定列序
|
||
``店铺 / 订单编号 / 负责人 / 绑定斗鱼号 / 礼包 / 备注``。
|
||
"""
|
||
rows: list[tuple[int, list[str]]] = []
|
||
for index, raw_line in enumerate((text or "").splitlines(), start=1):
|
||
line = raw_line.rstrip("\r\n")
|
||
if not line.strip():
|
||
continue
|
||
if "\t" in line:
|
||
cells = line.split("\t")
|
||
elif ";" in line:
|
||
cells = line.split(";")
|
||
elif "," in line:
|
||
cells = line.split(",")
|
||
else:
|
||
cells = [line]
|
||
rows.append((index, [cell.strip() for cell in cells]))
|
||
|
||
if not rows:
|
||
return list(IMPORT_COLUMNS), []
|
||
|
||
first_index, first_cells = rows[0]
|
||
mapped: list[str] = []
|
||
for cell in first_cells:
|
||
key = _HEADER_ALIASES.get(cell.strip().lower()) or _HEADER_ALIASES.get(
|
||
cell.strip()
|
||
)
|
||
mapped.append(key or "")
|
||
if mapped.count("order_no") == 1 and len(mapped) >= 2:
|
||
logger.debug(f"协作台账导入：识别到表头（行 {first_index}）{mapped}")
|
||
return mapped, rows[1:]
|
||
return list(IMPORT_COLUMNS), rows
|
||
|
||
|
||
def import_orders(
|
||
db: Session,
|
||
current: User,
|
||
text: str,
|
||
*,
|
||
owner_by_name: Mapping[str, int] | None = None,
|
||
commit: bool = True,
|
||
) -> dict:
|
||
"""批量导入：逐行处理，**单行失败不影响其他行**（AGENTS §2），错误按行号返回。
|
||
|
||
Returns:
|
||
``{created, updated, skipped, errors: [{line, order_no, message}]}``
|
||
"""
|
||
columns, rows = parse_import_text(text)
|
||
owners = {
|
||
str(name).strip(): int(uid) for name, uid in (owner_by_name or {}).items()
|
||
}
|
||
created = 0
|
||
updated = 0
|
||
skipped = 0
|
||
errors: list[dict[str, Any]] = []
|
||
|
||
seen_order_nos: set[str] = set()
|
||
|
||
for line_no, cells in rows:
|
||
record = {
|
||
column: (cells[index] if index < len(cells) else "")
|
||
for index, column in enumerate(columns)
|
||
if column
|
||
}
|
||
order_no = str(record.get("order_no") or "").strip()
|
||
if not order_no:
|
||
skipped += 1
|
||
continue
|
||
if order_no in seen_order_nos:
|
||
# 同一批粘贴里重复出现：只处理第一条，否则后一条会把刚建的覆盖掉。
|
||
errors.append(
|
||
{
|
||
"line": line_no,
|
||
"order_no": order_no,
|
||
"message": "本次粘贴里重复出现，只处理第一条",
|
||
}
|
||
)
|
||
continue
|
||
seen_order_nos.add(order_no)
|
||
|
||
owner_name = str(record.get("owner") or "").strip()
|
||
payload = {
|
||
"shop": record.get("shop", ""),
|
||
"bind_douyu_account": record.get("bind_douyu_account", ""),
|
||
"gift_name": record.get("gift_name", ""),
|
||
"remark": record.get("remark", ""),
|
||
}
|
||
if owner_name:
|
||
if owner_name not in owners:
|
||
errors.append(
|
||
{
|
||
"line": line_no,
|
||
"order_no": order_no,
|
||
"message": f"负责人「{owner_name}」不存在或不是活跃用户",
|
||
}
|
||
)
|
||
continue
|
||
payload["owner_id"] = owners[owner_name]
|
||
|
||
outcome = ""
|
||
try:
|
||
# 每行套一个 SAVEPOINT：这一行炸了只回滚这一行，整批继续。
|
||
# 若直接往同一个事务里写，一次约束冲突就会把会话打成失败态，
|
||
# 后面每一行都跟着报错 —— 那就成了「单行失败带崩整批」。
|
||
with db.begin_nested():
|
||
existing = (
|
||
db.query(HuyaExternalOrder)
|
||
.filter(HuyaExternalOrder.order_no == order_no)
|
||
.first()
|
||
)
|
||
if existing is None:
|
||
create_order(
|
||
db,
|
||
current,
|
||
order_no=order_no,
|
||
shop=payload.get("shop", ""),
|
||
owner_id=payload.get("owner_id"),
|
||
bind_douyu_account=payload.get("bind_douyu_account", ""),
|
||
gift_name=payload.get("gift_name", ""),
|
||
remark=payload.get("remark", ""),
|
||
commit=False,
|
||
)
|
||
outcome = "created"
|
||
else:
|
||
# 已存在按「更新」处理（设计文档 §10.2 E）；导入不带
|
||
# expected_version，视为「明确要覆盖」。
|
||
changes = update_order(
|
||
db,
|
||
current,
|
||
existing,
|
||
payload,
|
||
expected_version=None,
|
||
commit=False,
|
||
)
|
||
outcome = "updated" if changes else "skipped"
|
||
# SessionLocal 是 autoflush=False：显式 flush，让后面几行的
|
||
# 重复检查能看到刚写入的这行。
|
||
db.flush()
|
||
except CollabError as exc:
|
||
errors.append(
|
||
{"line": line_no, "order_no": order_no, "message": exc.detail}
|
||
)
|
||
continue
|
||
except Exception as exc: # noqa: BLE001 单行失败不该带崩整批导入
|
||
errors.append(
|
||
{
|
||
"line": line_no,
|
||
"order_no": order_no,
|
||
"message": f"{type(exc).__name__}: {exc}",
|
||
}
|
||
)
|
||
continue
|
||
|
||
if outcome == "created":
|
||
created += 1
|
||
elif outcome == "updated":
|
||
updated += 1
|
||
else:
|
||
skipped += 1
|
||
|
||
if commit:
|
||
db.commit()
|
||
return {
|
||
"created": created,
|
||
"updated": updated,
|
||
"skipped": skipped,
|
||
"errors": errors,
|
||
"parsed_columns": columns,
|
||
}
|