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

feat: 协作台账支持归档并精简状态列

Browse Source
This commit is contained in:
yml2213
2026-10-19 13:04:27 +08:00
parent 375567cbd2
commit 781b27e63a
11 changed files with 1040 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/backend/migrations/versions/20261019_0060_huya_order_archive.py
+74
View File
@@ -0,0 +1,74 @@
"""虎牙协作台账增加归档字段。
归档是**可见性状态**，不是业务状态：`status=done`（抢购成功）保持不变，
只把 `archived_at` 置为当前时间，默认列表即不再展示；取消归档后恢复。
只归档终态「抢购成功」；归档不删除 CK、截图、积分快照或审计记录。
可重复执行：逐列检查，已存在就跳过。
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20261019_0060"
down_revision: str | None = "20261016_0059"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
TABLE = "huya_external_orders"
USERS = "users"
INDEX = "ix_huya_external_orders_archived_at"
FK = "fk_huya_external_orders_archived_by"
def _columns(bind) -> set[str]:
return {column["name"] for column in sa.inspect(bind).get_columns(TABLE)}
def upgrade() -> None:
bind = op.get_bind()
if TABLE not in set(sa.inspect(bind).get_table_names()):
return
columns = _columns(bind)
if "archived_at" not in columns:
op.add_column(TABLE, sa.Column("archived_at", sa.DateTime(), nullable=True))
if "archived_by" not in columns:
op.add_column(TABLE, sa.Column("archived_by", sa.Integer(), nullable=True))
if "archive_reason" not in columns:
op.add_column(
TABLE,
sa.Column(
"archive_reason",
sa.String(length=256),
nullable=False,
server_default="",
),
)
indexes = {index.get("name") for index in sa.inspect(bind).get_indexes(TABLE)}
if INDEX not in indexes:
op.create_index(INDEX, TABLE, ["archived_at"])
foreign_keys = {fk.get("name") for fk in sa.inspect(bind).get_foreign_keys(TABLE)}
if FK not in foreign_keys:
op.create_foreign_key(FK, TABLE, USERS, ["archived_by"], ["id"])
def downgrade() -> None:
bind = op.get_bind()
if TABLE not in set(sa.inspect(bind).get_table_names()):
return
indexes = {index.get("name") for index in sa.inspect(bind).get_indexes(TABLE)}
if INDEX in indexes:
op.drop_index(INDEX, table_name=TABLE)
foreign_keys = {fk.get("name") for fk in sa.inspect(bind).get_foreign_keys(TABLE)}
if FK in foreign_keys:
op.drop_constraint(FK, TABLE, type_="foreignkey")
columns = _columns(bind)
for column in ("archive_reason", "archived_by", "archived_at"):
if column in columns:
op.drop_column(TABLE, column)
web/backend/models.py
+12
View File
@@ -1169,6 +1169,18 @@ class HuyaExternalOrder(Base):
done_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)
void_reason: Mapped[str] = mapped_column(String(256), default="", server_default="")
# ---- 归档（与业务状态正交：归档后仍是「抢购成功」，只是默认不出现在主台账）----
# NULL = 未归档；软隐藏不删数据，取消归档即可恢复。
archived_at: Mapped[datetime | None] = mapped_column(
UTCDateTime, nullable=True, index=True
)
archived_by: Mapped[int | None] = mapped_column(
ForeignKey("users.id"), nullable=True
)
archive_reason: Mapped[str] = mapped_column(
String(256), default="", server_default=""
)
# ---- 只读查询快照（不写 huya_accounts）----
points: Mapped[int | None] = mapped_column(Integer, nullable=True)
elite_opened: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
web/backend/routers/huya/collab.py
+65 -2
View File
@@ -44,6 +44,8 @@ from ...schemas import (
HuyaExternalOrderActivityItemOut,
HuyaExternalOrderActivityQueryOut,
HuyaExternalOrderActivityQueryRequest,
HuyaExternalOrderArchiveOut,
HuyaExternalOrderArchiveRequest,
HuyaExternalOrderAuditOut,
HuyaExternalOrderCkExportOut,
HuyaExternalOrderCkExportRequest,
@@ -59,6 +61,7 @@ from ...schemas import (
HuyaExternalOrderStatusBatchRequest,
HuyaExternalOrderStatusRequest,
HuyaExternalOrderSummaryOut,
HuyaExternalOrderUnarchiveRequest,
HuyaExternalOrderUpdate,
HuyaExternalOrderVerifyOut,
HuyaExternalOrderVerifyRequest,
@@ -99,11 +102,15 @@ def _publish_change(current: User, *, scope: str, action: str) -> None:
@router.get("/external-orders/summary", response_model=HuyaExternalOrderSummaryOut)
def external_orders_summary(
include_void: bool = Query(False, description="是否包含已作废的行"),
archive_scope: str = Query(
"active", description="归档范围：active / archived / all"
),
db: Session = Depends(get_db),
_current: User = Depends(require_permission("huya:collab:view")),
):
"""页头统计条：各业务状态计数 + CK 失效数。"""
return collab.summary(db)
"""页头统计条：计数口径与当前列表的归档/作废过滤一致。"""
return collab.summary(db, include_void=include_void, archive_scope=archive_scope)
@router.get("/external-orders/owners", response_model=list[HuyaExternalOrderOwnerOut])
@@ -239,6 +246,9 @@ def list_external_orders(
gift_name: str = Query("", max_length=128),
keyword: str = Query("", max_length=128),
include_void: bool = Query(False, description="是否包含已作废的行（默认隐藏）"),
archive_scope: str = Query(
"active", description="归档范围：active / archived / all"
),
sort_by: str = Query("seq", max_length=32, description="排序字段（白名单）"),
sort_order: str = Query("asc", max_length=8, description="asc / desc"),
page: int = Query(1, ge=1),
@@ -256,6 +266,7 @@ def list_external_orders(
gift_name=gift_name,
keyword=keyword,
include_void=include_void,
archive_scope=archive_scope,
sort_by=sort_by,
sort_order=sort_order,
page=page,
@@ -393,6 +404,58 @@ def batch_update_external_order_status(
return result
@router.post("/external-orders/archive", response_model=HuyaExternalOrderArchiveOut)
def archive_external_orders(
req: HuyaExternalOrderArchiveRequest,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""归档「抢购成功」的活跃行（单条/勾选/按当前筛选），可带归档备注。
归档只改可见性，不改变业务状态；默认列表随即隐藏，归档视图仍可搜索和查看。
"""
try:
if req.archive_filtered:
result = collab.archive_orders_by_filter(
db,
current,
reason=req.reason,
expected_count=req.expected_count,
shop=req.shop,
owner_id=req.owner_id,
status=req.status,
ck_status=req.ck_status,
gift_name=req.gift_name,
keyword=req.keyword,
include_void=req.include_void,
)
else:
orders = collab.list_orders_for_archive(db, req.ids)
result = collab.archive_orders(db, current, orders, reason=req.reason)
except collab.CollabError as exc:
raise _http_error(exc) from exc
if result.get("archived"):
_publish_change(current, scope="orders", action="order.archive")
return result
@router.post("/external-orders/unarchive", response_model=HuyaExternalOrderArchiveOut)
def unarchive_external_orders(
req: HuyaExternalOrderUnarchiveRequest,
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:collab:edit")),
):
"""取消归档，恢复回主台账。"""
try:
orders = collab.list_orders_for_archive(db, req.ids)
result = collab.unarchive_orders(db, current, orders)
except collab.CollabError as exc:
raise _http_error(exc) from exc
if result.get("unarchived"):
_publish_change(current, scope="orders", action="order.unarchive")
return result
@router.post(
"/external-orders/activity/query", response_model=HuyaExternalOrderActivityQueryOut
)
web/backend/schemas.py
+51 -1
View File
@@ -1213,6 +1213,29 @@ class HuyaExternalOrderStatusRequest(BaseModel):
expected_version: int | None = None
class HuyaExternalOrderArchiveRequest(BaseModel):
"""归档抢购成功的订单（单条/勾选/按当前筛选）。"""
ids: list[int] = Field(default_factory=list)
reason: str = Field("", max_length=256)
# true 时忽略 ids，按下面的筛选条件归档所有匹配的「抢购成功」活跃行。
archive_filtered: bool = False
expected_count: int | None = Field(None, ge=0)
shop: str = Field("", max_length=64)
owner_id: int | None = None
status: str = Field("", max_length=16)
ck_status: str = Field("", max_length=16)
gift_name: str = Field("", max_length=128)
keyword: str = Field("", max_length=128)
include_void: bool = False
class HuyaExternalOrderUnarchiveRequest(BaseModel):
"""取消归档（恢复回主台账）。"""
ids: list[int] = Field(default_factory=list)
class HuyaExternalOrderCkRequest(BaseModel):
"""贴入/换 CK，或从 raw_cookie 重建。"""
@@ -1302,6 +1325,25 @@ class HuyaExternalOrderStatusBatchOut(BaseModel):
results: list[HuyaExternalOrderStatusBatchItemOut] = Field(default_factory=list)
class HuyaExternalOrderArchiveItemOut(BaseModel):
"""归档/取消归档里的一行结论。"""
id: int
order_no: str = ""
ok: bool = False
message: str = ""
class HuyaExternalOrderArchiveOut(BaseModel):
"""归档/取消归档汇总：``archived`` 或 ``unarchived`` + skipped = total。"""
total: int
archived: int = 0
unarchived: int = 0
skipped: int = 0
results: list[HuyaExternalOrderArchiveItemOut] = Field(default_factory=list)
class HuyaExternalOrderActivityQueryRequest(BaseModel):
"""勾选多行查积分 / 兑换记录（一次最多 50 条，串行带间隔）。
@@ -1503,6 +1545,13 @@ class HuyaExternalOrderOut(BaseModel):
done_at: datetime | None = None
void_reason: str = ""
# 归档是可见性状态，不影响上面的业务状态。
is_archived: bool = False
archived_at: datetime | None = None
archived_by: int | None = None
archived_by_name: str = ""
archive_reason: str = ""
points: int | None = None
elite_opened: bool | None = None
snapshot_at: datetime | None = None
@@ -1540,11 +1589,12 @@ class HuyaExternalOrderListOut(BaseModel):
class HuyaExternalOrderSummaryOut(BaseModel):
"""页头统计条。"""
"""页头统计条（口径与当前 archive_scope / include_void 一致）。"""
total: int
by_status: dict[str, int]
ck_invalid: int
archived_total: int = 0
class HuyaExternalOrderOwnerOut(BaseModel):
web/backend/services/huya_collab_service.py
+320 -15
View File
@@ -123,6 +123,12 @@ STATUS_TIMESTAMP_FIELDS = {
STATUS_DONE: "done_at",
}
# 归档（可见性）与业务状态正交：只允许归档已经抢购成功、不再需要日常操作的终态行。
ARCHIVE_SCOPES: tuple[str, ...] = ("active", "archived", "all")
ARCHIVABLE_STATUSES: tuple[str, ...] = (STATUS_DONE,)
# DB-only 的批量归档不需要像外部探针那样限 50 条。
ARCHIVE_BATCH_MAX = 5000
def is_forward_status(from_status: str, target: str) -> bool:
"""这一步算不算「向前」（人工切换允许往前跳步，回退必须显式声明）。
@@ -186,6 +192,7 @@ FIELD_LIMITS = {
"gift_name": 128,
"remark": 512,
"void_reason": 256,
"archive_reason": 256,
}
FIELD_LABELS = {
@@ -198,6 +205,7 @@ FIELD_LABELS = {
"gift_name": "礼包名",
"remark": "备注",
"void_reason": "作废原因",
"archive_reason": "归档备注",
}
IMAGE_SLOTS: tuple[str, ...] = ("bind_images", "home_images")
@@ -325,6 +333,16 @@ def _audit_target(order: HuyaExternalOrder) -> str:
return f"huya_external_order:{order.id}"
def assert_order_mutable(order: HuyaExternalOrder) -> None:
"""归档后整行只读；要改状态/CK/备注必须先取消归档。"""
if order.archived_at is not None:
raise CollabError(
"这单已归档，只能查看；需要修改请先取消归档",
status_code=409,
code="order_archived",
)
def assert_version(order: HuyaExternalOrder, expected_version: int | None) -> None:
"""乐观锁：客户端提交时带的版本必须与库里一致，否则说明别人先改了。"""
if expected_version is None:
@@ -524,6 +542,7 @@ def set_status(
"""
if target not in ORDER_STATUSES:
raise CollabError(f"未知状态：{target}", code="status_unknown")
assert_order_mutable(order)
if target == order.status:
return {}
@@ -681,6 +700,7 @@ def set_ck(
同一个虎牙号**允许**挂在多张单上（2026-10-16 用户拍板放开 uid 唯一性）：
复用到的那些单号只记一笔审计，不挡操作 —— 挡了就变成"同一个买家的第二单建不了"。
"""
assert_order_mutable(order)
before = str(order.cookie or "")
detail = _apply_ck(order, cookie_text)
reused = [
@@ -726,6 +746,7 @@ def set_ck_status(
"""
if status not in CK_STATUSES:
raise CollabError(f"未知 CK 状态：{status}", code="ck_status_unknown")
assert_order_mutable(order)
previous = order.ck_status
order.ck_status = status
order.ck_check_message = clean_text(
@@ -756,6 +777,7 @@ def rebuild_ck(
db: Session, current: User | None, order: HuyaExternalOrder, *, commit: bool = True
) -> dict:
"""用 ``raw_cookie`` 重建网页运行 CK（复用 huya_service 的纯函数）。"""
assert_order_mutable(order)
if not str(order.raw_cookie or "").strip():
raise CollabError(
"这行没有 App 原始凭据，无法重建网页 CK（请重新贴一份）",
@@ -826,6 +848,7 @@ def set_images(
commit: bool = True,
) -> dict:
"""整槽替换图片（前端上传完拿到 object_key 后一次性提交）。"""
assert_order_mutable(order)
if slot not in IMAGE_SLOTS:
raise CollabError(f"未知图片槽位：{slot}", code="image_slot_unknown")
before = images_of(order, slot)
@@ -1409,6 +1432,7 @@ def query_order_activity(
"""
if scope not in ACTIVITY_SCOPES:
raise CollabError(f"未知的查询范围：{scope}", code="activity_scope_unknown")
assert_order_mutable(order)
want_score = scope in (ACTIVITY_SCOPE_BOTH, ACTIVITY_SCOPE_SCORE)
want_records = scope in (ACTIVITY_SCOPE_BOTH, ACTIVITY_SCOPE_RECORDS)
@@ -1734,6 +1758,7 @@ def verify_ck(
- 人工填的那些列（含 ``bind_douyu_account``）**一律不碰**；
- CK 失效**不改业务状态**（设计文档 §4.2：已绑定/已充值的单子业务上可能仍然有效）。
"""
assert_order_mutable(order)
if not str(order.cookie or "").strip():
raise CollabError("这行还没有 CK，先贴一份再验证", code="no_cookie")
if order.status == STATUS_VOID:
@@ -1868,6 +1893,21 @@ 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]
def list_orders_for_archive(db: Session, ids: Sequence[int]) -> list[HuyaExternalOrder]:
"""按 id 取归档/取消归档目标（不受探针 50 条上限约束）。"""
wanted = list(dict.fromkeys(int(value) for value in ids))
if not wanted:
raise CollabError("请先选择要归档的行", code="archive_empty")
if len(wanted) > ARCHIVE_BATCH_MAX:
raise CollabError(
f"一次最多处理 {ARCHIVE_BATCH_MAX} 行，请分批操作",
code="archive_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"
@@ -2395,6 +2435,7 @@ def update_order(
乐观锁：``expected_version`` 与库里不一致 → 409，绝不静默覆盖
（这正是 Excel 协同最痛的地方）。
"""
assert_order_mutable(order)
assert_version(order, expected_version)
changes: dict[str, dict[str, Any]] = {}
@@ -2446,6 +2487,7 @@ def delete_order(
只删行，**不删对象存储里的图**：撤图不可逆，交给孤儿清理脚本按保留期处理
（设计文档 §6.4）。
"""
assert_order_mutable(order)
snapshot = {
"order_no": order.order_no,
"uid": order.uid or "",
@@ -2509,7 +2551,13 @@ ORDER_SORT_KEYS: tuple[str, ...] = (
)
def list_orders(
def normalize_archive_scope(value: str | None) -> str:
"""归档可见范围：默认只看活跃行，归档视图只返回已归档行。"""
scope = str(value or "active").strip().lower()
return scope if scope in ARCHIVE_SCOPES else "active"
def _filtered_orders_query(
db: Session,
*,
shop: str = "",
@@ -2519,15 +2567,9 @@ def list_orders(
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。"""
archive_scope: str = "active",
):
"""列表与「按筛选归档」共用同一套筛选，保证两边口径一致。"""
query = db.query(HuyaExternalOrder)
if shop:
query = query.filter(HuyaExternalOrder.shop == shop)
@@ -2542,6 +2584,11 @@ def list_orders(
query = query.filter(HuyaExternalOrder.gift_name == skin)
if not include_void:
query = query.filter(HuyaExternalOrder.status != STATUS_VOID)
scope = normalize_archive_scope(archive_scope)
if scope == "active":
query = query.filter(HuyaExternalOrder.archived_at.is_(None))
elif scope == "archived":
query = query.filter(HuyaExternalOrder.archived_at.is_not(None))
text = (keyword or "").strip()
if text:
like = f"%{text}%"
@@ -2560,6 +2607,41 @@ def list_orders(
HuyaExternalOrder.remark.like(like),
)
)
return query
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,
archive_scope: str = "active",
sort_by: str = "seq",
sort_order: str = "asc",
page: int = 1,
page_size: int = 50,
) -> tuple[list[HuyaExternalOrder], int]:
"""共享视图：默认**不按人过滤**（负责人只是筛选维度，不是权限边界）。
默认按 ``id`` 升序（即创建顺序），新建的行永远追加在最后；表头排序走白名单，
不允许前端把任意列名拼进 SQL。归档行只有在 ``archive_scope=archived/all`` 时出现。
"""
query = _filtered_orders_query(
db,
shop=shop,
owner_id=owner_id,
status=status,
ck_status=ck_status,
gift_name=gift_name,
keyword=keyword,
include_void=include_void,
archive_scope=archive_scope,
)
total = query.count()
page = max(1, int(page or 1))
@@ -2594,23 +2676,236 @@ def list_orders(
return rows, total
def summary(db: Session) -> dict:
"""页头统计条：各状态计数 + CK 失效数。"""
def archive_orders(
db: Session,
current: User | None,
orders: Sequence[HuyaExternalOrder],
*,
reason: str = "",
commit: bool = True,
) -> dict:
"""归档抢购成功的行：只改可见性字段，不影响业务状态。
已有归档的行、非「抢购成功」的行都跳过并逐条回原因；归档备注写入每行，
取消归档时清空，审计里始终保留原始备注。
"""
rows = list(orders)
if not rows:
raise CollabError("请先选择要归档的行", code="archive_empty")
if len(rows) > ARCHIVE_BATCH_MAX:
raise CollabError(
f"一次最多归档 {ARCHIVE_BATCH_MAX} 行，请分批操作",
code="archive_too_large",
)
note = clean_text(reason, field="archive_reason", max_length=256)
now = _now()
results: list[dict] = []
archived = 0
for order in rows:
if order.archived_at is not None:
results.append(
{
"id": int(order.id),
"order_no": order.order_no,
"ok": False,
"message": "已经归档",
}
)
continue
if order.status not in ARCHIVABLE_STATUSES:
results.append(
{
"id": int(order.id),
"order_no": order.order_no,
"ok": False,
"message": "只有「抢购成功」的单可以归档",
}
)
continue
order.archived_at = now
order.archived_by = current.id if current else None
order.archive_reason = note
_touch(order, current)
record_audit(
db,
current,
action="huya:collab:archive",
target=_audit_target(order),
detail={
"order_no": order.order_no,
"status": order.status,
"reason": note,
},
)
archived += 1
results.append(
{
"id": int(order.id),
"order_no": order.order_no,
"ok": True,
"message": "已归档",
}
)
if commit:
db.commit()
for order in rows:
db.refresh(order)
return {
"total": len(rows),
"archived": archived,
"unarchived": 0,
"skipped": len(rows) - archived,
"results": results,
}
def archive_orders_by_filter(
db: Session,
current: User | None,
*,
reason: str = "",
expected_count: int | None = None,
shop: str = "",
owner_id: int | None = None,
status: str = "",
ck_status: str = "",
gift_name: str = "",
keyword: str = "",
include_void: bool = False,
) -> dict:
"""把当前筛选范围内的「抢购成功」活跃行一次归档。
``expected_count`` 是前端确认弹窗里展示的数量；执行前再数一次，避免确认期间
新完成/被其他人归档导致多隐藏了行。
"""
query = _filtered_orders_query(
db,
shop=shop,
owner_id=owner_id,
status=status,
ck_status=ck_status,
gift_name=gift_name,
keyword=keyword,
include_void=include_void,
archive_scope="active",
).filter(HuyaExternalOrder.status == STATUS_DONE)
total = query.count()
if expected_count is not None and int(expected_count) != total:
raise CollabError(
f"匹配数量已变化（确认时 {expected_count} 条，现在 {total} 条），请刷新后重新确认",
status_code=409,
code="archive_count_changed",
)
if total > ARCHIVE_BATCH_MAX:
raise CollabError(
f"一次最多归档 {ARCHIVE_BATCH_MAX} 行，请缩小筛选范围",
code="archive_too_large",
)
rows = query.order_by(HuyaExternalOrder.id.asc()).all()
return archive_orders(db, current, rows, reason=reason)
def unarchive_orders(
db: Session,
current: User | None,
orders: Sequence[HuyaExternalOrder],
*,
commit: bool = True,
) -> dict:
"""取消归档，恢复回主台账；归档备注清空但审计保留。"""
rows = list(orders)
if not rows:
raise CollabError("请先选择要取消归档的行", code="unarchive_empty")
if len(rows) > ARCHIVE_BATCH_MAX:
raise CollabError(
f"一次最多取消归档 {ARCHIVE_BATCH_MAX} 行，请分批操作",
code="unarchive_too_large",
)
results: list[dict] = []
unarchived = 0
for order in rows:
if order.archived_at is None:
results.append(
{
"id": int(order.id),
"order_no": order.order_no,
"ok": False,
"message": "没有归档",
}
)
continue
previous_reason = order.archive_reason or ""
order.archived_at = None
order.archived_by = None
order.archive_reason = ""
_touch(order, current)
record_audit(
db,
current,
action="huya:collab:unarchive",
target=_audit_target(order),
detail={
"order_no": order.order_no,
"status": order.status,
"reason": previous_reason,
},
)
unarchived += 1
results.append(
{
"id": int(order.id),
"order_no": order.order_no,
"ok": True,
"message": "已取消归档",
}
)
if commit:
db.commit()
for order in rows:
db.refresh(order)
return {
"total": len(rows),
"archived": 0,
"unarchived": unarchived,
"skipped": len(rows) - unarchived,
"results": results,
}
def summary(
db: Session, *, include_void: bool = False, archive_scope: str = "active"
) -> dict:
"""页头统计条：统计口径与当前列表的归档/作废过滤保持一致。"""
scope = normalize_archive_scope(archive_scope)
query = db.query(HuyaExternalOrder)
if not include_void:
query = query.filter(HuyaExternalOrder.status != STATUS_VOID)
if scope == "active":
query = query.filter(HuyaExternalOrder.archived_at.is_(None))
elif scope == "archived":
query = query.filter(HuyaExternalOrder.archived_at.is_not(None))
status_rows = (
db.query(HuyaExternalOrder.status, func.count(HuyaExternalOrder.id))
query.with_entities(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))
query.with_entities(func.count(HuyaExternalOrder.id))
.filter(HuyaExternalOrder.ck_status == CK_INVALID)
.scalar()
)
archived_total = (
db.query(func.count(HuyaExternalOrder.id))
.filter(HuyaExternalOrder.archived_at.is_not(None))
.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),
"archived_total": int(archived_total or 0),
}
@@ -2630,7 +2925,12 @@ def users_by_id(db: Session, orders: Sequence[HuyaExternalOrder]) -> dict[int, s
[
value
for order in orders
for value in (order.owner_id, order.created_by, order.updated_by)
for value in (
order.owner_id,
order.created_by,
order.updated_by,
order.archived_by,
)
],
)
@@ -2710,6 +3010,11 @@ def serialize_order(
"scripted_at": order.scripted_at,
"done_at": order.done_at,
"void_reason": order.void_reason or "",
"is_archived": order.archived_at is not None,
"archived_at": order.archived_at,
"archived_by": order.archived_by,
"archived_by_name": name_of(order.archived_by),
"archive_reason": order.archive_reason or "",
"points": order.points,
"elite_opened": order.elite_opened,
"snapshot_at": order.snapshot_at,