fix(huya): 修复外部单积分复核与抢购结果误判

web/backend/schemas.py

@@ -1963,6 +1963,8 @@ class HuyaExternalOrderRechargeOut(BaseModel):
     updated: int
     failed: int
     threshold: int = 100
+    confirmed_groups: int = 0
+    confirmation_delay_seconds: float = 0
     groups: list[HuyaExternalOrderRechargeGroupOut] = Field(default_factory=list)
     failures: list[HuyaExternalOrderRechargeFailureOut] = Field(default_factory=list)
 

web/backend/services/huya_collab_service.py

@@ -23,7 +23,7 @@ import hashlib
 import re
 import time
 from collections.abc import Mapping, Sequence
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
 from loguru import logger
@@ -142,6 +142,9 @@ ARCHIVE_BATCH_MAX = 5000
 # 充值核账阈值：账号积分 - 礼包积分合计 的允许结余上限（默认 100）。
 RECHARGE_POINTS_THRESHOLD_DEFAULT = 100
 RECHARGE_POINTS_THRESHOLD_MAX = 1_000_000
+# 积分到账/接口缓存可能晚于充值动作。首次算出异常后，统一等待一次再复核，
+# 避免把瞬时 0 分直接落成 points_error；同批只等待一次，不按账号组累加。
+RECHARGE_POINTS_CONFIRM_DELAY_SECONDS = 30
 
 
 def normalize_points_threshold(value: Any) -> int:
@@ -926,6 +929,9 @@ def _check_recharge_group(
     *,
     threshold: int,
     client: Any | None = None,
+    confirm_points_error: bool = True,
+    confirmation: Mapping[str, Any] | None = None,
+    score_source: str = "recharge_check",
 ) -> dict:
     group_label = _recharge_group_label(group_key)
     group_orders = _recharge_group_orders(db, group_key)
@@ -998,7 +1004,7 @@ def _check_recharge_group(
             order,
             scope=ACTIVITY_SCOPE_SCORE,
             client=client,
-            source="recharge_check",
+            source=score_source,
             commit=False,
         )
         score = result.get("score")
@@ -1031,6 +1037,21 @@ def _check_recharge_group(
             f"结余 {surplus} 分（阈值 {threshold}）"
         )
 
+    if result_status == STATUS_POINTS_ERROR and not confirm_points_error:
+        return {
+            "_needs_confirmation": True,
+            "group_key": group_label,
+            "bind_game_account": group_key[0],
+            "bind_game_role": group_key[1],
+            "threshold": threshold,
+            "balance": balance,
+            "required_total": required_total,
+            "surplus": surplus,
+            "reason": reason,
+            "message": message,
+            "order_ids": [int(order.id) for order in group_orders],
+        }
+
     now = _now()
     detail = {
         "group_key": group_label,
@@ -1063,6 +1084,8 @@ def _check_recharge_group(
             for order in group_orders
         ],
     }
+    if confirmation:
+        detail["confirmation"] = dict(confirmation)
 
     for order in group_orders:
         set_status(
@@ -1125,6 +1148,28 @@ def _check_recharge_group(
     }
 
 
+def _failed_recharge_group_result(
+    group_key: RechargeGroupKey,
+    group_orders: Sequence[HuyaExternalOrder],
+    *,
+    threshold: int,
+    message: str,
+) -> dict:
+    return {
+        "group_key": _recharge_group_label(group_key),
+        "status": "",
+        "status_label": "",
+        "balance": None,
+        "required_total": None,
+        "surplus": None,
+        "threshold": threshold,
+        "reason": "error",
+        "message": message,
+        "order_ids": [int(order.id) for order in group_orders],
+        "orders": [],
+    }
+
+
 def check_recharge_groups(
     db: Session,
     current: User | None,
@@ -1132,6 +1177,7 @@ def check_recharge_groups(
     *,
     threshold: Any = RECHARGE_POINTS_THRESHOLD_DEFAULT,
     client: Any | None = None,
+    confirm_delay_seconds: float = RECHARGE_POINTS_CONFIRM_DELAY_SECONDS,
     commit: bool = True,
 ) -> dict:
     """按「``bind_game_account`` + ``bind_game_role``」整组核账，并整组落到「已充值 / 积分异常」。
@@ -1176,9 +1222,9 @@ def check_recharge_groups(
         if key not in group_keys:
             group_keys.append(key)
 
-    groups: list[dict] = []
+    group_results: dict[RechargeGroupKey, dict] = {}
+    pending_confirmation: dict[RechargeGroupKey, dict] = {}
     for key in group_keys:
-        group_label = _recharge_group_label(key)
         try:
             with db.begin_nested():
                 group = _check_recharge_group(
@@ -1187,46 +1233,84 @@ def check_recharge_groups(
                     key,
                     threshold=threshold_value,
                     client=client,
+                    confirm_points_error=False,
                 )
-            groups.append(group)
+            if group.get("_needs_confirmation"):
+                pending_confirmation[key] = group
+            else:
+                group_results[key] = group
         except CollabError as exc:
             group_rows = _recharge_group_orders(db, key)
             failed_order_count += max(1, len(group_rows))
-            groups.append(
-                {
-                    "group_key": group_label,
-                    "status": "",
-                    "status_label": "",
-                    "balance": None,
-                    "required_total": None,
-                    "surplus": None,
-                    "threshold": threshold_value,
-                    "reason": "error",
-                    "message": exc.detail,
-                    "order_ids": [int(order.id) for order in group_rows],
-                    "orders": [],
-                }
+            group_results[key] = _failed_recharge_group_result(
+                key,
+                group_rows,
+                threshold=threshold_value,
+                message=exc.detail,
             )
         except Exception as exc:  # noqa: BLE001 单个账号组失败不影响其它组
             logger.debug(f"[huya-collab] 充值核账单组失败：{type(exc).__name__}: {exc}")
             group_rows = _recharge_group_orders(db, key)
             failed_order_count += max(1, len(group_rows))
-            groups.append(
-                {
-                    "group_key": group_label,
-                    "status": "",
-                    "status_label": "",
-                    "balance": None,
-                    "required_total": None,
-                    "surplus": None,
-                    "threshold": threshold_value,
-                    "reason": "error",
-                    "message": f"{type(exc).__name__}: {exc}",
-                    "order_ids": [int(order.id) for order in group_rows],
-                    "orders": [],
-                }
+            group_results[key] = _failed_recharge_group_result(
+                key,
+                group_rows,
+                threshold=threshold_value,
+                message=f"{type(exc).__name__}: {exc}",
             )
 
+    delay = max(0.0, float(confirm_delay_seconds or 0))
+    if pending_confirmation and delay:
+        # 已通过的组先落库，避免等待二次确认时把它们的行锁持有整整 30 秒。
+        if commit:
+            db.commit()
+        time.sleep(delay)
+
+    for key, first in pending_confirmation.items():
+        try:
+            with db.begin_nested():
+                group = _check_recharge_group(
+                    db,
+                    current,
+                    key,
+                    threshold=threshold_value,
+                    client=client,
+                    confirm_points_error=True,
+                    confirmation={
+                        "attempts": 2,
+                        "delay_seconds": delay,
+                        "first_balance": first.get("balance"),
+                        "first_required_total": first.get("required_total"),
+                        "first_surplus": first.get("surplus"),
+                        "first_reason": first.get("reason"),
+                    },
+                    score_source="recharge_retry",
+                )
+            group_results[key] = group
+        except CollabError as exc:
+            group_rows = _recharge_group_orders(db, key)
+            failed_order_count += max(1, len(group_rows))
+            group_results[key] = _failed_recharge_group_result(
+                key,
+                group_rows,
+                threshold=threshold_value,
+                message=exc.detail,
+            )
+        except Exception as exc:  # noqa: BLE001 单个账号组二次确认失败不影响其它组
+            logger.debug(
+                f"[huya-collab] 充值核账二次确认失败：{type(exc).__name__}: {exc}"
+            )
+            group_rows = _recharge_group_orders(db, key)
+            failed_order_count += max(1, len(group_rows))
+            group_results[key] = _failed_recharge_group_result(
+                key,
+                group_rows,
+                threshold=threshold_value,
+                message=f"{type(exc).__name__}: {exc}",
+            )
+
+    groups = [group_results[key] for key in group_keys if key in group_results]
+
     updated = sum(
         len(group.get("order_ids") or []) for group in groups if group.get("status")
     )
@@ -1239,6 +1323,8 @@ def check_recharge_groups(
         "updated": updated,
         "failed": failed_order_count,
         "threshold": threshold_value,
+        "confirmed_groups": len(pending_confirmation),
+        "confirmation_delay_seconds": delay,
         "groups": groups,
         "failures": failures,
     }
@@ -2009,6 +2095,10 @@ ACTIVITY_SCOPE_LABELS = {
     ACTIVITY_SCOPE_RECORDS: "兑换记录",
 }
 
+# 官方兑换记录存在落库延迟：生产实测三单在查询后 20～54 秒才出现记录。
+# 因此脚本开始后的观察窗口内，「没查到」只能算暂时无法判定，不能自动判失败。
+ACTIVITY_RESULT_SETTLE_SECONDS = 10 * 60
+
 # metric.raw 里最多留多少条兑换记录（避免 JSON 列被几百条记录撑爆）
 _METRIC_RECORD_LIMIT = 50
 # 兑换记录里跟订单礼包名比对时，只取有意义的长度（太短的名字容易误匹配）
@@ -2021,26 +2111,57 @@ def _metric_message(text: Any, limit: int = 480) -> str:
 
 
 def gift_matches_records(
-    records: Sequence[Mapping[str, Any]], gift_name: str
+    records: Sequence[Mapping[str, Any]],
+    gift_name: str,
+    *,
+    not_before: datetime | None = None,
 ) -> list[dict[str, Any]]:
     """在兑换记录里找这一单的礼包（名字完全相等或互相包含都算）。
 
     虎牙记录里的 ``prize_name`` 与活动列表里的名字基本一字不差，但偶尔带后缀/前缀，
     所以用「互相包含」宽松匹配；太短的名字（<2 字）不参与，免得 "礼包" 两个字匹配一堆。
+    传入 ``not_before`` 时还会排除上脚本之前的旧记录，避免同一买家重复购买同款时，
+    把上次兑换误判成这次成功。
     """
     wanted = str(gift_name or "").strip()
     if len(wanted) < _MIN_GIFT_MATCH_LENGTH:
         return []
+    threshold = not_before
+    if threshold is not None and threshold.tzinfo is None:
+        threshold = threshold.replace(tzinfo=UTC)
+    if threshold is not None:
+        # exchange_time 只有秒精度，上脚本时间可能带微秒；同秒记录不能误排掉。
+        threshold = threshold.replace(microsecond=0)
+
     matched: list[dict[str, Any]] = []
     for record in records:
         name = str(record.get("prize_name") or "").strip()
         if not name:
             continue
-        if name == wanted or wanted in name or name in wanted:
-            matched.append(dict(record))
+        if not (name == wanted or wanted in name or name in wanted):
+            continue
+        raw_exchange_time = record.get("exchange_time")
+        if threshold is not None and raw_exchange_time not in (None, ""):
+            try:
+                exchange_at = datetime.fromtimestamp(float(raw_exchange_time), tz=UTC)
+            except (TypeError, ValueError, OSError):
+                exchange_at = None
+            if exchange_at is not None and exchange_at < threshold:
+                continue
+        matched.append(dict(record))
     return matched
 
 
+def _activity_result_settle_at(order: HuyaExternalOrder) -> datetime | None:
+    """返回脚本后的最早自动判失败时间；没有上脚本时间时不自动判失败。"""
+    scripted_at = order.scripted_at
+    if scripted_at is None:
+        return None
+    if scripted_at.tzinfo is None:
+        scripted_at = scripted_at.replace(tzinfo=UTC)
+    return scripted_at + timedelta(seconds=ACTIVITY_RESULT_SETTLE_SECONDS)
+
+
 def _extract_score(resp: Any) -> tuple[int | None, int | None, int | None, str]:
     """从 ``getUserScore`` 响应里取「可用 / 累计获得 / 已消耗」+ 可读结论。"""
     if resp is None:
@@ -2294,7 +2415,13 @@ def query_order_activity(
         _extract_score(score_resp) if want_score else (None, None, None, "")
     )
     records, records_msg = _extract_records(records_resp) if want_records else ([], "")
-    matched = gift_matches_records(records, order.gift_name)
+    matched = gift_matches_records(
+        records,
+        order.gift_name,
+        not_before=order.scripted_at,
+    )
+    now = _now()
+    settle_at = _activity_result_settle_at(order)
 
     if not want_records:
         # 只查积分：兑换记录一个字节都没查 —— 不许按"记录是空的"判成「兑换失败」，
@@ -2324,6 +2451,19 @@ def query_order_activity(
             f"（最近一条：{latest.get('status_label') or '状态未知'}"
             f"{('，单号 ' + str(latest.get('order_id'))) if latest.get('order_id') else ''}）"
         )
+    elif order.status == STATUS_SCRIPTED and (settle_at is None or now < settle_at):
+        verdict = VERDICT_UNKNOWN
+        if settle_at is None:
+            message = (
+                f"兑换记录里暂未查到「{order.gift_name}」；"
+                "这单没有上脚本时间，不能自动判定抢购失败"
+            )
+        else:
+            message = (
+                f"兑换记录里暂未查到「{order.gift_name}」"
+                f"（当前共 {len(records)} 条记录）；处于上脚本后的 10 分钟抢购确认窗口，"
+                "暂不自动判失败"
+            )
     else:
         verdict = VERDICT_FAILED
         message = (
@@ -2338,13 +2478,26 @@ def query_order_activity(
         target = STATUS_FAILED
     else:
         target = ""
-    if target and order.status == STATUS_SCRIPTED:
-        set_status(db, current, order, target, commit=False)
+    can_promote_success = target == STATUS_DONE and order.status in (
+        STATUS_SCRIPTED,
+        STATUS_FAILED,
+    )
+    can_promote_failure = target == STATUS_FAILED and order.status == STATUS_SCRIPTED
+    if can_promote_success or can_promote_failure:
+        set_status(
+            db,
+            current,
+            order,
+            target,
+            rollback=order.status == STATUS_FAILED,
+            commit=False,
+        )
         auto_status = target
     elif target:
         message = (
             f"{message}（当前状态是「{ORDER_STATUS_LABELS.get(order.status, order.status)}」，"
-            f"只有「{ORDER_STATUS_LABELS[STATUS_SCRIPTED]}」的单才会按查询结果自动改状态）"
+            "只有「已上脚本」或「抢购失败」的单才会按抢购成功结果自动纠正；"
+            "自动判失败只作用于「已上脚本」且已过确认窗口的单）"
         )
 
     raw: dict[str, Any] = {
@@ -2355,6 +2508,7 @@ def query_order_activity(
         "score_used": used,
         "records_count": len(records),
         "matched": matched[:5],
+        "result_settle_at": settle_at.isoformat() if settle_at else None,
         "records": records[:_METRIC_RECORD_LIMIT],
     }
     if score_resp is not None: