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

feat(huya): 协作台账状态机新口径（跳步/已确认绑定/兑换失败）+ 账号全生命周期快照（迁移 0059）

Browse Source
This commit is contained in:
yml2213
2026-10-16 22:00:34 +08:00
parent aac264b639
commit e8abd60734
11 changed files with 1897 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/backend/migrations/versions/20261016_0059_huya_order_metrics.py
+76
View File
@@ -0,0 +1,76 @@
"""新增 `huya_external_order_metrics`：账号全生命周期快照（积分 + 兑换记录）。
背景（2026-10-16 用户需求）：抢购完成后要能查 `getUserScore`（积分）与
`getUserPrizeRecords`（兑换记录），**每次查询都落一条快照** —— 这样"积分变化 + 兑换记录"
两条线索合起来就能判断这一单到底兑换成功没有，而不是只看某一次的结果。
只追加、不更新；`verdict` 三态 success / failed / unknown；`raw` 存接口原始明细
（里面**没有凭据**）。订单被删时这些快照跟着删（外键 CASCADE）。
可重复执行：表已存在就直接返回。
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20261016_0059"
down_revision: str | None = "20261016_0058"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
TABLE = "huya_external_order_metrics"
ORDERS = "huya_external_orders"
USERS = "users"
def upgrade() -> None:
bind = op.get_bind()
if TABLE in set(sa.inspect(bind).get_table_names()):
return
op.create_table(
TABLE,
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("order_id", sa.Integer(), nullable=False),
sa.Column("order_no", sa.String(length=64), nullable=False, server_default=""),
sa.Column("uid", sa.String(length=32), nullable=False, server_default=""),
sa.Column("sid", sa.Integer(), nullable=True),
sa.Column("score", sa.Integer(), nullable=True),
sa.Column("gained_score", sa.Integer(), nullable=True),
sa.Column("used_score", sa.Integer(), nullable=True),
sa.Column("records_total", sa.Integer(), nullable=False, server_default="0"),
sa.Column("records_matched", sa.Integer(), nullable=False, server_default="0"),
sa.Column(
"verdict", sa.String(length=16), nullable=False, server_default="unknown"
),
sa.Column("message", sa.String(length=512), nullable=False, server_default=""),
sa.Column(
"source", sa.String(length=16), nullable=False, server_default="manual"
),
sa.Column("raw", sa.JSON(), nullable=True),
sa.Column("created_by", sa.Integer(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(
["order_id"],
[f"{ORDERS}.id"],
name=f"fk_{TABLE}_order_id",
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["created_by"], [f"{USERS}.id"], name=f"fk_{TABLE}_created_by"
),
)
op.create_index(f"ix_{TABLE}_order_id", TABLE, ["order_id"])
op.create_index(f"ix_{TABLE}_uid", TABLE, ["uid"])
op.create_index(f"ix_{TABLE}_verdict", TABLE, ["verdict"])
op.create_index(f"ix_{TABLE}_created_at", TABLE, ["created_at"])
def downgrade() -> None:
bind = op.get_bind()
if TABLE in set(sa.inspect(bind).get_table_names()):
# MySQL 删表会连索引一起删。
op.drop_table(TABLE)
web/backend/models.py
+53
View File
@@ -1191,3 +1191,56 @@ class HuyaExternalOrder(Base):
updated_at: Mapped[datetime] = mapped_column(
UTCDateTime, default=_utcnow, onupdate=_utcnow, index=True
)
class HuyaExternalOrderMetric(Base):
"""虎牙外部订单的**账号全生命周期**快照：每查一次积分/兑换记录写一条。
2026-10-16 用户需求：抢购完成后查兑换记录与积分，**每次积分变化都要落一条**，
这样才能靠"兑换记录 + 积分变化"判断到底兑换成功没有。
设计取舍：
1. **只追加、不更新、不删除**（快照语义）：历史要能回放，改一行就等于毁证；
2. **不存凭据**：`raw` 只放积分响应与兑换记录明细（里面没有 cookie），
凭据永远只待在 `huya_external_orders.cookie`；
3. `verdict` 三态：`success`（兑换记录里有这一单的礼包）/ `failed`（接口通了但没记录）/
`unknown`（接口没通或没可比的礼包名），**只有 success / failed 会去改业务状态**；
4. 订单删掉时这条也跟着删（`ondelete="CASCADE"`）—— 它没有独立价值。
"""
__tablename__ = "huya_external_order_metrics"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
order_id: Mapped[int] = mapped_column(
ForeignKey("huya_external_orders.id", ondelete="CASCADE"), index=True
)
# 冗余一份单号与 uid：订单被改单号后，历史快照仍然说得清"当时查的是谁"。
order_no: Mapped[str] = mapped_column(String(64), default="", server_default="")
uid: Mapped[str] = mapped_column(
String(32), default="", server_default="", index=True
)
sid: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 积分快照（可用积分 / 累计获得 / 已消耗）—— 变化靠相邻两条相减就能看出来
score: Mapped[int | None] = mapped_column(Integer, nullable=True)
gained_score: Mapped[int | None] = mapped_column(Integer, nullable=True)
used_score: Mapped[int | None] = mapped_column(Integer, nullable=True)
records_total: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
records_matched: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
# success / failed / unknown（见类文档）＋ 一句人话结论
verdict: Mapped[str] = mapped_column(
String(16), default="unknown", server_default="unknown", index=True
)
message: Mapped[str] = mapped_column(String(512), default="", server_default="")
# manual（人工点）/ batch（勾选批量）：出问题时能分清是谁按的
source: Mapped[str] = mapped_column(
String(16), default="manual", server_default="manual"
)
# 接口原始内容（积分响应 + 兑换记录明细；**不含凭据**）
raw: Mapped[dict | None] = mapped_column(JSON, nullable=True)
created_by: Mapped[int | None] = mapped_column(
ForeignKey("users.id"), nullable=True
)
created_at: Mapped[datetime] = mapped_column(
UTCDateTime, default=_utcnow, index=True
)
web/backend/routers/huya/collab.py
+115 -2
View File
@@ -26,9 +26,13 @@ from ...models import AuditLog, HuyaExternalOrder, User
from ...schemas import (
HuyaCollabCkPreviewOut,
HuyaCollabCkPreviewRequest,
HuyaCollabMetricOut,
HuyaCollabShopCreate,
HuyaCollabShopOut,
HuyaCollabShopUpdate,
HuyaExternalOrderActivityItemOut,
HuyaExternalOrderActivityQueryOut,
HuyaExternalOrderActivityQueryRequest,
HuyaExternalOrderAuditOut,
HuyaExternalOrderCkRequest,
HuyaExternalOrderCkStatusRequest,
@@ -38,6 +42,8 @@ from ...schemas import (
HuyaExternalOrderListOut,
HuyaExternalOrderOut,
HuyaExternalOrderOwnerOut,
HuyaExternalOrderStatusBatchOut,
HuyaExternalOrderStatusBatchRequest,
HuyaExternalOrderStatusRequest,
HuyaExternalOrderSummaryOut,
HuyaExternalOrderUpdate,
@@ -58,10 +64,12 @@ def _http_error(exc: collab.CollabError) -> HTTPException:
def _serialize_one(db: Session, order: HuyaExternalOrder) -> dict:
clusters = collab.buyer_clusters(db, [order])
metrics = collab.latest_metrics(db, [order])
return collab.serialize_order(
order,
users=collab.users_by_id(db, [order]),
buyer_cluster=clusters.get(int(order.id)),
last_metric=metrics.get(int(order.id)),
)
@@ -200,12 +208,16 @@ def list_external_orders(
page_size=page_size,
)
names = collab.users_by_id(db, rows)
# 「同一买家的其它单」整页一次算完（按绑定账号 + 绑定角色软识别，只读派生）
# 「同一买家的其它单」+「最近一次积分/兑换快照」整页一次算完，别按行发 SQL
clusters = collab.buyer_clusters(db, rows)
metrics = collab.latest_metrics(db, rows)
return {
"items": [
collab.serialize_order(
row, users=names, buyer_cluster=clusters.get(int(row.id))
row,
users=names,
buyer_cluster=clusters.get(int(row.id)),
last_metric=metrics.get(int(row.id)),
)
for row in rows
],
@@ -289,6 +301,62 @@ def preview_external_order_ck(
return preview
@router.post(
"/external-orders/status/batch", response_model=HuyaExternalOrderStatusBatchOut
)
def batch_update_external_order_status(
req: HuyaExternalOrderStatusBatchRequest,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""勾选多行一次改业务状态（单个也走这里）。
人工切换允许**往前跳步**（待处理 → 已上脚本），回退要带 ``rollback=true``；
某一行前置校验没过（如没主页截图不能进「已确认绑定」）只记在这一条的 ``message`` 里。
"""
try:
orders = collab.list_orders_by_ids(db, req.ids)
result = collab.set_status_batch(
db,
current,
orders,
req.status,
reason=req.reason,
rollback=req.rollback,
)
except collab.CollabError as exc:
raise _http_error(exc) from exc
result["target_label"] = collab.ORDER_STATUS_LABELS.get(req.status, req.status)
return result
@router.post(
"/external-orders/activity/query", response_model=HuyaExternalOrderActivityQueryOut
)
def batch_query_external_order_activity(
req: HuyaExternalOrderActivityQueryRequest,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""勾选多行查询积分 + 兑换记录（**串行带间隔**，防风控）。
每行都会落一条生命周期快照；「已上脚本」的单按查询结果自动改状态
（查到礼包 → 已完成；没查到 → 兑换失败），其它状态只记录不动状态。
"""
try:
orders = collab.list_orders_by_ids(db, req.ids)
result = collab.query_activity_batch(db, current, orders)
except collab.CollabError as exc:
raise _http_error(exc) from exc
for item in result["results"]:
item["status_label"] = collab.ORDER_STATUS_LABELS.get(
item.get("status", ""), item.get("status", "")
)
auto = item.get("auto_status") or ""
item["auto_status_label"] = collab.ORDER_STATUS_LABELS.get(auto, auto)
return result
@router.post("/external-orders/ck/verify", response_model=HuyaExternalOrderVerifyOut)
def verify_external_orders_ck(
req: HuyaExternalOrderVerifyRequest,
@@ -399,6 +467,51 @@ def update_external_order_status(
return _serialize_one(db, order)
@router.post(
"/external-orders/{order_id}/activity/query",
response_model=HuyaExternalOrderActivityItemOut,
)
def query_external_order_activity(
order_id: int,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""查这一单的积分 + 兑换记录（单行「查询」按钮）：写一条生命周期快照。
「已上脚本」的单按查询结果自动改状态：查到这一单的礼包 → 已完成；
接口通了但没查到 → 兑换失败；接口没通 → 不动（数据不够不许瞎判）。
"""
try:
order = collab.get_order(db, order_id)
result = collab.query_order_activity(db, current, order)
except collab.CollabError as exc:
raise _http_error(exc) from exc
result["status_label"] = collab.ORDER_STATUS_LABELS.get(
result.get("status", ""), result.get("status", "")
)
auto = result.get("auto_status") or ""
result["auto_status_label"] = collab.ORDER_STATUS_LABELS.get(auto, auto)
return result
@router.get(
"/external-orders/{order_id}/metrics", response_model=list[HuyaCollabMetricOut]
)
def list_external_order_metrics(
order_id: int,
db: Session = Depends(get_db),
_current: User = Depends(require_permission("huya:collab:view")),
):
"""这一单的「账号全生命周期」时间线：每次查积分/兑换记录的快照（新→旧）。"""
try:
order = collab.get_order(db, order_id)
except collab.CollabError as exc:
raise _http_error(exc) from exc
rows = collab.list_metrics(db, order)
names = collab.user_names(db, [row.created_by for row in rows])
return [collab.serialize_metric(row, users=names) for row in rows]
@router.post("/external-orders/{order_id}/ck", response_model=HuyaExternalOrderOut)
def update_external_order_ck(
order_id: int,
web/backend/schemas.py
+108
View File
@@ -1268,6 +1268,107 @@ class HuyaExternalOrderVerifyOut(BaseModel):
results: list[HuyaExternalOrderVerifyItemOut] = Field(default_factory=list)
class HuyaExternalOrderStatusBatchRequest(BaseModel):
"""勾选多行，一次切到同一个业务状态（单个也走这里，ids 只放一个）。"""
ids: list[int] = Field(default_factory=list)
status: str = Field(..., min_length=1, max_length=16)
reason: str = Field("", max_length=256)
# 目标不是「向前」时必须显式置 True（和单行接口一致，避免误点）
rollback: bool = False
class HuyaExternalOrderStatusBatchItemOut(BaseModel):
"""批量改状态里的一行结论（``ok=False`` 时看 ``message``，不影响其它行）。"""
id: int
order_no: str = ""
ok: bool = False
message: str = ""
class HuyaExternalOrderStatusBatchOut(BaseModel):
"""批量改状态汇总：``updated + failed = total``。"""
total: int
updated: int
failed: int
target: str = ""
target_label: str = ""
results: list[HuyaExternalOrderStatusBatchItemOut] = Field(default_factory=list)
class HuyaExternalOrderActivityQueryRequest(BaseModel):
"""勾选多行，一次查这些行的积分 + 兑换记录（一次最多 50 条，串行带间隔）。"""
ids: list[int] = Field(default_factory=list)
class HuyaCollabMetricMatchedOut(BaseModel):
"""匹配到的那条兑换记录（只留界面要用的几个字段）。"""
prize_name: str = ""
score: int | None = None
status_label: str = ""
order_id: str = ""
exchange_time: int | None = None
class HuyaCollabMetricOut(BaseModel):
"""一条「账号全生命周期」快照（积分 + 兑换记录 + 判定）。"""
id: int
order_id: int
order_no: str = ""
uid: str = ""
sid: int | None = None
score: int | None = None
gained_score: int | None = None
used_score: int | None = None
records_total: int = 0
records_matched: int = 0
verdict: str = "unknown"
verdict_label: str = ""
message: str = ""
source: str = "manual"
matched_preview: list[HuyaCollabMetricMatchedOut] = Field(default_factory=list)
created_by: int | None = None
created_by_name: str = ""
created_at: datetime | None = None
class HuyaExternalOrderActivityItemOut(BaseModel):
"""批量查询里的一行结论。"""
ok: bool = False
order_id: int
order_no: str = ""
verdict: str = "unknown"
verdict_label: str = ""
score: int | None = None
gained_score: int | None = None
used_score: int | None = None
records_total: int = 0
records_matched: int = 0
message: str = ""
# 查询后是否自动改了状态（空 = 没改，见服务层：只有「已上脚本」的单会自动改）
auto_status: str = ""
auto_status_label: str = ""
status: str = ""
status_label: str = ""
class HuyaExternalOrderActivityQueryOut(BaseModel):
"""批量查询汇总：``success + failed + unknown = total``。"""
total: int
success: int
failed: int
unknown: int
auto_changed: int = 0
results: list[HuyaExternalOrderActivityItemOut] = Field(default_factory=list)
class HuyaCollabBuyerClusterOut(BaseModel):
"""同一买家的多单（软识别，只读派生）。
@@ -1346,6 +1447,13 @@ class HuyaExternalOrderOut(BaseModel):
# 同一买家的其它单（软识别：绑定账号 + 绑定角色相同；只读派生，没识别到就是 null）
buyer_cluster: HuyaCollabBuyerClusterOut | None = None
# 最近一次「积分 / 兑换记录」快照（完整时间线走 GET /{id}/metrics）
last_score: int | None = None
last_metric_at: datetime | None = None
last_verdict: str = ""
last_verdict_label: str = ""
last_metric_message: str = ""
class HuyaExternalOrderListOut(BaseModel):
"""列表响应：共享视图，带上总数供分页使用。"""
web/backend/services/huya_collab_service.py
+549 -26
View File
@@ -35,7 +35,13 @@ from ..huya_defaults import (
HUYA_DEFAULT_OUTER_ACT_ID,
HUYA_DEFAULT_SID,
)
from ..models import HuyaCollabShop, HuyaConfig, HuyaExternalOrder, User
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
@@ -50,6 +56,7 @@ STATUS_BOUND = "bound"
STATUS_RECHARGED = "recharged"
STATUS_SCRIPTED = "scripted"
STATUS_DONE = "done"
STATUS_FAILED = "failed"
STATUS_VOID = "void"
ORDER_STATUSES: tuple[str, ...] = (
@@ -58,19 +65,32 @@ ORDER_STATUSES: tuple[str, ...] = (
STATUS_RECHARGED,
STATUS_SCRIPTED,
STATUS_DONE,
STATUS_FAILED,
STATUS_VOID,
)
ORDER_STATUS_LABELS = {
STATUS_PENDING: "待处理",
STATUS_BOUND: "已绑定",
# 人工核对过「探针读到的绑定角色」与「主页截图」一致，才允许进这一步（见 _assert_can_enter_bound）
STATUS_BOUND: "已确认绑定",
STATUS_RECHARGED: "已充值",
STATUS_SCRIPTED: "已上脚本",
STATUS_DONE: "已完成",
# 旁路终态：抢购后查兑换记录，没查到这一单的礼包 → 判失败（人工确认后可作废或回退重试）
STATUS_FAILED: "兑换失败",
STATUS_VOID: "已作废",
}
# 「向前一步」的下一站；done 是终点，void 是任意态可达的旁路，都不在这条链上。
# 流程链（时间戳同步与「能不能向前走」都只看这条链）；failed / void 是**旁路**终态。
STATUS_FLOW: tuple[str, ...] = (
STATUS_PENDING,
STATUS_BOUND,
STATUS_RECHARGED,
STATUS_SCRIPTED,
STATUS_DONE,
)
# 「向前一步」的下一站；done 是终点，failed / void 是任意态可达的旁路，都不在这条链上。
NEXT_STATUS = {
STATUS_PENDING: STATUS_BOUND,
STATUS_BOUND: STATUS_RECHARGED,
@@ -94,6 +114,25 @@ STATUS_TIMESTAMP_FIELDS = {
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 状态机（凭据本身）
# --------------------------------------------------------------------------
@@ -386,26 +425,27 @@ def _apply_ck(order: HuyaExternalOrder, cookie_text: str) -> dict:
def _assert_can_enter_bound(order: HuyaExternalOrder) -> None:
"""进入「已绑定」的前置校验（在服务层做，前端校验只是体验）。
"""进入「已确认绑定」的前置校验（服务层做，前端校验只是体验）。
绑定对象有两条来源：人工填的 `bind_douyu_account`（Excel D 列），或**探针读到的
`bind_game_role`**（2026-10-16 起探针把绑定角色写进快照列）。有任意一条就算"绑上了"，
否则只验证过 CK、没读过绑定角色的行会被挡在 `bound` 外面。
2026-10-16 用户口径：**人工核对过「探针读到的绑定角色」与「主页截图」一致**，才允许
切到这一步。所以这里要两样东西齐全：
- **绑定角色**（`bind_game_role`，探针写）—— 没读到角色就没人知道该跟哪张图对；
- **主页截图**（`home_images`，人工上传）—— 没有图就没有"核对"这件事的依据。
绑定图已停用（绑定关系改由探针读），不再要求；`bind_douyu_account` 也不再算数
（那是人工自由文本，且已从界面下线）。
"""
missing: list[str] = []
if not str(order.cookie or "").strip():
missing.append("CK")
has_binding = bool(
str(order.bind_douyu_account or "").strip()
or str(order.bind_game_role or "").strip()
)
if not has_binding:
missing.append("绑定账号（人工填绑定斗鱼号，或先「验证 CK」读到绑定角色）")
if not images_of(order, "bind_images"):
missing.append("绑定图")
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"还不能标记为「已绑定」，缺少：{'、'.join(missing)}",
f"还不能标记为「{ORDER_STATUS_LABELS[STATUS_BOUND]}」，缺少：{'、'.join(missing)}",
code="bound_precondition",
)
@@ -418,9 +458,11 @@ def _sync_stage_timestamps(
- 目标状态**之前**（含自身）缺失的时间戳补上：状态说「已上脚本」，
时间戳就不能是空的，否则前端那格显示空白、没人知道什么时候做的；
- 目标状态**之后**的时间戳清空：回退之后还留着「已上脚本时间」，
就是状态与时间戳互相矛盾。
就是状态与时间戳互相矛盾；
- `void` / `failed` 是旁路终态：不动流程链上的时间戳（从"已上脚本"判失败时，
"什么时候上的脚本"这条信息必须留着）。
"""
if target == STATUS_VOID:
if target in (STATUS_VOID, STATUS_FAILED):
return
target_index = STAGE_INDEX[target]
for stage, field in STATUS_TIMESTAMP_FIELDS.items():
@@ -443,10 +485,13 @@ def set_status(
) -> dict:
"""推进/回退业务状态（本模块唯一的 ``status`` 写入口）。
人工切换**允许往前跳步**（如 待处理 → 已上脚本），回退必须显式 ``rollback=True``；
规则见 :func:`is_forward_status`。
Args:
target: 目标状态，必须在 ``ORDER_STATUSES`` 里。
reason: 作废原因（``target=void`` 时必填）。
rollback: 目标不是「向前一步」时，必须显式置 True 才允许改。
rollback: 目标不是「向前」时必须显式置 True 才允许改。
"""
if target not in ORDER_STATUSES:
raise CollabError(f"未知状态：{target}", code="status_unknown")
@@ -454,16 +499,18 @@ def set_status(
return {}
from_status = order.status
if target != STATUS_VOID and target != NEXT_STATUS.get(from_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]}」"
f"要往前的话下一步是「{ORDER_STATUS_LABELS[next_status]}」（也可以直接跳到更后面的状态）"
if next_status
else "这单已经走到终点，只能作废"
else "这单已经走到终点"
)
raise CollabError(
f"状态只能向前推进（{hint}）；要回退请显式勾选「回退」",
f"不能把「{ORDER_STATUS_LABELS.get(from_status, from_status)}」改回"
f"「{ORDER_STATUS_LABELS[target]}」（{hint}）；确实要回退请显式勾选「回退」",
code="status_not_forward",
)
if from_status == STATUS_VOID:
@@ -494,7 +541,7 @@ def set_status(
detail={
"from": from_status,
"to": target,
"rollback": bool(rollback and target != NEXT_STATUS.get(from_status)),
"rollback": bool(not forward),
"reason": order.void_reason,
"order_no": order.order_no,
},
@@ -505,6 +552,81 @@ def set_status(
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 写入口
# --------------------------------------------------------------------------
@@ -969,6 +1091,395 @@ def fetch_activity_gifts(
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,
*,
@@ -1762,14 +2273,16 @@ def serialize_order(
*,
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`），由调用方
传进来（列表是一次查询算出整页的，别在这里按行查）。
``buyer_cluster`` = 同一买家的其它单（见 :func:`buyer_clusters`），
``last_metric`` = 最近一次「积分 / 兑换记录」快照（见 :func:`latest_metrics`），
都由调用方传进来（列表是一次查询算出整页的，别在这里按行查）。
"""
names = users or {}
bind_images = images_of(order, "bind_images")
@@ -1833,6 +2346,16 @@ def serialize_order(
"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 "",
}