feat(huya): 协作台账店铺字典·新建强校验·同买家多单 + uid 放开(迁移 0057/0058)

- 店铺字典(迁移 0057):大按钮点选 + 「新店」自定义自动登记;重命名可级联、
  停用/备注;列表行内改、详情、筛选条共用同一个 picker
- 新建强校验:订单编号/店铺/CK/礼包名/主页截图必填;贴 CK 即探针预览,
  读不到绑定角色禁止提交;礼包用工作台配置的活动 SID 现拉(浏览器缓存 + 强制刷新)
- 图片:绑定图停用、「绑定斗鱼号」下线(绑定关系改由探针读 bind_game_*),
  主页截图改为新建必填(弹窗内先传 OSS 再随表单落库)
- 同买家多单软识别:按「绑定账号 + 绑定角色」整批一次查询,列表「同买家」列 +
  详情分组,点单号直接切单
- uid 放开唯一性(迁移 0058):同一份 CK 可服务多单,原 409 冲突改为提示 + 审计
- 列表体验:序号列、搜索扩到 11 个字段、CK 状态列截断、新建弹窗放大并去滚动条
- 验证:pytest 932 passed / 6 skipped;接口冒烟 56 项全绿;tsc+eslint+build 通过
This commit is contained in:
yml2213
2026-10-16 21:29:08 +08:00
parent df8c63b829
commit aac264b639
17 changed files with 3237 additions and 155 deletions
@@ -0,0 +1,53 @@
"""协作台账的店铺字典表 `huya_collab_shops`。
为什么要有它:新建协作单时"店铺"原来是手打的自由文本,同一个店铺各种写法(错别字、
多空格)会把筛选打散。改成一张可维护的字典:新建时**点一下选中**,
下拉里也能直接输一个新名字(自定义,会自动登记进字典)。
**不回填历史订单里的店铺名**(2026-10-16 用户明确"不用看之前的"):字典从空开始,
需要哪些店铺就在「店铺管理」里加、或在新建时自定义输入。
注意:订单里存的仍是店铺**名字**(`huya_external_orders.shop`),不是 id —— 历史 Excel
就是文本,本轮不改语义;改名走 `update_shop(..., rename_orders=True)` 级联。
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20261016_0057"
down_revision: str | None = "20261016_0056"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
TABLE = "huya_collab_shops"
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("name", sa.String(length=64), nullable=False),
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"),
sa.Column("remark", sa.String(length=255), nullable=False, server_default=""),
sa.Column("created_by", sa.Integer(), nullable=True),
sa.Column("updated_by", sa.Integer(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("updated_at", sa.DateTime(), nullable=False),
)
op.create_index(f"ix_{TABLE}_name", TABLE, ["name"], unique=True)
op.create_index(f"ix_{TABLE}_is_active", TABLE, ["is_active"])
def downgrade() -> None:
bind = op.get_bind()
if TABLE in set(sa.inspect(bind).get_table_names()):
# MySQL 删表会连索引一起删。
op.drop_table(TABLE)
@@ -0,0 +1,78 @@
"""放开 `huya_external_orders.uid` 的唯一性:一个虎牙号可以挂多单。
背景(2026-10-16 用户拍板):真机上"同一个买家买两件"往往**继续用同一份外部 CK**,
而 0055 建的 `ix_huya_external_orders_uid` 是**唯一索引** —— 第二单会被数据库顶掉
(`Duplicate entry ... for key ...uid`)。原来"一个虎牙号只能挂一单"的规则随之作废:
服务层不再抛 `uid_conflict`,「这个号还挂在哪些单上」只作为提示/审计信息(见 §5.6)。
做法:把该索引改成**普通索引**(保留查询性能,不再约束唯一)。
uid 仍然可空 —— 「先建单、后贴 CK」不受影响(MySQL 索引对多个 NULL 宽松,普通索引更是)。
可重复执行:按当前索引的实际唯一性判断,已经是非唯一就直接返回。
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20261016_0058"
down_revision: str | None = "20261016_0057"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
TABLE = "huya_external_orders"
INDEX = "ix_huya_external_orders_uid"
COLUMNS = ["uid"]
def _has_table(bind) -> bool:
return TABLE in set(sa.inspect(bind).get_table_names())
def _index(bind) -> dict | None:
for index in sa.inspect(bind).get_indexes(TABLE):
if index.get("name") == INDEX:
return index
return None
def upgrade() -> None:
bind = op.get_bind()
if not _has_table(bind):
return
index = _index(bind)
if index is None:
# 索引丢了(手工库):补一个普通索引就行
op.create_index(INDEX, TABLE, COLUMNS, unique=False)
return
if not index.get("unique"):
return
op.drop_index(INDEX, table_name=TABLE)
op.create_index(INDEX, TABLE, COLUMNS, unique=False)
def downgrade() -> None:
bind = op.get_bind()
if not _has_table(bind):
return
index = _index(bind)
if index is None:
op.create_index(INDEX, TABLE, COLUMNS, unique=True)
return
if index.get("unique"):
return
# 回退前先确认没有重复 uid,否则唯一索引建不起来(宁可报错也不要静默丢数据)
duplicates = bind.execute(
sa.text(
f"SELECT uid FROM {TABLE} WHERE uid IS NOT NULL AND uid <> '' "
"GROUP BY uid HAVING COUNT(*) > 1 LIMIT 5"
)
).fetchall()
if duplicates:
names = ", ".join(str(row[0]) for row in duplicates)
raise RuntimeError(f"存在重复 uid({names}),先合并/清理这些订单再降级")
op.drop_index(INDEX, table_name=TABLE)
op.create_index(INDEX, TABLE, COLUMNS, unique=True)
+43 -5
View File
@@ -1033,6 +1033,45 @@ class FileAsset(Base):
)
class HuyaCollabShop(Base):
"""协作台账的**店铺字典**(新建协作单时点一下选,不用手打)。
几个取舍,改之前先看:
1. **订单里存的是店铺名字符串,不是 id**(`huya_external_orders.shop`)——
历史 Excel 数据就是文本,改成外键要动所有读写与导入。代价:改名要么级联
(`update_shop(..., rename_orders=True)`,默认开),要么只影响以后的单子;
2. **订单里出现过的店铺名会自动登记**(新建/编辑/粘贴导入都走
`ensure_shops_registered`),所以字典永远覆盖已有数据,筛选下拉不会漏;
3. **停用(`is_active=0`)≠ 删除**:只是不再出现在新建的选择里,历史订单与筛选照旧;
4. **不加乐观锁 version**:字典是十几行的低频维度,改名冲突靠审计看,不值得为它
再加一套版本号(台账订单行才需要,见 `HuyaExternalOrder`)。
"""
__tablename__ = "huya_collab_shops"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(
String(64), unique=True, nullable=False, index=True
)
# 越小越靠前(新建下拉、筛选下拉都按它排)。
sort_order: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
is_active: Mapped[bool] = mapped_column(
Boolean, default=True, server_default="1", index=True
)
remark: Mapped[str] = mapped_column(String(255), default="", server_default="")
created_by: Mapped[int | None] = mapped_column(
ForeignKey("users.id"), nullable=True
)
updated_by: Mapped[int | None] = mapped_column(
ForeignKey("users.id"), nullable=True
)
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=_utcnow)
updated_at: Mapped[datetime] = mapped_column(
UTCDateTime, default=_utcnow, onupdate=_utcnow
)
class HuyaExternalOrder(Base):
"""虎牙外部 CK 协作台账(替代 Excel 多人协同表)。
@@ -1067,11 +1106,10 @@ class HuyaExternalOrder(Base):
)
# ---- 外部 CK 本体(独立于 huya_accounts)----
# uid 唯一但可空:MySQL 唯一索引允许多个 NULL,于是「先建单、后贴 CK」与
# 「一个虎牙号只能挂一单」同时成立。
uid: Mapped[str | None] = mapped_column(
String(32), unique=True, nullable=True, index=True
)
# uid 可空、**不再唯一**(2026-10-16 用户拍板):同一个买家买两件常常继续用同一份 CK,
# 唯一索引会让第二单直接插不进去(迁移 0058 把唯一索引降成普通索引)。
# 「这个虎牙号还挂在哪些单上」只作为提示与审计信息,不做约束(见设计文档 §5.6)。
uid: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True)
yyuid: Mapped[str] = mapped_column(String(32), default="", server_default="")
hy_username: Mapped[str] = mapped_column(String(128), default="", server_default="")
nickname: Mapped[str] = mapped_column(String(128), default="", server_default="")
+126 -4
View File
@@ -24,6 +24,11 @@ from ...database import get_db
from ...deps import require_permission
from ...models import AuditLog, HuyaExternalOrder, User
from ...schemas import (
HuyaCollabCkPreviewOut,
HuyaCollabCkPreviewRequest,
HuyaCollabShopCreate,
HuyaCollabShopOut,
HuyaCollabShopUpdate,
HuyaExternalOrderAuditOut,
HuyaExternalOrderCkRequest,
HuyaExternalOrderCkStatusRequest,
@@ -52,7 +57,12 @@ def _http_error(exc: collab.CollabError) -> HTTPException:
def _serialize_one(db: Session, order: HuyaExternalOrder) -> dict:
return collab.serialize_order(order, users=collab.users_by_id(db, [order]))
clusters = collab.buyer_clusters(db, [order])
return collab.serialize_order(
order,
users=collab.users_by_id(db, [order]),
buyer_cluster=clusters.get(int(order.id)),
)
# --------------------------------------------------------------------------
@@ -89,10 +99,81 @@ def external_orders_shops(
db: Session = Depends(get_db),
_current: User = Depends(require_permission("huya:collab:view")),
):
"""已用过的店铺列表(筛选下拉)。"""
"""店铺名列表(现在就是**店铺字典**里的名字;筛选下拉用)。"""
return collab.list_shops(db)
# --------------------------------------------------------------------------
# 店铺字典(新建协作单时点选;也支持自定义输入后自动登记)
# --------------------------------------------------------------------------
@router.get("/collab-shops", response_model=list[HuyaCollabShopOut])
def list_collab_shops(
db: Session = Depends(get_db),
_current: User = Depends(require_permission("huya:collab:view")),
):
"""全部店铺(**含停用的**:管理弹窗要看到;新建下拉由前端按 `is_active` 过滤)。"""
shops = collab.list_shop_records(db)
names = collab.user_names(
db,
[shop.created_by for shop in shops] + [shop.updated_by for shop in shops],
)
return [collab.serialize_shop(shop, users=names) for shop in shops]
@router.post("/collab-shops", response_model=HuyaCollabShopOut)
def create_collab_shop(
req: HuyaCollabShopCreate,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""新增店铺(重名 409)。"""
try:
shop = collab.create_shop(
db, current, name=req.name, sort_order=req.sort_order, remark=req.remark
)
except collab.CollabError as exc:
raise _http_error(exc) from exc
return collab.serialize_shop(shop)
@router.put("/collab-shops/{shop_id}", response_model=HuyaCollabShopOut)
def update_collab_shop(
shop_id: int,
req: HuyaCollabShopUpdate,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""改店铺:名字 / 排序 / 停用启用 / 备注。
``rename_orders``(默认 true)为真时,改名会级联改掉已有订单里的旧名字。
"""
payload = req.model_dump(exclude_unset=True)
rename_orders = bool(payload.pop("rename_orders", True))
try:
shop = collab.get_shop(db, shop_id)
collab.update_shop(db, current, shop, payload, rename_orders=rename_orders)
except collab.CollabError as exc:
raise _http_error(exc) from exc
return collab.serialize_shop(shop)
@router.delete("/collab-shops/{shop_id}")
def delete_collab_shop(
shop_id: int,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:manage")),
):
"""从字典里删掉(**只给运营**)。不动订单:订单存的是店铺名,历史单子照旧。"""
try:
shop = collab.get_shop(db, shop_id)
snapshot = collab.delete_shop(db, current, shop)
except collab.CollabError as exc:
raise _http_error(exc) from exc
return {"success": True, "name": snapshot["name"]}
@router.get("/external-orders", response_model=HuyaExternalOrderListOut)
def list_external_orders(
shop: str = Query("", max_length=64),
@@ -119,8 +200,15 @@ def list_external_orders(
page_size=page_size,
)
names = collab.users_by_id(db, rows)
# 「同一买家的其它单」整页一次算完(按绑定账号 + 绑定角色软识别,只读派生)
clusters = collab.buyer_clusters(db, rows)
return {
"items": [collab.serialize_order(row, users=names) for row in rows],
"items": [
collab.serialize_order(
row, users=names, buyer_cluster=clusters.get(int(row.id))
)
for row in rows
],
"total": total,
"page": page,
"page_size": page_size,
@@ -133,7 +221,12 @@ def create_external_order(
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""新建一行(可同时贴一份外部 CK)。订单号唯一,重复会 409。"""
"""新建一行(可同时贴一份外部 CK)。订单号唯一,重复会 409。
``require_verified``(默认 true):人工新建表单的严格模式 —— 订单编号 / 店铺 / CK /
礼包名 / 主页截图必填,且 CK 必须探针通并读到绑定角色,否则 400
(粘贴导入走另一个接口,不受影响)。
"""
try:
order = collab.create_order(
db,
@@ -144,7 +237,9 @@ def create_external_order(
ck_text=req.ck_text,
bind_douyu_account=req.bind_douyu_account,
gift_name=req.gift_name,
home_images=req.home_images,
remark=req.remark,
require_verified=req.require_verified,
)
except collab.CollabError as exc:
raise _http_error(exc) from exc
@@ -167,6 +262,33 @@ def import_external_orders(
return collab.import_orders(db, current, req.text, owner_by_name=owners)
@router.post("/external-orders/ck/preview", response_model=HuyaCollabCkPreviewOut)
def preview_external_order_ck(
req: HuyaCollabCkPreviewRequest,
db: Session = Depends(get_db),
_current: User = Depends(require_permission("huya:collab:edit")),
):
"""粘贴 CK 后的**即时预览**(只读):探针 + 顺带拉活动礼包,一行都不写库。
新建弹窗据此决定能不能提交:``ok`` 且 ``has_role`` 才允许
(服务端在创建时还会再校验一次,见 ``POST /external-orders``)。
礼包用**工作台配置里的活动 SID**(``huya_config.sid``)拉:SID 就是活动 ID,
换活动就得换它,写死会出现「接口 200 但礼包列表是空的」(真机 2026-10-16)。
"""
try:
preview = collab.preview_ck(
req.ck_text,
with_gifts=req.with_gifts,
sid=collab.resolve_gifts_sid(db),
)
except collab.CollabError as exc:
raise _http_error(exc) from exc
# 「这个虎牙号已经在哪些单上」:允许同一份 CK 服务多单,这里只提示、不挡
preview["uid_used_by"] = collab.uids_in_use(db, preview.get("uid"))
return preview
@router.post("/external-orders/ck/verify", response_model=HuyaExternalOrderVerifyOut)
def verify_external_orders_ck(
req: HuyaExternalOrderVerifyRequest,
+110 -1
View File
@@ -1079,16 +1079,109 @@ class FileAssetOut(BaseModel):
size: int
class HuyaCollabShopCreate(BaseModel):
"""新增店铺(字典),新建协作单时就能点选。"""
name: str = Field(..., min_length=1, max_length=64)
sort_order: int = 0
remark: str = Field("", max_length=255)
class HuyaCollabShopUpdate(BaseModel):
"""改店铺:名字 / 排序 / 停用启用 / 备注(只提交要改的字段)。"""
name: str | None = Field(None, min_length=1, max_length=64)
sort_order: int | None = None
is_active: bool | None = None
remark: str | None = Field(None, max_length=255)
# 改名时是否把已有订单里的旧名字一起改掉(默认改,避免筛选里堆一堆改名前的老值)
rename_orders: bool = True
class HuyaCollabShopOut(BaseModel):
"""店铺字典项。"""
id: int
name: str
sort_order: int = 0
is_active: bool = True
remark: str = ""
created_by_name: str = ""
updated_by_name: str = ""
created_at: datetime | None = None
updated_at: datetime | None = None
# ---- 虎牙外部 CK 协作台账(见 docs/设计-虎牙外部CK协作台账.md) ----
class HuyaCollabGiftOut(BaseModel):
"""活动礼包(用一份有效 CK 现拉,不落库)。"""
product_id: str = ""
name: str = ""
price: int | None = None
remain_text: str = ""
stock: int | None = None
score: int | None = None
category_name: str = ""
icon: str = ""
class HuyaCollabCkPreviewRequest(BaseModel):
"""粘贴 CK 后的即时预览(只读:探针 + 顺带拉礼包)。"""
ck_text: str = Field(..., min_length=1)
with_gifts: bool = True
class HuyaCollabBuyerOrderOut(BaseModel):
"""「同一买家的其它单」里的一条(够列表/详情里一眼看清是哪单)。"""
id: int
order_no: str
shop: str = ""
gift_name: str = ""
uid: str = ""
status: str = "pending"
status_label: str = ""
created_at: datetime | None = None
class HuyaCollabCkPreviewOut(BaseModel):
"""预览结论:``ok`` 且 ``has_role`` 才允许提交新建。"""
ok: bool = False
message: str = ""
uid: str = ""
yyuid: str = ""
hy_username: str = ""
ck_status: str = "unverified"
bind_game_account: str = ""
bind_game_role: str = ""
bind_game_channel: str = ""
has_role: bool = False
gifts: list[HuyaCollabGiftOut] = Field(default_factory=list)
gifts_message: str = ""
# 这个虎牙号已经挂在哪些单上(放开 uid 唯一性后只是提示,不拦提交)
uid_used_by: list[HuyaCollabBuyerOrderOut] = Field(default_factory=list)
class HuyaExternalOrderCreate(BaseModel):
"""新建一行(可同时贴一份外部 CK)。"""
"""新建一行(可同时贴一份外部 CK)。
``require_verified``(默认开):人工新建表单的严格模式 —— 订单编号 / 店铺 / CK /
礼包名 / **主页截图**必填,且 CK 必须探针通并读到绑定角色;粘贴导入不走这条路。
"""
order_no: str = Field(..., min_length=1, max_length=64)
shop: str = Field("", max_length=64)
require_verified: bool = True
owner_id: int | None = None
ck_text: str = ""
bind_douyu_account: str = Field("", max_length=128)
gift_name: str = Field("", max_length=128)
# 主页截图(object_key 数组,第一张作封面)。绑定图不再收:绑定关系由探针自动读,
# 人工截图没有意义(2026-10-16 用户拍板)。
home_images: list[str] = Field(default_factory=list)
remark: str = Field("", max_length=512)
@@ -1175,6 +1268,19 @@ class HuyaExternalOrderVerifyOut(BaseModel):
results: list[HuyaExternalOrderVerifyItemOut] = Field(default_factory=list)
class HuyaCollabBuyerClusterOut(BaseModel):
"""同一买家的多单(软识别,只读派生)。
``key`` 是买家指纹的可读文本(``绑定账号 · 绑定角色``);``method`` 说明识别口径,
免得以后有人以为这是人工"软绑定"关系。
"""
key: str = ""
method: str = ""
total: int = 0
others: list[HuyaCollabBuyerOrderOut] = Field(default_factory=list)
class HuyaExternalOrderOut(BaseModel):
"""一行的完整视图。
@@ -1237,6 +1343,9 @@ class HuyaExternalOrderOut(BaseModel):
created_at: datetime | None = None
updated_at: datetime | None = None
# 同一买家的其它单(软识别:绑定账号 + 绑定角色相同;只读派生,没识别到就是 null)
buyer_cluster: HuyaCollabBuyerClusterOut | None = None
class HuyaExternalOrderListOut(BaseModel):
"""列表响应:共享视图,带上总数供分页使用。"""
+662 -45
View File
@@ -19,6 +19,7 @@
from __future__ import annotations
import dataclasses
import hashlib
import re
import time
from collections.abc import Mapping, Sequence
@@ -26,11 +27,15 @@ from datetime import UTC, datetime
from typing import Any
from loguru import logger
from sqlalchemy import func, or_
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
from ..models import HuyaExternalOrder, User
from ..huya_defaults import (
HUYA_DEFAULT_BIND_ACT_ID,
HUYA_DEFAULT_OUTER_ACT_ID,
HUYA_DEFAULT_SID,
)
from ..models import HuyaCollabShop, HuyaConfig, HuyaExternalOrder, User
from .audit_service import record_audit
from .file_service import build_file_url
from .file_storage import normalize_object_key
@@ -127,6 +132,7 @@ EDITABLE_FIELDS: tuple[str, ...] = (
FIELD_LIMITS = {
"order_no": 64,
"shop": 64,
"shop_remark": 255,
"bind_douyu_account": 128,
"bind_douyu_uid": 32,
"gift_name": 128,
@@ -137,6 +143,7 @@ FIELD_LIMITS = {
FIELD_LABELS = {
"order_no": "订单编号",
"shop": "店铺",
"shop_remark": "店铺备注",
"owner_id": "负责人",
"bind_douyu_account": "绑定斗鱼号",
"bind_douyu_uid": "绑定斗鱼号 uid",
@@ -149,7 +156,7 @@ IMAGE_SLOTS: tuple[str, ...] = ("bind_images", "home_images")
IMAGE_SLOT_LABELS = {
"bind_images": "绑定图",
"home_images": "主页账号图",
"home_images": "主页截图",
}
MAX_IMAGES_PER_SLOT = 20
@@ -303,23 +310,53 @@ def _assert_order_no_available(
)
def _assert_uid_available(
db: Session, uid: str | None, *, exclude_id: int | None = None
) -> None:
"""一个虎牙号只能挂一单(已确认 1 单 = 1 CK)。"""
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)
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)
other = query.first()
if other is not None:
raise CollabError(
f"虎牙号 {text} 已经挂在订单 {other.order_no}(第 {other.id} 行)",
status_code=409,
code="uid_conflict",
)
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:
@@ -481,10 +518,16 @@ def set_ck(
*,
commit: bool = True,
) -> dict:
"""贴入 / 换一份 CK(uid 冲突会挡住),状态回到「未验证」。"""
"""贴入 / 换一份 CK,状态回到「未验证」。
同一个虎牙号**允许**挂在多张单上(2026-10-16 用户拍板放开 uid 唯一性):
复用到的那些单号只记一笔审计,不挡操作 —— 挡了就变成"同一个买家的第二单建不了"。
"""
before = str(order.cookie or "")
detail = _apply_ck(order, cookie_text)
_assert_uid_available(db, order.uid, exclude_id=order.id)
reused = [
item.order_no for item in uid_used_orders(db, order.uid, exclude_id=order.id)
]
_touch(order, current)
record_audit(
db,
@@ -499,6 +542,8 @@ def set_ck(
# 凭据也不该多经一手(脱敏规则一改就是凭据泄漏)。
# 只留一个「内容是否变了」的布尔,够用来追溯换没换过。
"credential_changed": before != str(order.cookie or ""),
# 这个虎牙号还挂在哪些单上(同一份 CK 服务多单是允许的,留个痕)
"uid_reused_in": reused,
},
)
if commit:
@@ -667,6 +712,18 @@ PROBE_SNAPSHOT_FIELDS: tuple[str, ...] = (
"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
@@ -681,6 +738,12 @@ def _clip(value: Any, limit: int = _PROBE_PIECE_LIMIT) -> str:
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` 负责)。"""
@@ -775,6 +838,9 @@ def probe_ck(order: HuyaExternalOrder, *, client: Any | None = None) -> CkProbeR
实测支持这个判据:裸 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()
@@ -785,6 +851,12 @@ def probe_ck(order: HuyaExternalOrder, *, client: Any | None = None) -> CkProbeR
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(
@@ -806,7 +878,141 @@ def probe_ck(order: HuyaExternalOrder, *, client: Any | None = None) -> CkProbeR
msg = _clip(getattr(resp, "msg", ""))
suffix = f":{msg}" if msg else ""
return CkProbeResult(False, f"探针返回 status={status}{suffix}")
return CkProbeResult(True, _probe_success_message(resp), _probe_snapshot(resp))
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)} 个礼包"
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(
@@ -960,6 +1166,263 @@ def list_orders_by_ids(db: Session, ids: Sequence[int]) -> list[HuyaExternalOrde
return [by_id[value] for value in wanted if value in by_id]
# --------------------------------------------------------------------------
# 店铺字典(新建协作单时点一下选;也支持自定义输入)
# --------------------------------------------------------------------------
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))
# --------------------------------------------------------------------------
# CRUD
# --------------------------------------------------------------------------
@@ -975,34 +1438,91 @@ def create_order(
ck_text: str = "",
bind_douyu_account: str = "",
gift_name: str = "",
home_images: Any = None,
remark: str = "",
require_verified: bool = False,
client: Any | None = None,
commit: bool = True,
) -> HuyaExternalOrder:
"""新建一行(可同时贴 CK)。订单号唯一,重复直接报冲突。"""
"""新建一行(可同时贴 CK)。订单号唯一,重复直接报冲突。
**负责人不填就是当前登录用户**(创建人)—— 新建时不用选,谁建的单归谁;
要转给别人在列表里点一下改负责人即可(负责人不是权限边界,见设计文档 §7.1)。
``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")
order = HuyaExternalOrder(
order_no=order_no,
shop=clean_text(shop, field="shop"),
owner_id=_clean_owner_id(owner_id),
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=home_images_keys,
status=STATUS_PENDING,
ck_status=CK_UNVERIFIED,
created_by=current.id,
updated_by=current.id,
version=0,
)
if str(ck_text or "").strip():
uid_reused: list[str] = []
ck_provided = bool(str(ck_text or "").strip())
if ck_provided:
_apply_ck(order, ck_text)
_assert_uid_available(db, order.uid)
# 同一个虎牙号允许挂多单(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,
@@ -1013,6 +1533,11 @@ def create_order(
"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 []),
# 这个虎牙号还挂在哪些单上(允许同一份 CK 服务多单,留个痕)
"uid_reused_in": uid_reused,
},
)
if commit:
@@ -1060,6 +1585,10 @@ def update_order(
if not changes:
return {}
if "shop" in changes:
# 改成字典里没有的名字也允许(自定义),顺手登记,筛选下拉不会漏。
ensure_shops_registered(db, current, [changes["shop"]["new"]])
_touch(order, current)
record_audit(
db,
@@ -1160,9 +1689,11 @@ def list_orders(
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),
@@ -1205,38 +1736,40 @@ def summary(db: Session) -> dict:
}
def list_shops(db: Session) -> list[str]:
rows = (
db.query(HuyaExternalOrder.shop)
.filter(HuyaExternalOrder.shop != "")
.distinct()
.order_by(HuyaExternalOrder.shop)
.all()
)
return [str(row[0]) for row in rows]
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)。"""
ids = {
int(value)
for order in orders
for value in (order.owner_id, order.created_by, order.updated_by)
if value
}
if not ids:
return {}
rows = db.query(User.id, User.username).filter(User.id.in_(ids)).all()
return {int(row[0]): str(row[1]) for row in rows}
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
order: HuyaExternalOrder,
*,
users: Mapping[int, str] | None = None,
buyer_cluster: Mapping[str, Any] | None = None,
) -> dict:
"""把行对象转成接口输出。
图片存的是 object_key,URL 在这里现拼 —— 库里存 URL 会在换存储/改路径时
全部失效(order_site 踩过「存下来的签名链接过期后到处裂图」)。
``buyer_cluster`` = 同一买家的其它单(见 :func:`buyer_clusters`),由调用方
传进来(列表是一次查询算出整页的,别在这里按行查)。
"""
names = users or {}
bind_images = images_of(order, "bind_images")
@@ -1298,9 +1831,93 @@ def serialize_order(
"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,
}
# --------------------------------------------------------------------------
# 同一买家的多单(软识别:只读派生,不落库、不改 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 粘贴)
# --------------------------------------------------------------------------