"""微博外部单协作台账服务。 一行 = 一单，引用现有 ``weibo_accounts`` 中的外部 CK 账号，不复制凭据。 账号可以同时服务多单；余额核账按账号分组。 """ from __future__ import annotations import hashlib import json from collections.abc import Mapping, Sequence from datetime import UTC, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import func, or_ from sqlalchemy.orm import Session from core.weibo import WeiboClient, WeiboError from ..models import ( AuditLog, User, WeiboAccount, WeiboCollabGoods, WeiboCollabShop, WeiboExternalOrder, WeiboExternalOrderMetric, ) from .audit_service import record_audit from .weibo_service import query_weibo_mall, refresh_weibo_account STATUS_INCOMPLETE = "incomplete" STATUS_TO_RECHARGE = "to_recharge" STATUS_RECHARGED = "recharged" STATUS_EXCHANGE_PENDING = "exchange_pending" STATUS_DONE = "done" STATUS_WEIDOU_ERROR = "weidou_error" STATUS_FAILED = "failed" STATUS_VOID = "void" ORDER_STATUSES = ( STATUS_INCOMPLETE, STATUS_TO_RECHARGE, STATUS_RECHARGED, STATUS_EXCHANGE_PENDING, STATUS_DONE, STATUS_WEIDOU_ERROR, STATUS_FAILED, STATUS_VOID, ) ORDER_STATUS_LABELS = { STATUS_INCOMPLETE: "待补资料", STATUS_TO_RECHARGE: "待充值", STATUS_RECHARGED: "已充值", STATUS_EXCHANGE_PENDING: "兑换待确认", STATUS_DONE: "兑换成功", STATUS_WEIDOU_ERROR: "微豆异常", STATUS_FAILED: "兑换失败", STATUS_VOID: "已作废", } STATUS_FLOW = ( STATUS_INCOMPLETE, STATUS_TO_RECHARGE, STATUS_RECHARGED, STATUS_EXCHANGE_PENDING, STATUS_DONE, ) STAGE_INDEX = {status: index for index, status in enumerate(STATUS_FLOW)} NEXT_STATUS = { STATUS_INCOMPLETE: STATUS_TO_RECHARGE, STATUS_TO_RECHARGE: STATUS_RECHARGED, STATUS_RECHARGED: STATUS_EXCHANGE_PENDING, STATUS_EXCHANGE_PENDING: STATUS_DONE, } BALANCE_GROUP_STATUSES = ( STATUS_TO_RECHARGE, STATUS_RECHARGED, STATUS_WEIDOU_ERROR, ) VERDICT_SUCCESS = "success" VERDICT_FAILED = "failed" VERDICT_UNKNOWN = "unknown" VERDICT_LABELS = { VERDICT_SUCCESS: "兑换成功", VERDICT_FAILED: "兑换失败", VERDICT_UNKNOWN: "无法判定", } ARCHIVE_SCOPES = ("active", "archived", "all") ARCHIVABLE_STATUSES = (STATUS_DONE,) MAX_IMAGES = 20 CHECK_BATCH_MAX = 50 ARCHIVE_BATCH_MAX = 5000 DEFAULT_WEIDOU_THRESHOLD = 100 class CollabError(RuntimeError): def __init__(self, detail: str, *, status_code: int = 400, code: str = ""): super().__init__(detail) self.detail = detail self.status_code = status_code self.code = code def _now() -> datetime: return datetime.now(UTC) def _to_int(value: Any) -> int | None: if value is None or value == "": return None try: return int(float(value)) except (TypeError, ValueError): return None def _text(value: Any, *, max_length: int = 512) -> str: return str(value or "").strip()[:max_length] def _json_safe(value: Any) -> Any: if isinstance(value, dict): return {str(key): _json_safe(item) for key, item in value.items()} if isinstance(value, list): return [_json_safe(item) for item in value] if isinstance(value, (str, int, float, bool)) or value is None: return value return str(value) def normalize_images(value: Any) -> list[str]: if value is None: return [] if not isinstance(value, list): raise CollabError("图片必须是 key 数组", code="images_invalid") result: list[str] = [] for item in value: key = _text(item, max_length=512) if key and key not in result: result.append(key) if len(result) > MAX_IMAGES: raise CollabError(f"最多 {MAX_IMAGES} 张图片", code="images_too_many") return result def _users_by_id(db: Session, ids: set[int]) -> dict[int, str]: clean = {int(item) for item in ids if item} if not clean: return {} return { int(user_id): str(username) for user_id, username in db.query(User.id, User.username) .filter(User.id.in_(clean)) .all() } def _accounts_by_id(db: Session, ids: set[int]) -> dict[int, WeiboAccount]: clean = {int(item) for item in ids if item} if not clean: return {} return { int(account.id): account for account in db.query(WeiboAccount).filter(WeiboAccount.id.in_(clean)).all() } def _touch(order: WeiboExternalOrder, current: User | None) -> None: order.updated_by = current.id if current else None order.version = int(order.version or 0) + 1 def _audit_target(order: WeiboExternalOrder) -> str: return f"weibo_external_order:{order.id}" def _audit( db: Session, current: User | None, order: WeiboExternalOrder, action: str, detail: Mapping[str, Any], ) -> None: record_audit( db, current, action=action, target=_audit_target(order), detail=detail, ) def get_order(db: Session, order_id: int) -> WeiboExternalOrder: order = ( db.query(WeiboExternalOrder).filter(WeiboExternalOrder.id == order_id).first() ) if order is None: raise CollabError("协作单不存在", status_code=404, code="order_not_found") return order def list_orders_by_ids(db: Session, ids: Sequence[int]) -> list[WeiboExternalOrder]: clean = [int(item) for item in ids if int(item) > 0] if not clean: return [] return ( db.query(WeiboExternalOrder) .filter(WeiboExternalOrder.id.in_(clean)) .order_by(WeiboExternalOrder.id.asc()) .all() ) def _missing_for_recharge(order: WeiboExternalOrder) -> list[str]: missing: list[str] = [] if not order.shop: missing.append("店铺") if not order.account_id: missing.append("微博账号") if not order.game_role_name: missing.append("绑定角色（先校验账号）") if not order.goods_name: missing.append("商品") if not order.required_weidou: missing.append("所需微豆") return missing def _maybe_advance_to_recharge( db: Session, current: User, order: WeiboExternalOrder, ) -> None: if order.status != STATUS_INCOMPLETE or _missing_for_recharge(order): return old = order.status order.status = STATUS_TO_RECHARGE _touch(order, current) _audit( db, current, order, "weibo:collab:status_auto", {"from": old, "to": order.status, "reason": "资料齐全"}, ) def _account_for_order(db: Session, order: WeiboExternalOrder) -> WeiboAccount: account = ( db.query(WeiboAccount).filter(WeiboAccount.id == order.account_id).first() if order.account_id else None ) if account is None: raise CollabError("订单关联的微博账号不存在", code="account_not_found") if not account.is_enabled: raise CollabError("订单关联的微博账号已停用", code="account_disabled") return account def _enabled_account(db: Session, account_id: int) -> WeiboAccount: account = db.query(WeiboAccount).filter(WeiboAccount.id == account_id).first() if account is None: raise CollabError("微博账号不存在", code="account_not_found") if not account.is_enabled: raise CollabError("微博账号已停用", code="account_disabled") return account def _apply_account_snapshot( order: WeiboExternalOrder, account: WeiboAccount, *, strict_role: bool, ) -> None: role = _text(account.game_role_name, max_length=128) if strict_role and order.game_role_name and role != order.game_role_name: raise CollabError( f"账号绑定角色已从「{order.game_role_name}」变为「{role or '未绑定'}」，请人工确认后更新订单", code="role_changed", ) order.account_id = account.id order.account_uid = _text(account.uid, max_length=32) order.account_nickname = _text(account.nickname, max_length=128) order.game_role_name = role order.bind_snapshot_at = _now() def verify_order_account( db: Session, current: User, order: WeiboExternalOrder, *, activity: str = "", commit: bool = True, ) -> WeiboAccount: if order.archived_at is not None: raise CollabError("已归档订单不能校验账号", code="order_archived") account = _account_for_order(db, order) snapshot = refresh_weibo_account(db, account, activity=activity, commit=False) _apply_account_snapshot(order, account, strict_role=True) if not snapshot.is_bound: raise CollabError("微博账号尚未绑定游戏角色", code="role_unbound") _touch(order, current) _audit( db, current, order, "weibo:collab:verify", { "account_id": account.id, "uid": account.uid, "ck_status": account.ck_check_status, "role": account.game_role_name, "balance": account.weidou_balance, }, ) _maybe_advance_to_recharge(db, current, order) if commit: db.commit() db.refresh(order) return account def _resolve_goods( db: Session, *, goods_id: int | None, goods_name: str, goods_image: str, required_weidou: int | None, ) -> tuple[int | None, str, str, int | None]: name = _text(goods_name, max_length=128) image = _text(goods_image, max_length=512) price = _to_int(required_weidou) catalog = None if name: catalog = ( db.query(WeiboCollabGoods).filter(WeiboCollabGoods.name == name).first() ) elif goods_id: catalog = ( db.query(WeiboCollabGoods) .filter(WeiboCollabGoods.goods_id == goods_id) .first() ) if catalog is not None: name = name or catalog.name image = image or catalog.goods_image goods_id = goods_id or catalog.goods_id if price is None: price = int(catalog.price or 0) return goods_id, name, image, price def create_order( db: Session, current: User, *, order_no: str, shop: str = "", owner_id: int | None = None, account_id: int | None = None, goods_id: int | None = None, goods_name: str = "", goods_image: str = "", required_weidou: int | None = None, home_images: Any = None, remark: str = "", draft: bool = False, require_verified: bool = True, activity: str = "", ) -> WeiboExternalOrder: order_text = _text(order_no, max_length=64) if not order_text: raise CollabError("订单号不能为空", code="order_no_required") if ( db.query(WeiboExternalOrder) .filter(WeiboExternalOrder.order_no == order_text) .first() ): raise CollabError("订单号已存在", status_code=409, code="order_no_exists") goods_id, goods_name, goods_image, required_weidou = _resolve_goods( db, goods_id=goods_id, goods_name=goods_name, goods_image=goods_image, required_weidou=required_weidou, ) order = WeiboExternalOrder( order_no=order_text, shop=_text(shop, max_length=64), owner_id=int(owner_id) if owner_id is not None else int(current.id), goods_id=goods_id, goods_name=goods_name, goods_image=goods_image, required_weidou=required_weidou, home_images=normalize_images(home_images), remark=_text(remark, max_length=512), status=STATUS_INCOMPLETE, weidou_threshold=DEFAULT_WEIDOU_THRESHOLD, created_by=current.id, updated_by=current.id, version=0, ) db.add(order) db.flush() if account_id: account = _enabled_account(db, account_id) _apply_account_snapshot(order, account, strict_role=False) if order.shop: ensure_shops_registered(db, current, [order.shop]) _audit( db, current, order, "weibo:collab:create", { "order_no": order.order_no, "shop": order.shop, "account_id": order.account_id, "uid": order.account_uid, "goods_name": order.goods_name, "required_weidou": order.required_weidou, "draft": draft, }, ) if not draft and require_verified: missing = _missing_for_recharge(order) if missing: raise CollabError( f"下面这几项必须填：{'、'.join(missing)}", code="create_field_required", ) verify_order_account(db, current, order, activity=activity, commit=False) _maybe_advance_to_recharge(db, current, order) db.commit() db.refresh(order) return order def update_order( db: Session, current: User, order: WeiboExternalOrder, payload: Mapping[str, Any], *, expected_version: int | None = None, ) -> dict[str, Any]: _assert_mutable(order) _assert_version(order, expected_version) changes: dict[str, dict[str, Any]] = {} editable = { "order_no", "shop", "owner_id", "goods_id", "goods_name", "goods_image", "required_weidou", "remark", } for field in editable: if field not in payload: continue old = getattr(order, field) value = payload[field] if field == "owner_id": new = int(value) if value is not None else None elif field in ("goods_id", "required_weidou"): new = _to_int(value) elif field == "remark" or field == "goods_image": new = _text(value, max_length=512) elif field == "goods_name": new = _text(value, max_length=128) elif field == "order_no": new = _text(value, max_length=64) else: new = _text(value, max_length=64) if new != old: changes[field] = {"old": old, "new": new} setattr(order, field, new) if "home_images" in payload: images = normalize_images(payload.get("home_images")) if images != list(order.home_images or []): changes["home_images"] = { "old": len(order.home_images or []), "new": len(images), } order.home_images = images if order.order_no != _text(order.order_no, max_length=64): raise CollabError("订单号格式不正确", code="order_no_invalid") duplicate = ( db.query(WeiboExternalOrder.id) .filter( WeiboExternalOrder.order_no == order.order_no, WeiboExternalOrder.id != order.id, ) .first() ) if duplicate: raise CollabError("订单号已存在", status_code=409, code="order_no_exists") if changes: if "shop" in changes and order.shop: ensure_shops_registered(db, current, [order.shop]) _touch(order, current) _audit(db, current, order, "weibo:collab:update", changes) _maybe_advance_to_recharge(db, current, order) db.commit() db.refresh(order) return changes def set_order_account( db: Session, current: User, order: WeiboExternalOrder, account_id: int, *, expected_version: int | None = None, activity: str = "", ) -> WeiboAccount: _assert_mutable(order) _assert_version(order, expected_version) account = _enabled_account(db, account_id) order.account_id = account.id order.game_role_name = "" verify_order_account(db, current, order, activity=activity, commit=False) db.commit() db.refresh(order) return account def _assert_version(order: WeiboExternalOrder, expected_version: int | None) -> None: if expected_version is None: return if int(expected_version) != int(order.version or 0): raise CollabError( "订单已被其他人修改，请刷新后重试", status_code=409, code="version_conflict", ) def _assert_mutable(order: WeiboExternalOrder) -> None: if order.archived_at is not None: raise CollabError("已归档订单只读，请先取消归档", code="order_archived") def set_status( db: Session, current: User, order: WeiboExternalOrder, target: str, *, reason: str = "", rollback: bool = False, allow_check: bool = False, commit: bool = True, ) -> dict[str, Any]: _assert_mutable(order) if target not in ORDER_STATUSES: raise CollabError("非法状态", code="status_invalid") if target in (STATUS_RECHARGED, STATUS_WEIDOU_ERROR) and not allow_check: raise CollabError( "「已充值 / 微豆异常」必须通过余额核账写入", code="balance_check_required", ) old = order.status if old == target: return {"from": old, "to": target} if old == STATUS_VOID: raise CollabError("已作废订单不能直接改回流程内", code="order_void") if target == STATUS_VOID: clean_reason = _text(reason, max_length=256) if not clean_reason: raise CollabError("作废必须填写原因", code="void_reason_required") order.status = target order.void_reason = clean_reason _touch(order, current) _audit( db, current, order, "weibo:collab:void", {"from": old, "reason": clean_reason}, ) if commit: db.commit() db.refresh(order) return {"from": old, "to": target} if ( target in STAGE_INDEX and old in STAGE_INDEX and STAGE_INDEX[target] < STAGE_INDEX[old] and not rollback ): raise CollabError("回退状态必须显式勾选「回退」", code="status_not_forward") if target == STATUS_TO_RECHARGE: missing = _missing_for_recharge(order) if missing: raise CollabError( f"进入待充值前还缺：{'、'.join(missing)}", code="recharge_precondition", ) now = _now() if target == STATUS_TO_RECHARGE: order.recharged_at = None order.exchange_started_at = None order.done_at = None elif target in (STATUS_RECHARGED, STATUS_WEIDOU_ERROR): if order.recharged_at is None: order.recharged_at = now if rollback: order.exchange_started_at = None order.done_at = None elif target == STATUS_EXCHANGE_PENDING and not order.recharged_at: raise CollabError("请先完成余额核账", code="recharge_check_required") elif target == STATUS_EXCHANGE_PENDING: if order.exchange_started_at is None: order.exchange_started_at = now order.done_at = None elif target == STATUS_DONE: if order.exchange_started_at is None: order.exchange_started_at = now order.done_at = now elif target == STATUS_FAILED: order.done_at = None order.status = target if target != STATUS_VOID: order.void_reason = "" _touch(order, current) _audit( db, current, order, "weibo:collab:status", {"from": old, "to": target, "rollback": rollback, "reason": reason}, ) if commit: db.commit() db.refresh(order) return {"from": old, "to": target} def set_status_batch( db: Session, current: User, orders: Sequence[WeiboExternalOrder], target: str, *, reason: str = "", rollback: bool = False, ) -> dict[str, Any]: rows = list(orders) if not rows: raise CollabError("请先勾选订单", code="status_batch_empty") if len(rows) > CHECK_BATCH_MAX: raise CollabError(f"一次最多操作 {CHECK_BATCH_MAX} 条", code="batch_too_large") results: list[dict[str, Any]] = [] updated = 0 for order in rows: try: with db.begin_nested(): set_status( db, current, order, target, reason=reason, rollback=rollback, allow_check=False, commit=False, ) updated += 1 results.append( { "id": order.id, "order_no": order.order_no, "ok": True, "message": "", } ) except CollabError as exc: results.append( { "id": order.id, "order_no": order.order_no, "ok": False, "message": exc.detail, } ) db.commit() return { "total": len(rows), "updated": updated, "failed": len(rows) - updated, "target": target, "target_label": ORDER_STATUS_LABELS.get(target, target), "results": results, } def lookup_account( db: Session, account_id: int, *, with_goods: bool, activity: str = "", ) -> dict[str, Any]: account = db.query(WeiboAccount).filter(WeiboAccount.id == account_id).first() if account is None: raise CollabError("微博账号不存在", code="account_not_found") goods: list[dict[str, Any]] = [] ok = account.is_enabled message = "" if ok else "账号已停用" if ok and with_goods: try: data = query_weibo_mall(db, account, activity=activity) for item in data.get("list") or []: if not isinstance(item, dict): continue goods_id = _to_int(item.get("id")) if not goods_id: continue goods.append( { "id": goods_id, "goods_name": _text(item.get("goods_name"), max_length=128), "goods_image": _text(item.get("goods_image"), max_length=512), "price": _to_int(item.get("price")) or 0, "remain_count": _to_int(item.get("remain_count")), "is_sold_out": bool(item.get("is_sold_out")), } ) except (WeiboError, ValueError) as exc: ok = False message = str(exc) orders_using = ( db.query(WeiboExternalOrder) .filter( WeiboExternalOrder.account_id == account.id, WeiboExternalOrder.archived_at.is_(None), WeiboExternalOrder.status != STATUS_VOID, ) .order_by(WeiboExternalOrder.id.asc()) .all() ) return { "ok": ok, "message": message, "account_id": account.id, "uid": account.uid, "nickname": account.nickname, "is_enabled": account.is_enabled, "ck_check_status": account.ck_check_status, "ck_check_message": account.ck_check_message, "is_bound": account.is_bound, "game_role_name": account.game_role_name, "weidou_balance": account.weidou_balance, "goods": goods, "orders_using_account": [ { "id": item.id, "order_no": item.order_no, "status": item.status, "status_label": ORDER_STATUS_LABELS.get(item.status, item.status), } for item in orders_using ], } def _metric( db: Session, current: User, order: WeiboExternalOrder, *, balance: int | None, required_total: int | None, surplus: int | None, records_total: int, records_matched: int, verdict: str, message: str, source: str, raw: Mapping[str, Any], ) -> WeiboExternalOrderMetric: metric = WeiboExternalOrderMetric( order_id=order.id, order_no=order.order_no, account_id=order.account_id, account_uid=order.account_uid, weidou_balance=balance, weidou_required_total=required_total, weidou_surplus=surplus, records_total=records_total, records_matched=records_matched, verdict=verdict, message=message[:512], source=source, raw=_json_safe(dict(raw)), created_by=current.id, ) db.add(metric) return metric def check_balance_groups( db: Session, current: User, orders: Sequence[WeiboExternalOrder], *, threshold: int, activity: str = "", ) -> dict[str, Any]: rows = list(orders) if not rows: raise CollabError("请先勾选订单", code="balance_check_empty") if len(rows) > CHECK_BATCH_MAX: raise CollabError(f"一次最多核账 {CHECK_BATCH_MAX} 条", code="batch_too_large") threshold = max(0, min(int(threshold), 1_000_000)) grouped: dict[int, list[WeiboExternalOrder]] = {} failures: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] for order in rows: if order.archived_at is not None: skipped.append( {"id": order.id, "order_no": order.order_no, "message": "已归档"} ) continue if not order.account_id: failures.append( {"id": order.id, "order_no": order.order_no, "message": "未关联账号"} ) continue if order.status not in BALANCE_GROUP_STATUSES: skipped.append( { "id": order.id, "order_no": order.order_no, "message": f"状态「{ORDER_STATUS_LABELS.get(order.status, order.status)}」不参与核账", } ) continue grouped.setdefault(int(order.account_id), []).append(order) selected_account_ids = set(grouped) grouped = {} for account_id in selected_account_ids: grouped[account_id] = ( db.query(WeiboExternalOrder) .filter( WeiboExternalOrder.account_id == account_id, WeiboExternalOrder.status.in_(BALANCE_GROUP_STATUSES), WeiboExternalOrder.archived_at.is_(None), ) .order_by(WeiboExternalOrder.id.asc()) .all() ) groups: list[dict[str, Any]] = [] updated = 0 for account_id, group_orders in grouped.items(): account = db.query(WeiboAccount).filter(WeiboAccount.id == account_id).first() if account is None: for order in group_orders: failures.append( { "id": order.id, "order_no": order.order_no, "message": "账号不存在", } ) continue if not account.is_enabled: for order in group_orders: failures.append( { "id": order.id, "order_no": order.order_no, "message": "账号已停用", } ) continue try: data = WeiboClient(account.cookie or "").list_mall(activity) except (WeiboError, ValueError) as exc: for order in group_orders: failures.append( {"id": order.id, "order_no": order.order_no, "message": str(exc)} ) continue balance = _to_int(data.get("weidou_balance")) required_total = sum(int(order.required_weidou or 0) for order in group_orders) surplus = balance - required_total if balance is not None else None if surplus is None: reason = "error" message = "未拿到微豆余额" target = None elif surplus < 0: reason = "shortage" message = ( f"微豆不足：余额 {balance}，订单合计 {required_total}，还差 {-surplus}" ) target = STATUS_WEIDOU_ERROR elif surplus > threshold: reason = "excess" message = ( f"微豆超出阈值：余额 {balance}，订单合计 {required_total}，" f"多出 {surplus}（阈值 {threshold}）" ) target = STATUS_WEIDOU_ERROR else: reason = "ok" message = ( f"核账通过：余额 {balance}，订单合计 {required_total}，结余 {surplus}" ) target = STATUS_RECHARGED if balance is not None: account.weidou_balance = balance now = _now() for order in group_orders: order.weidou_balance = balance order.weidou_required_total = required_total order.weidou_surplus = surplus order.weidou_threshold = threshold order.weidou_checked_at = now order.weidou_check_reason = reason order.weidou_check_message = message[:512] order.weidou_check_detail = { "account_id": account.id, "uid": account.uid, "balance": balance, "required_total": required_total, "surplus": surplus, "threshold": threshold, "order_ids": [item.id for item in group_orders], } if target is not None: set_status( db, current, order, target, reason=message, allow_check=True, commit=False, ) order.recharged_at = now updated += 1 _metric( db, current, order, balance=balance, required_total=required_total, surplus=surplus, records_total=0, records_matched=0, verdict=VERDICT_UNKNOWN if target is None else "", message=message, source="balance_check", raw={"reason": reason, "order_ids": [item.id for item in group_orders]}, ) _audit( db, current, order, "weibo:collab:balance_check", { "account_id": account.id, "balance": balance, "required_total": required_total, "surplus": surplus, "threshold": threshold, "reason": reason, }, ) groups.append( { "account_id": account.id, "uid": account.uid, "status": target or "", "status_label": ORDER_STATUS_LABELS.get(target or "", ""), "balance": balance, "required_total": required_total, "surplus": surplus, "threshold": threshold, "reason": reason, "message": message, "order_ids": [item.id for item in group_orders], } ) db.commit() return { "total": len(rows), "updated": updated, "threshold": threshold, "groups": groups, "failures": failures, "skipped": skipped, } def _record_key(record: Mapping[str, Any], index: int) -> str: raw = json.dumps( {"index": index, "record": _json_safe(dict(record))}, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(raw.encode()).hexdigest() def _record_matches(order: WeiboExternalOrder, record: Mapping[str, Any]) -> bool: if _text(record.get("game_account")).strip() != _text(order.game_role_name).strip(): return False if _text(record.get("goods_name")).strip() != _text(order.goods_name).strip(): return False if _to_int(record.get("cost_weidou")) != _to_int(order.required_weidou): return False if order.exchange_started_at is not None: started = order.exchange_started_at.astimezone( ZoneInfo("Asia/Shanghai") ).strftime("%Y-%m-%d %H:%M:%S") if _text(record.get("exchange_time")) < started: return False return True def check_exchange_records( db: Session, current: User, orders: Sequence[WeiboExternalOrder], *, activity: str = "", ) -> dict[str, Any]: rows = list(orders) if not rows: raise CollabError("请先勾选订单", code="exchange_check_empty") if len(rows) > CHECK_BATCH_MAX: raise CollabError(f"一次最多查询 {CHECK_BATCH_MAX} 条", code="batch_too_large") grouped: dict[int, list[WeiboExternalOrder]] = {} failures: list[dict[str, Any]] = [] for order in rows: if order.archived_at is not None or not order.account_id: failures.append( { "id": order.id, "order_no": order.order_no, "message": "订单已归档或未关联账号", } ) continue grouped.setdefault(int(order.account_id), []).append(order) results: list[dict[str, Any]] = [] success = failed = unknown = 0 for account_id, group_orders in grouped.items(): account = db.query(WeiboAccount).filter(WeiboAccount.id == account_id).first() if account is None or not account.is_enabled: for order in group_orders: failures.append( { "id": order.id, "order_no": order.order_no, "message": "账号不存在或已停用", } ) continue try: data = WeiboClient(account.cookie or "").list_exchange_records( activity, page=1, page_size=100, ) records = [ item for item in (data.get("list") or []) if isinstance(item, dict) ] except (WeiboError, ValueError) as exc: for order in group_orders: failures.append( {"id": order.id, "order_no": order.order_no, "message": str(exc)} ) continue used_keys = { str(item.exchange_record_key or "") for item in db.query(WeiboExternalOrder) .filter( WeiboExternalOrder.account_id == account_id, WeiboExternalOrder.exchange_record_key != "", WeiboExternalOrder.id.notin_([order.id for order in group_orders]), ) .all() if item.exchange_record_key } indexed = [ (_record_key(record, index), record) for index, record in enumerate(records) ] for order in group_orders: candidates = [ (key, record) for key, record in indexed if key not in used_keys and _record_matches(order, record) ] now = _now() order.exchange_checked_at = now if len(candidates) == 1: key, record = candidates[0] used_keys.add(key) order.exchange_record_key = key[:128] order.exchange_record_snapshot = _json_safe(record) order.exchange_verdict = VERDICT_SUCCESS order.exchange_message = "兑换记录唯一命中" if order.status == STATUS_EXCHANGE_PENDING: set_status( db, current, order, STATUS_DONE, reason="兑换记录唯一命中", allow_check=True, commit=False, ) verdict = VERDICT_SUCCESS message = "兑换记录唯一命中" success += 1 elif len(candidates) > 1: order.exchange_verdict = VERDICT_UNKNOWN order.exchange_message = ( f"找到 {len(candidates)} 条候选记录，需人工确认" ) verdict = VERDICT_UNKNOWN message = order.exchange_message unknown += 1 else: order.exchange_verdict = VERDICT_UNKNOWN order.exchange_message = "未找到匹配记录，需人工核对后再判定失败" verdict = VERDICT_UNKNOWN message = order.exchange_message unknown += 1 metric = _metric( db, current, order, balance=account.weidou_balance, required_total=order.required_weidou, surplus=None, records_total=len(records), records_matched=len(candidates), verdict=verdict, message=message, source="exchange_records", raw={ "candidates": [record for _, record in candidates[:5]], "role": order.game_role_name, "goods_name": order.goods_name, }, ) db.flush() _audit( db, current, order, "weibo:collab:exchange_check", { "verdict": verdict, "message": message, "records_total": len(records), "candidates": len(candidates), }, ) results.append( { "id": order.id, "order_no": order.order_no, "ok": True, "verdict": verdict, "verdict_label": VERDICT_LABELS.get(verdict, verdict), "message": message, "records_total": len(records), "records_matched": len(candidates), "status": order.status, "status_label": ORDER_STATUS_LABELS.get(order.status, order.status), "metric_id": metric.id, } ) db.commit() return { "total": len(rows), "success": success, "failed": failed, "unknown": unknown, "results": results, "failures": failures, } def archive_orders( db: Session, current: User, ids: Sequence[int], *, reason: str = "", ) -> dict[str, Any]: rows = list_orders_by_ids(db, ids) if not rows: raise CollabError("请先勾选订单", code="archive_empty") if len(rows) > ARCHIVE_BATCH_MAX: raise CollabError( f"一次最多归档 {ARCHIVE_BATCH_MAX} 条", code="batch_too_large" ) results: list[dict[str, Any]] = [] archived = 0 for order in rows: if order.archived_at is not None: results.append( { "id": order.id, "order_no": order.order_no, "ok": False, "message": "已归档", } ) continue if order.status not in ARCHIVABLE_STATUSES: results.append( { "id": order.id, "order_no": order.order_no, "ok": False, "message": "只有兑换成功的订单可以归档", } ) continue order.archived_at = _now() order.archived_by = current.id order.archive_reason = _text(reason, max_length=256) _touch(order, current) _audit( db, current, order, "weibo:collab:archive", {"reason": order.archive_reason}, ) archived += 1 results.append( {"id": order.id, "order_no": order.order_no, "ok": True, "message": ""} ) db.commit() return { "total": len(rows), "archived": archived, "skipped": len(rows) - archived, "results": results, } def unarchive_orders( db: Session, current: User, ids: Sequence[int], ) -> dict[str, Any]: rows = list_orders_by_ids(db, ids) if not rows: raise CollabError("请先勾选订单", code="archive_empty") results: list[dict[str, Any]] = [] unarchived = 0 for order in rows: if order.archived_at is None: results.append( { "id": order.id, "order_no": order.order_no, "ok": False, "message": "未归档", } ) continue previous = order.archive_reason order.archived_at = None order.archived_by = None order.archive_reason = "" _touch(order, current) _audit( db, current, order, "weibo:collab:unarchive", {"previous_reason": previous}, ) unarchived += 1 results.append( {"id": order.id, "order_no": order.order_no, "ok": True, "message": ""} ) db.commit() return { "total": len(rows), "unarchived": unarchived, "skipped": len(rows) - unarchived, "results": results, } def _account_clusters( db: Session, orders: Sequence[WeiboExternalOrder] ) -> dict[int, dict[str, Any]]: account_ids = {int(order.account_id) for order in orders if order.account_id} if not account_ids: return {} siblings = ( db.query(WeiboExternalOrder) .filter( WeiboExternalOrder.account_id.in_(account_ids), WeiboExternalOrder.archived_at.is_(None), WeiboExternalOrder.status != STATUS_VOID, ) .order_by(WeiboExternalOrder.id.asc()) .all() ) grouped: dict[int, list[WeiboExternalOrder]] = {} for row in siblings: if row.account_id: grouped.setdefault(int(row.account_id), []).append(row) result: dict[int, dict[str, Any]] = {} for order in orders: rows = grouped.get(int(order.account_id or 0), []) if len(rows) <= 1: continue result[int(order.id)] = { "total": len(rows), "orders": [ { "id": item.id, "order_no": item.order_no, "status": item.status, "status_label": ORDER_STATUS_LABELS.get(item.status, item.status), "goods_name": item.goods_name, } for item in rows if item.id != order.id ], } return result def _latest_metrics( db: Session, order_ids: Sequence[int] ) -> dict[int, WeiboExternalOrderMetric]: if not order_ids: return {} rows = ( db.query(WeiboExternalOrderMetric) .filter(WeiboExternalOrderMetric.order_id.in_(order_ids)) .order_by(WeiboExternalOrderMetric.id.desc()) .all() ) result: dict[int, WeiboExternalOrderMetric] = {} for row in rows: result.setdefault(int(row.order_id), row) return result def serialize_order( order: WeiboExternalOrder, *, users: Mapping[int, str] | None = None, accounts: Mapping[int, WeiboAccount] | None = None, cluster: Mapping[str, Any] | None = None, last_metric: WeiboExternalOrderMetric | None = None, ) -> dict[str, Any]: names = users or {} account_map = accounts or {} account = account_map.get(int(order.account_id)) if order.account_id else None return { "id": order.id, "order_no": order.order_no, "shop": order.shop or "", "owner_id": order.owner_id, "owner_name": names.get(int(order.owner_id), "") if order.owner_id else "", "account_id": order.account_id, "account_uid": order.account_uid or "", "account_nickname": order.account_nickname or "", "account_exists": account is not None, "account_enabled": bool(account and account.is_enabled), "account_ck_status": account.ck_check_status if account else "", "account_ck_message": account.ck_check_message if account else "", "account_is_bound": account.is_bound if account else None, "game_role_name": order.game_role_name or "", "bind_snapshot_at": order.bind_snapshot_at, "goods_id": order.goods_id, "goods_name": order.goods_name or "", "goods_image": order.goods_image or "", "required_weidou": order.required_weidou, "home_images": list(order.home_images or []), "remark": order.remark or "", "status": order.status, "status_label": ORDER_STATUS_LABELS.get(order.status, order.status), "next_status": NEXT_STATUS.get(order.status), "missing_fields": _missing_for_recharge(order), "weidou_balance": order.weidou_balance, "weidou_required_total": order.weidou_required_total, "weidou_surplus": order.weidou_surplus, "weidou_threshold": order.weidou_threshold, "weidou_checked_at": order.weidou_checked_at, "weidou_check_reason": order.weidou_check_reason or "", "weidou_check_message": order.weidou_check_message or "", "weidou_check_detail": order.weidou_check_detail, "exchange_started_at": order.exchange_started_at, "exchange_checked_at": order.exchange_checked_at, "exchange_record_key": order.exchange_record_key or "", "exchange_record_snapshot": order.exchange_record_snapshot, "exchange_verdict": order.exchange_verdict or "", "exchange_message": order.exchange_message or "", "recharged_at": order.recharged_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": names.get(int(order.archived_by), "") if order.archived_by else "", "archive_reason": order.archive_reason or "", "created_by": order.created_by, "created_by_name": names.get(int(order.created_by), "") if order.created_by else "", "updated_by": order.updated_by, "updated_by_name": names.get(int(order.updated_by), "") if order.updated_by else "", "version": int(order.version or 0), "created_at": order.created_at, "updated_at": order.updated_at, "account_cluster": dict(cluster) if cluster else None, "last_metric_at": last_metric.created_at if last_metric else None, "last_metric_message": last_metric.message if last_metric else "", } def _base_query(db: Session): return db.query(WeiboExternalOrder) def _apply_filters( query, *, archive_scope: str = "active", status: str = "", shop: str = "", owner_id: int | None = None, goods_name: str = "", keyword: str = "", include_void: bool = False, ): scope = archive_scope if archive_scope in ARCHIVE_SCOPES else "active" if scope == "active": query = query.filter(WeiboExternalOrder.archived_at.is_(None)) elif scope == "archived": query = query.filter(WeiboExternalOrder.archived_at.isnot(None)) if not include_void: query = query.filter(WeiboExternalOrder.status != STATUS_VOID) if _text(status): query = query.filter(WeiboExternalOrder.status == status) if _text(shop): query = query.filter(WeiboExternalOrder.shop == shop) if owner_id is not None: query = query.filter(WeiboExternalOrder.owner_id == int(owner_id)) if _text(goods_name): query = query.filter(WeiboExternalOrder.goods_name == goods_name) search = _text(keyword) if search: pattern = f"%{search}%" query = query.filter( or_( WeiboExternalOrder.order_no.ilike(pattern), WeiboExternalOrder.account_uid.ilike(pattern), WeiboExternalOrder.account_nickname.ilike(pattern), WeiboExternalOrder.game_role_name.ilike(pattern), WeiboExternalOrder.shop.ilike(pattern), WeiboExternalOrder.goods_name.ilike(pattern), WeiboExternalOrder.remark.ilike(pattern), ) ) return query def list_orders( db: Session, *, page: int, page_size: int, archive_scope: str = "active", status: str = "", shop: str = "", owner_id: int | None = None, goods_name: str = "", keyword: str = "", include_void: bool = False, ) -> dict[str, Any]: query = _apply_filters( _base_query(db), archive_scope=archive_scope, status=status, shop=shop, owner_id=owner_id, goods_name=goods_name, keyword=keyword, include_void=include_void, ) total = query.count() rows = ( query.order_by(WeiboExternalOrder.id.desc()) .offset((max(1, page) - 1) * page_size) .limit(page_size) .all() ) users = _users_by_id( db, { int(item) for order in rows for item in ( order.owner_id, order.created_by, order.updated_by, order.archived_by, ) if item }, ) accounts = _accounts_by_id( db, {int(order.account_id) for order in rows if order.account_id} ) clusters = _account_clusters(db, rows) metrics = _latest_metrics(db, [order.id for order in rows]) return { "items": [ serialize_order( order, users=users, accounts=accounts, cluster=clusters.get(order.id), last_metric=metrics.get(order.id), ) for order in rows ], "total": total, "page": page, "page_size": page_size, } def serialize_one(db: Session, order: WeiboExternalOrder) -> dict[str, Any]: users = _users_by_id( db, { int(item) for item in ( order.owner_id, order.created_by, order.updated_by, order.archived_by, ) if item }, ) accounts = _accounts_by_id( db, {int(order.account_id)} if order.account_id else set() ) cluster = _account_clusters(db, [order]).get(order.id) metric = _latest_metrics(db, [order.id]).get(order.id) return serialize_order( order, users=users, accounts=accounts, cluster=cluster, last_metric=metric, ) def summary( db: Session, *, include_void: bool = False, archive_scope: str = "active" ) -> dict[str, Any]: query = _apply_filters( _base_query(db), archive_scope=archive_scope, include_void=include_void, ) by_status = { str(status): int(count) for status, count in query.with_entities( WeiboExternalOrder.status, func.count(WeiboExternalOrder.id) ) .group_by(WeiboExternalOrder.status) .all() } account_disabled = ( query.join(WeiboAccount, WeiboAccount.id == WeiboExternalOrder.account_id) .filter(WeiboAccount.is_enabled.is_(False)) .with_entities(func.count(WeiboExternalOrder.id)) .scalar() or 0 ) ck_invalid = ( query.join(WeiboAccount, WeiboAccount.id == WeiboExternalOrder.account_id) .filter(WeiboAccount.ck_check_status.in_(("invalid", "error"))) .with_entities(func.count(WeiboExternalOrder.id)) .scalar() or 0 ) return { "total": sum(by_status.values()), "by_status": by_status, "account_missing": ( db.query(func.count(WeiboExternalOrder.id)) .filter( WeiboExternalOrder.account_id.is_(None), WeiboExternalOrder.archived_at.is_(None), ) .scalar() or 0 ), "account_disabled": int(account_disabled), "ck_invalid": int(ck_invalid), "archived_total": ( db.query(func.count(WeiboExternalOrder.id)) .filter(WeiboExternalOrder.archived_at.isnot(None)) .scalar() or 0 ), } def serialize_metric(metric: WeiboExternalOrderMetric) -> dict[str, Any]: return { "id": metric.id, "order_id": metric.order_id, "order_no": metric.order_no, "account_uid": metric.account_uid or "", "weidou_balance": metric.weidou_balance, "weidou_required_total": metric.weidou_required_total, "weidou_surplus": metric.weidou_surplus, "records_total": metric.records_total, "records_matched": metric.records_matched, "verdict": metric.verdict or "", "verdict_label": VERDICT_LABELS.get(metric.verdict, metric.verdict or ""), "message": metric.message or "", "source": metric.source or "", "created_at": metric.created_at, } def order_metrics(db: Session, order_id: int) -> list[dict[str, Any]]: rows = ( db.query(WeiboExternalOrderMetric) .filter(WeiboExternalOrderMetric.order_id == order_id) .order_by(WeiboExternalOrderMetric.id.desc()) .all() ) return [serialize_metric(row) for row in rows] def order_timeline( db: Session, order_id: int, *, limit: int = 100 ) -> list[dict[str, Any]]: rows = ( db.query(AuditLog) .filter(AuditLog.target == f"weibo_external_order:{order_id}") .order_by(AuditLog.id.desc()) .limit(limit) .all() ) return [ { "id": row.id, "action": row.action, "username": row.username, "detail": row.detail, "success": row.success, "created_at": row.created_at, } for row in rows ] def list_shops(db: Session, *, active_only: bool = True) -> list[WeiboCollabShop]: query = db.query(WeiboCollabShop) if active_only: query = query.filter(WeiboCollabShop.is_active.is_(True)) return query.order_by(WeiboCollabShop.sort_order, WeiboCollabShop.id).all() def ensure_shops_registered(db: Session, current: User, names: Sequence[str]) -> None: existing = {item.name for item in db.query(WeiboCollabShop.name).all() if item.name} for raw in names: name = _text(raw, max_length=64) if name and name not in existing: db.add( WeiboCollabShop(name=name, created_by=current.id, updated_by=current.id) ) existing.add(name) def create_shop( db: Session, current: User, *, name: str, remark: str = "" ) -> WeiboCollabShop: clean = _text(name, max_length=64) if not clean: raise CollabError("店铺名不能为空", code="shop_name_required") if db.query(WeiboCollabShop).filter(WeiboCollabShop.name == clean).first(): raise CollabError("店铺已存在", status_code=409, code="shop_exists") shop = WeiboCollabShop( name=clean, remark=_text(remark, max_length=255), created_by=current.id, updated_by=current.id, ) db.add(shop) db.commit() db.refresh(shop) return shop def update_shop( db: Session, current: User, shop: WeiboCollabShop, payload: Mapping[str, Any] ) -> None: old_name = shop.name for field in ("name", "sort_order", "is_active", "remark"): if field not in payload: continue value = payload[field] if field == "sort_order": value = _to_int(value) or 0 elif field == "is_active": value = bool(value) elif field == "remark": value = _text(value, max_length=255) else: value = _text(value, max_length=64) setattr(shop, field, value) if shop.name != old_name: db.query(WeiboExternalOrder).filter(WeiboExternalOrder.shop == old_name).update( {"shop": shop.name} ) shop.updated_by = current.id db.commit() def delete_shop(db: Session, shop: WeiboCollabShop) -> None: if ( db.query(WeiboExternalOrder) .filter(WeiboExternalOrder.shop == shop.name) .first() ): raise CollabError("仍有订单使用该店铺，请改为停用", code="shop_in_use") db.delete(shop) db.commit() def list_goods(db: Session, *, active_only: bool = True) -> list[WeiboCollabGoods]: query = db.query(WeiboCollabGoods) if active_only: query = query.filter(WeiboCollabGoods.is_active.is_(True)) return query.order_by(WeiboCollabGoods.sort_order, WeiboCollabGoods.id).all() def create_goods( db: Session, current: User, *, name: str, price: int, goods_id: int | None = None, goods_image: str = "", remark: str = "", ) -> WeiboCollabGoods: clean = _text(name, max_length=128) if not clean: raise CollabError("商品名不能为空", code="goods_name_required") if db.query(WeiboCollabGoods).filter(WeiboCollabGoods.name == clean).first(): raise CollabError("商品已存在", status_code=409, code="goods_exists") goods = WeiboCollabGoods( name=clean, price=max(0, int(price)), goods_id=goods_id, goods_image=_text(goods_image, max_length=512), remark=_text(remark, max_length=255), created_by=current.id, updated_by=current.id, ) db.add(goods) db.commit() db.refresh(goods) return goods def update_goods( db: Session, current: User, goods: WeiboCollabGoods, payload: Mapping[str, Any], ) -> None: old_name = goods.name for field in ( "name", "price", "goods_id", "goods_image", "sort_order", "is_active", "remark", ): if field not in payload: continue value = payload[field] if field in ("price", "goods_id", "sort_order"): value = _to_int(value) elif field == "is_active": value = bool(value) elif field == "goods_image": value = _text(value, max_length=512) elif field == "remark": value = _text(value, max_length=255) else: value = _text(value, max_length=128) setattr(goods, field, value) if goods.name != old_name: db.query(WeiboExternalOrder).filter( WeiboExternalOrder.goods_name == old_name ).update({"goods_name": goods.name}) goods.updated_by = current.id db.commit() def delete_goods(db: Session, goods: WeiboCollabGoods) -> None: if ( db.query(WeiboExternalOrder) .filter(WeiboExternalOrder.goods_name == goods.name) .first() ): raise CollabError("仍有订单使用该商品，请改为停用", code="goods_in_use") db.delete(goods) db.commit()