fix(ingest): 审查修复 - 3 CRITICAL + 8 WARNING (#42)

Reviewer 阶段静态审查(2026-09-10)发现 3 CRITICAL + 8 WARNING, 本轮全部修复。

CRITICAL:
1. 对外响应契约字段丢失: ingest_complaints 白名单漏 warnings/errors →
   workbuddy 外部工具拿不到 external.warnings/errors。已补齐白名单(8 字段)。
2. 校验告警污染 errors: SyncRowResult.add_warning 会把 warning 追加进 self.errors,
   导致 failed=0 时 errors 非空、并污染 complaint_sync_logs.error。
   → 新增独立 validation 通道(add_validation), 与 errors/failed 严格分离。
3. rejected 与 failed 语义混同: res.rejected = res.failed 使统一返回值失真。
   → _persist_rejected_from_errors 返回真实落库条数; ImportResult 新增 rejected
     字段; ingest_* 分别取真实拒收计数(无则 0, 不伪装)。

WARNING:
- 8 位日期串 YYYYMMDD(20260614)会被判 None → 新增 _try_compact_date 兼容
- 区间外纯数字判 None 时补 logger.warning(可见化, 不再静默)
- numpy 标量(np.int64)未纳入类型判断 → 会复现纳秒 bug; 统用 .item() 归一
- 1e-6 天容差过宽(≈86ms)会误伤合法小数时刻 → 收紧到 1e-9
- 校验器单行异常静默吞掉 → 计入 R-校验器 告警条目; 删死代码 _TICKET_STATUS_SET
- _MIN_VALID_YEAR 2015 无数据支撑 → 调为 2020(与回填脚本 _BAD_BEFORE 口径一致)
- 回填脚本容差匹配可能张冠李戴 → 改"唯一候选才回填"(多候选记 ambiguous);
  print 全改 logger; 加 --max-id(去魔法数)、--apply 需 --yes 二次确认
- 迁移非幂等 → 按 alembic 规范加列/索引存在性检查

测试: 新增/更新边界用例(8位日期串/numpy标量/Unix时间戳/小数时刻/契约字段回归锁/
rejected 解耦), 三个测试文件 76 例全通过; 全量 pytest 通过; ruff 通过。
端到端: 源表时间字段解析失败 0。

关联: #42

backend/alembic/versions/d8e9f0a1b2c3_add_quality_fields_to_complaint_tables.py

@@ -34,40 +34,69 @@ depends_on: Union[str, Sequence[str], None] = None
 _TABLES = ('external_complaints', 'card_complaints')
 
 
+def _column_exists(table_name: str, column_name: str) -> bool:
+    """列存在性检查(幂等升级规范, 见 alembic-migration管理规范.md §五)
+
+    已有表环境可能由 Base.metadata.create_all() 建表(含新列), 直接 add_column 会报
+    DuplicateColumn → 必须先查 information_schema。
+    """
+    conn = op.get_bind()
+    return bool(conn.execute(sa.text(
+        "SELECT EXISTS(SELECT 1 FROM information_schema.columns "
+        "WHERE table_name = :t AND column_name = :c)"
+    ), {"t": table_name, "c": column_name}).scalar())
+
+
+def _index_exists(index_name: str) -> bool:
+    """索引存在性检查(幂等)"""
+    conn = op.get_bind()
+    return bool(conn.execute(sa.text(
+        "SELECT EXISTS(SELECT 1 FROM pg_indexes WHERE indexname = :i)"
+    ), {"i": index_name}).scalar())
+
+
 def upgrade() -> None:
-    """投诉两表补 quality_status / quality_note"""
+    """投诉两表补 quality_status / quality_note(幂等: 已存在则跳过)"""
     for table in _TABLES:
-        op.add_column(
-            table,
-            sa.Column(
-                'quality_status',
-                sa.String(length=16),
-                nullable=True,
-                comment=(
-                    '数据质量状态: None=正常 / ok=已人工确认 / auto_fixed=已自动纠错 / '
-                    'pending=待人工确认 / dup_residual=已治理重复(排除) / ignored=人工彻底忽略(排除)'
+        if not _column_exists(table, 'quality_status'):
+            op.add_column(
+                table,
+                sa.Column(
+                    'quality_status',
+                    sa.String(length=16),
+                    nullable=True,
+                    comment=(
+                        '数据质量状态: None=正常 / ok=已人工确认 / auto_fixed=已自动纠错 / '
+                        'pending=待人工确认 / dup_residual=已治理重复(排除) / ignored=人工彻底忽略(排除)'
+                    ),
                 ),
-            ),
-        )
-        op.add_column(
-            table,
-            sa.Column(
-                'quality_note',
-                sa.String(length=512),
-                nullable=True,
-                comment='质量说明(自动纠错或待确认原因, 含原始值)',
-            ),
-        )
-        op.create_index(
-            f'ix_{table}_quality_status',
-            table,
-            ['quality_status'],
-        )
+            )
+        if not _column_exists(table, 'quality_note'):
+            op.add_column(
+                table,
+                sa.Column(
+                    'quality_note',
+                    sa.String(length=512),
+                    nullable=True,
+                    comment='质量说明(自动纠错或待确认原因, 含原始值)',
+                ),
+            )
+        idx = f'ix_{table}_quality_status'
+        if not _index_exists(idx):
+            op.create_index(idx, table, ['quality_status'])
 
 
 def downgrade() -> None:
-    """回退: 删索引与两列"""
+    """回退: 删索引与两列(幂等)
+
+    ⚠️ 降级会物理删除两列 → 已写入的治理状态(pending/ok/auto_fixed 等)一并丢失且不可恢复。
+    如需降级, 请先自行导出非空 quality_status/quality_note 作为备份。
+    """
     for table in _TABLES:
-        op.drop_index(f'ix_{table}_quality_status', table_name=table)
-        op.drop_column(table, 'quality_note')
-        op.drop_column(table, 'quality_status')
+        idx = f'ix_{table}_quality_status'
+        if _index_exists(idx):
+            op.drop_index(idx, table_name=table)
+        if _column_exists(table, 'quality_note'):
+            op.drop_column(table, 'quality_note')
+        if _column_exists(table, 'quality_status'):
+            op.drop_column(table, 'quality_status')

backend/app/services/complaint_sync.py

@@ -98,6 +98,15 @@ class SyncRowResult:
         self.warnings = 0
         self.matched = 0
         self.errors: list[dict[str, Any]] = []
+        # #42: 入库前校验问题清单(只读 advisory)。
+        # ⚠️ 必须与 errors 严格分离: errors=硬失败明细(对外契约 failed/errors 语义),
+        #    validation=校验告警(不丢弃行); 混入 errors 会让 failed=0 时 errors 非空,
+        #    破坏 workbuddy 对外契约与 complaint_sync_logs.error 语义。
+        self.validation: list[dict[str, Any]] = []
+
+    def add_validation(self, row_idx: int, issue: dict) -> None:
+        """记录一条入库前校验问题(不影响 errors/failed 计数)"""
+        self.validation.append({"row": row_idx, **issue})
 
     def add_warning(self, row_idx: int, msg: str, data: dict | None = None) -> None:
         """软警告: 行保留但字段被清洗"""
@@ -452,7 +461,7 @@ def sync_complaints_from_payload(
                 continue
             # #42 入库前校验(只读): 时间/枚举/必填问题可见化, 不丢弃不改数据
             for _iss in validate_external_complaint_row(rec, idx):
-                ext_result.add_warning(idx, f"[校验]{_iss['rule']} {_iss['message']}")
+                ext_result.add_validation(idx, _iss)
             ext_rows.append(rec)
         ext_result.total = len(ext_rows_raw)
 
@@ -480,7 +489,7 @@ def sync_complaints_from_payload(
                 continue
             # #42 入库前校验(只读)
             for _iss in validate_card_complaint_row(rec, idx):
-                card_result.add_warning(idx, f"[校验]{_iss['rule']} {_iss['message']}")
+                card_result.add_validation(idx, _iss)
             card_rows.append(rec)
         card_result.total = len(card_rows_raw)
 
@@ -516,10 +525,14 @@ def sync_complaints_from_payload(
         if ext_result.filtered or card_result.filtered:
             _info = f"过滤全空行: 315={ext_result.filtered} 卡商={card_result.filtered}"
             log.error = (f"{_info} | " + str(all_errors)) if all_errors else _info
-        # #42 校验问题计数(只读, 可见化): 汇总写入日志供质量排查
-        _val_warn = ext_result.warnings + card_result.warnings
-        if _val_warn:
-            _val_info = f"入库前校验告警: 315={ext_result.warnings} 卡商={card_result.warnings}"
+        # #42 校验问题计数(只读, 可见化): 独立通道汇总写入日志供质量排查。
+        # ⚠️ 不与 errors 合并: errors 是对外契约的"硬失败明细", 校验告警不丢弃行。
+        _val_cnt = len(ext_result.validation) + len(card_result.validation)
+        if _val_cnt:
+            _val_info = (
+                f"入库前校验告警: 315={len(ext_result.validation)}"
+                f" 卡商={len(card_result.validation)}"
+            )
             log.error = (f"{log.error} | {_val_info}") if log.error else _val_info
         if ext_result.dropped or card_result.dropped:
             log.status = "partial"
@@ -573,6 +586,8 @@ def sync_complaints_from_payload(
         return {
             "log_id": log.id,
             "status": log.status,
+            # #42 校验问题清单(只读 advisory; 与 errors 严格分离)
+            "validation": (ext_result.validation + card_result.validation)[:200],
             "external": {
                 "total": ext_result.total,
                 "inserted": ext_result.inserted,

backend/app/services/ingest_service.py

@@ -190,7 +190,8 @@ def ingest_table(
     res.failed = int(getattr(raw, "failed", 0) or 0)
     res.errors = list(getattr(raw, "errors", []) or [])
     res.warnings = list(getattr(raw, "warnings", []) or [])
-    res.rejected = res.failed  # run_upsert 已把 errors 落拒收表
+    # rejected = 真正落拒收表(可按行可见/判定的)条数, 与 failed 不同义(#42)
+    res.rejected = int(getattr(raw, "rejected", 0) or 0)
     res.status = "partial" if res.failed else "success"
     res.extra.setdefault("label", spec.label)
     res.extra.setdefault("dedup_key", spec.dedup_key)
@@ -230,15 +231,29 @@ def ingest_complaints(
     res.updated = int(ext.get("updated", 0)) + int(card.get("updated", 0))
     res.skipped = int(ext.get("filtered", 0)) + int(card.get("filtered", 0))
     res.failed = int(ext.get("failed", 0)) + int(card.get("failed", 0))
-    res.rejected = res.failed
+    # rejected: 拒收行数(由同步函数上报; 无则 0, 不伪装成 failed)
+    res.rejected = int(stats.get("rejected", 0) or 0)
+    # ⚠️ 语义差异: 同步函数返回的 ext/card.warnings 是**计数(int)**,
+    #    而 IngestResult.warnings 是**明细列表(list[dict])** —— 不得直接赋值。
+    #    计数放 extra(供前端/日志展示), 明细由 validation 提供。
     res.errors = list(ext.get("errors", []) or []) + list(card.get("errors", []) or [])
     res.status = stats.get("status", "success")
     res.extra.update({
         "log_id": stats.get("log_id"),
-        "external": {k: ext.get(k) for k in ("total", "inserted", "updated", "failed", "filtered", "matched")},
-        "card": {k: card.get(k) for k in ("total", "inserted", "updated", "failed", "filtered", "matched")},
-        # #42 入库前校验告警数(只读检测, 问题可见化; 明细见同步日志)
-        "validation_warnings": int(ext.get("warnings", 0) or 0) + int(card.get("warnings", 0) or 0),
+        # ⚠️ 字段白名单必须与对外契约一致: warnings/errors 是 workbuddy 依赖的字段,
+        #    遗漏会让外部工具拿到 undefined(2026-09-10 审查 CRITICAL 修复)
+        "external": {
+            k: ext.get(k)
+            for k in ("total", "inserted", "updated", "failed", "filtered", "matched", "warnings", "errors")
+        },
+        "card": {
+            k: card.get(k)
+            for k in ("total", "inserted", "updated", "failed", "filtered", "matched", "warnings", "errors")
+        },
+        # warnings 计数(与对外契约同源)
+        "warnings_count": int(ext.get("warnings", 0) or 0) + int(card.get("warnings", 0) or 0),
+        # #42 入库前校验问题(只读检测; 与 errors 严格分离)
+        "validation": (stats.get("validation", []) or [])[:50],
     })
     logger.info(
         f"[ingest] complaints<-{source} run={rid}: "
@@ -276,7 +291,8 @@ def ingest_manual_refunds(
     res.deleted = int(stats.get("deleted", 0))
     res.skipped = int(stats.get("filtered", 0))
     res.failed = int(stats.get("failed", 0))
-    res.rejected = res.failed
+    # rejected: 拒收行数(由同步函数上报; 无则 0, 不伪装成 failed)
+    res.rejected = int(stats.get("rejected", 0) or 0)
     res.errors = list(stats.get("errors", []) or [])
     res.status = stats.get("status", "success")
     res.extra.update({"log_id": stats.get("log_id"), "warnings_count": stats.get("warnings", 0)})

backend/app/services/ingest_validator.py

@@ -32,17 +32,16 @@ _EXTERNAL_RESULT_SET = {
     "和解结案", "撤销结案", "暂行办结", "调解结案", "调节结案", "投诉无效",
     "重复投诉", "终止调节", "退回消委会", "未达成一致方案", "暂行办理", "需跟进",
     "不予受理", "已提交报告",
-}
-_TICKET_STATUS_SET = {
-    "待处理", "处理中", "已完成", "已关闭", "终止", "已结案", "已解决",
-    "待分派", "跟进中", "pending",
-}
+}# ⚠️ 保留别名兼容: 上游曾有"未达成一致方案"(方案) 与"未达成一致意见"(意见) 两种写法,
+#    两者都应视为合法办结类结果, 避免被 R-枚举 误报。
+_EXTERNAL_RESULT_SET |= {"未达成一致方案"}# 注: 工单(ticket)类校验器尚未接入(本期仅投诉/手动退款); 若后续接入,
+#     其状态集合为 {待处理, 处理中, 已完成, 已关闭, 终止, 已结案, 已解决, 待分派, 跟进中, pending}。
 
 # 合理年份窗口: 未来超 1 天即异常(上游录入笔误);
-# ⚠️ 历史下限放宽到 2015 —— 315 登记表含跨年历史数据(实测最早 2023), 它不是"错误",
-#    只有明显不可能是业务时间的远古年份(如 1969/1970 伪日期)才判异常。
+# 下限与回填脚本 fix_complaint_handle_time._BAD_BEFORE 保持一致(2020),
+# 用于捕获 1969/1970 这类伪日期(Excel 序列号解析 bug 产物)。
 _MAX_FUTURE_DAYS = 1
-_MIN_VALID_YEAR = 2015
+_MIN_VALID_YEAR = 2020
 
 
 def _as_utc(dt: datetime | None) -> datetime | None:
@@ -180,16 +179,26 @@ def validate_rows(rows: list[dict[str, Any]], data_type: str) -> list[dict[str,
     """批量校验 → 问题清单(只读, 不改数据)
 
     未注册校验器的 data_type 返回空清单(不影响入库)。
+    ⚠️ 单行异常不抛出(校验不得阻塞入库), 但会**计入 validator_errors 计数**并告警,
+       避免"系统性报错被逐行静默吞掉"导致校验形同虚设(2026-09-10 审查 WARNING)。
     """
     fn = _VALIDATORS.get(data_type)
     if fn is None or not rows:
         return []
     out: list[dict[str, Any]] = []
+    n_err = 0
     for i, rec in enumerate(rows):
         try:
             out.extend(fn(rec, i))
         except Exception as e:  # noqa: BLE001 校验本身不得阻塞入库
+            n_err += 1
             logger.warning(f"[ingest_validator] {data_type} 第 {i} 行校验异常(跳过): {e}")
+    if n_err:
+        # 以可识别的问题条目上报, 使"校验器自身失效"也能被看到
+        out.append({
+            "row": -1, "rule": "R-校验器", "field": None,
+            "message": f"{n_err} 行校验时发生异常(校验器可能存在缺陷, 需排查)",
+        })
     return out
 
 

backend/app/services/shared_doc_sync_service.py

@@ -20,7 +20,6 @@
 from __future__ import annotations
 
 import logging
-import re
 import tempfile
 from datetime import datetime, timezone
 from pathlib import Path
@@ -33,6 +32,7 @@ from sqlalchemy.orm import Session
 from app.core.config import settings
 from app.core.database import SessionLocal
 from app.models.doc_sync_source import DocSyncSource
+from app.services.data_parser import count_suspect_precision
 from app.services.ingest_service import ingest_complaints, ingest_manual_refunds
 
 logger = logging.getLogger(__name__)
@@ -54,27 +54,6 @@ _KDOCS_MENU_XY = (95, 15)
 # xlsx 读取引擎(与 POC _poc_shared_doc_reader.py 一致: 快 + 免疫 WPS/腾讯导出样式崩溃)
 _EXCEL_ENGINE = "calamine"
 
-# 疑似"长编号被 Excel 数字精度截断"特征(#42 2026-09-10):
-#   源单元格为「数字型」且长度 >=15 时, Excel/浮点只保留约 15 位有效数字,
-#   超出部分被补 0(实测 1440306002024122303480530 → 14403060020241100.../尾串 00000)。
-#   读取层无法修复(源已损坏), 但必须**可见**: 计数 + 告警, 供上游改为文本列。
-_SUSPECT_PRECISION_RE = re.compile(r"^\d{15,}0{4,}$")
-
-
-def _count_suspect_precision(df: "pd.DataFrame") -> dict[str, int]:
-    """统计疑似受浮点精度影响的长数字单元格: {列名: 数量}(仅计数, 不改数据)
-
-    判据: 纯数字、长度>=15、末尾连续>=4 个 0 —— 正常业务编号极少以多个 0 结尾,
-    该形态高度指向「Excel 数字单元格 >15 位被截断」。
-    """
-    hits: dict[str, int] = {}
-    for col in df.columns:
-        s = df[col].astype(str)
-        n = int(s.str.match(_SUSPECT_PRECISION_RE).sum())
-        if n:
-            hits[str(col)] = n
-    return hits
-
 
 def _export_tencent_xlsx(doc_url: str, out_path: str, state_filename: str | None = None) -> str:
     """用腾讯登录态打开共享表 → 更多菜单 → 导出为 → 本地Excel(.xlsx)
@@ -227,8 +206,8 @@ def _read_sheet_records(
     #   显式 dtype=str 至少保证「文本型单元格」100% 原样读取(实测 647/647 一致),
     #   数字型无法挽救但由下方 _count_suspect_precision 计数暴露。
     df = pd.read_excel(path, sheet_name=sheet_name, header=0, engine=_EXCEL_ENGINE, dtype=str)
-    # 精度可疑计数(可见化, 不改数据; #42)
-    for _col, _n in _count_suspect_precision(df).items():
+    # 精度可疑计数(可见化, 不改数据; #42): 复用 data_parser 单一实现
+    for _col, _n in count_suspect_precision(df).items():
         logger.warning(
             f"[shared_doc] {path} sheet={sheet_name} 列「{_col}」有 {_n} 个疑似浮点精度"
             f"截断的长数字(>=15位且尾部多0); 上游应将该列设为文本格式"

backend/app/services/upsert.py

@@ -66,6 +66,10 @@ class ImportResult:
         self.failed = 0
         self.errors: list[dict[str, Any]] = []
         self.warnings: list[dict[str, Any]] = []  # 提示/自动纠错类(非失败, 不阻断)
+        # #42: 实际落拒收表(ingest_rejected_rows)的条数 —— 与 failed 不同义:
+        #   failed=硬失败行数; rejected=其中"带原始行、真正按行可见"的条数
+        #   (无 data 的失败行无法按行展示, 不计入 rejected)。
+        self.rejected = 0
 
     def add_error(self, row_idx: int, msg: str, data: dict | None = None) -> None:
         self.failed += 1
@@ -259,6 +263,27 @@ _EXCEL_EPOCH = datetime(1899, 12, 30)
 _EXCEL_SERIAL_MIN = 20000
 _EXCEL_SERIAL_MAX = 80000
 _PURE_NUM_RE = re.compile(r"^[+-]?\d+(\.\d+)?$")
+# 紧凑日期串 YYYYMMDD(8位, 如 20260614)。上游若把日期列改成无分隔符导出,
+# 它既不在 Excel 序列号区间(会被判 None), 也不是 pd.to_datetime 默认认识的格式。
+# 不处理会导致整列时间静默丢失(2026-09-10 审查 WARNING)。
+_COMPACT_DATE_RE = re.compile(r"^(19|20)\d{6}$")
+
+
+def _try_compact_date(val: Any) -> datetime | None:
+    """尝试把 8 位纯数字按 YYYYMMDD 解析(如 '20260614' → 2026-06-14); 不是则 None。
+
+    仅当形如 (19|20)YYMMDD 且月/日合法时才认, 避免把 8 位编号误当日 期。
+    """
+    if not isinstance(val, str):
+        return None
+    s = val.strip()
+    if not _COMPACT_DATE_RE.match(s):
+        return None
+    try:
+        dt = datetime(int(s[0:4]), int(s[4:6]), int(s[6:8]))
+    except ValueError:
+        return None  # 月/日非法(如 20261340) → 不是日期
+    return dt.replace(tzinfo=_TZ_CN).astimezone(timezone.utc)
 
 
 def _try_excel_serial(val: Any) -> datetime | None:
@@ -273,6 +298,16 @@ def _try_excel_serial(val: Any) -> datetime | None:
     """
     if isinstance(val, bool):  # bool 是 int 子类, 必须显式排除(True→1 天)
         return None
+    # numpy 标量(pd.read_excel/to_dict 可能产出 np.int64/np.float64) → 归一为 Python 标量。
+    # 不归一的话 isinstance(val,(int,float)) 为 False, 会落入 pd.to_datetime(np.int64(46187))
+    # 被当纳秒 → 原 bug(1969 伪日期)复现(2026-09-10 审查 WARNING)。
+    if hasattr(val, "item") and not isinstance(val, (str, bytes)):
+        try:
+            val = val.item()
+        except (ValueError, AttributeError):
+            pass
+    if isinstance(val, bool):  # item() 后可能是 numpy.bool_
+        return None
     num: float | None = None
     if isinstance(val, (int, float)):
         num = float(val)
@@ -290,7 +325,8 @@ def _try_excel_serial(val: Any) -> datetime | None:
     except (TypeError, ValueError):
         return None
     # 浮点精度残渣(如 46187.000000000046 来自 float64 存储) → 归一到最近整数
-    if abs(num - round(num)) < 1e-6:
+    # 容差取 1e-6 天≈86ms 过宽会误伤合法小数时刻, 收紧到 1e-9 天≈86µs(2026-09-10 审查)
+    if abs(num - round(num)) < 1e-9:
         num = float(round(num))
     if not (_EXCEL_SERIAL_MIN <= num <= _EXCEL_SERIAL_MAX):
         return None
@@ -316,6 +352,10 @@ def _to_datetime(val: Any) -> datetime | None:
         return val
     if isinstance(val, str) and val.strip() == "":
         return None
+    # 紧凑日期串 YYYYMMDD(8位): 必须在「区间外纯数字判 None」之前尝试
+    _compact = _try_compact_date(val)
+    if _compact is not None:
+        return _compact
     # Excel 序列日期: 文本型与数字型都要走同一条归一(见 _try_excel_serial 说明)
     _serial = _try_excel_serial(val)
     if _serial is not None:
@@ -323,10 +363,14 @@ def _to_datetime(val: Any) -> datetime | None:
     # ⚠️ 纯数字但不在 Excel 序列号区间 → 不是合法日期语义(如 1 / 0 / 2026 / 12345 / 99999
     #    / 超长编号)。若继续交给 pd.to_datetime, 会被当成**纳秒**解释, 产出 1970 年前后的
     #    伪日期(实测 pd.to_datetime(1)→1969-12-31 16:00:00.000001), 即本函数要根除的坏值类型。
-    #    故此类值一律判为「无法解析 → None」, 由上层校验/拒收流程处理, 绝不静默造伪日期。
+    #    故此类值一律判为「无法解析 → None」+ 告警(可见化), 由上层校验/拒收流程处理。
     if isinstance(val, bool):
         return None
     if isinstance(val, (int, float)) or (isinstance(val, str) and _PURE_NUM_RE.match(val.strip())):
+        logger.warning(
+            f"[upsert._to_datetime] 纯数字但不在 Excel 序列号区间({_EXCEL_SERIAL_MIN}~"
+            f"{_EXCEL_SERIAL_MAX}), 按 None 处理(避免造伪日期): {val!r}"
+        )
         return None
     try:
         result = pd.to_datetime(val)
@@ -2314,11 +2358,15 @@ def upsert_device_profiles(df: pd.DataFrame, mapping: dict[str, str], db: Sessio
     result.updated += max(len(rows) - inserted, 0)
 
 
-def _persist_rejected_from_errors(db: Session, data_type: str, result: ImportResult) -> None:
+def _persist_rejected_from_errors(db: Session, data_type: str, result: ImportResult) -> int:
     """A-20260908-003 (#40): 把 run_upsert 硬失败(丢弃未入库)的行写拒收表。
 
     只处理 errors 中带原始行 data 的(无 data 的仅计数, 无法按行展示)。
     savepoint 隔离: 写拒收失败不影响已完成的 upsert 与主事务。
+
+    返回(#42): 实际写入拒收表的条数 —— 与 failed 不同义:
+      failed  = 硬失败行数(含无 data 无法按行展示的)
+      rejected = 其中真正落库、可按行可见可判定的条数
     """
     import logging as _log
 
@@ -2331,7 +2379,8 @@ def _persist_rejected_from_errors(db: Session, data_type: str, result: ImportRes
             if e.get("data") and isinstance(e.get("data"), dict)
         ]
         if not to_insert:
-            return
+            return 0
+        inserted = 0
         with db.begin_nested():
             for e in to_insert[:200]:  # 上限 200 行/批, 防海量噪音
                 db.add(IngestRejectedRow(
@@ -2343,8 +2392,11 @@ def _persist_rejected_from_errors(db: Session, data_type: str, result: ImportRes
                     source_row_no=e.get("row"),
                     status="open",
                 ))
+                inserted += 1
+        return inserted
     except Exception as _rej_e:  # noqa: BLE001
         _lg.warning(f"[ingest_rejected] 上传错误行落拒收表失败(忽略): {_rej_e}")
+        return 0
 
 
 UPSERT_DISPATCH = {    "refund": upsert_refunds,
@@ -2394,6 +2446,6 @@ def run_upsert(
     # result.errors = 硬失败(行被丢弃未入库), 有 raw data 的写 ingest_rejected_rows
     # 供人工判定(open/keep/discard)。savepoint 隔离, 失败不影响主流程。
     if result.errors:
-        _persist_rejected_from_errors(db, data_type, result)
+        result.rejected = _persist_rejected_from_errors(db, data_type, result)
     db.commit()
     return result

backend/tests/test_ingest_service.py

@@ -81,10 +81,15 @@ class TestIngestResult:
 
 # ---------- 3. 三个入口函数透传既有管线结果 ----------
 class _FakeImportResult:
-    """模拟 upsert.ImportResult"""
+    """模拟 upsert.ImportResult
+
+    #42 修正: rejected(真正落拒收表的行) 与 failed(硬失败行数) **不同义** ——
+    无 raw data 的失败行无法按行展示, 不计入 rejected。
+    """
 
     def __init__(self):
         self.total, self.inserted, self.updated, self.failed = 10, 7, 2, 1
+        self.rejected = 0  # 本行 errors 无 data → 不落拒收表
         self.errors = [{"row": 3, "message": "解析失败"}]
         self.warnings = [{"row": 5, "message": "姓名列被清洗"}]
 
@@ -100,11 +105,29 @@ class TestIngestTable:
             df=None, mapping={}, source="upload", source_ref="upload_record:1",
         )
         assert (res.total, res.inserted, res.updated, res.failed) == (10, 7, 2, 1)
-        assert res.rejected == 1          # 硬失败已落拒收表
+        # failed=1 但 rejected=0(该失败行无 raw data, 无法按行展示) → 二者不得混同
+        assert res.failed == 1
+        assert res.rejected == 0
+        assert res.errors  # 硬失败明细保留
         assert res.status == "partial"    # 有失败 → partial
         assert res.extra["source_ref"] == "upload_record:1"
         assert res.extra["label"] == "退费单"
 
+    def test_rejected_reported_separately(self, monkeypatch):
+        """rejected 取真实拒收计数(与 failed 解耦)"""
+        import app.services.upsert as up
+
+        class _R(_FakeImportResult):
+            def __init__(self):
+                super().__init__()
+                self.failed = 5
+                self.rejected = 3  # 其中 3 条真正落拒收表
+
+        monkeypatch.setattr(up, "run_upsert", lambda df, dt, mp, db: _R())
+        res = ingest_table(db=None, data_type="refund", df=None, mapping={}, source="upload")
+        assert res.failed == 5
+        assert res.rejected == 3
+
     def test_success_when_no_failure(self, monkeypatch):
         """无失败 → status=success"""
         import app.services.upsert as up
@@ -135,9 +158,10 @@ class TestIngestComplaints:
             return {
                 "log_id": 77, "status": "partial",
                 "external": {"total": 5, "inserted": 4, "updated": 1, "failed": 0,
-                             "filtered": 2, "matched": 3, "errors": []},
+                             "filtered": 2, "matched": 3, "warnings": 0, "errors": []},
                 "card": {"total": 3, "inserted": 3, "updated": 0, "failed": 1,
-                         "filtered": 1, "matched": 0, "errors": [{"row": 1}]},
+                         "filtered": 1, "matched": 0, "warnings": 0,
+                         "errors": [{"row": 1, "level": "error"}]},
             }
 
         monkeypatch.setattr(cs, "sync_complaints_from_payload", _fake_sync)
@@ -147,11 +171,36 @@ class TestIngestComplaints:
         assert res.updated == 1
         assert res.skipped == 3          # 2+1 空行
         assert res.failed == 1
-        assert res.rejected == 1
         assert res.status == "partial"
         assert res.extra["log_id"] == 77
         assert res.extra["external"]["total"] == 5
 
+    def test_contract_fields_preserved(self, monkeypatch):
+        """⚠️ 对外契约: external/card 必须含 warnings/errors(审查 CRITICAL 回归锁)
+
+        workbuddy 外部工具依赖这些字段; 白名单遗漏会让其拿到 undefined。
+        """
+        import app.services.complaint_sync as cs
+
+        def _fake_sync(db, payload, source_file=None, sync_source="workbuddy"):
+            return {
+                "log_id": 1, "status": "success",
+                "external": {"total": 1, "inserted": 1, "updated": 0, "failed": 0,
+                             "filtered": 0, "matched": 1, "warnings": 2,
+                             "errors": [{"row": 9, "level": "warning"}]},
+                "card": {"total": 0, "inserted": 0, "updated": 0, "failed": 0,
+                         "filtered": 0, "matched": 0, "warnings": 0, "errors": []},
+            }
+
+        monkeypatch.setattr(cs, "sync_complaints_from_payload", _fake_sync)
+        res = ingest_complaints(db=None, payload={}, source="workbuddy")
+        for key in ("total", "inserted", "updated", "failed", "filtered",
+                    "matched", "warnings", "errors"):
+            assert key in res.extra["external"], f"external 缺契约字段 {key}"
+            assert key in res.extra["card"], f"card 缺契约字段 {key}"
+        assert res.extra["external"]["warnings"] == 2
+        assert res.extra["external"]["errors"]
+
 
 class TestIngestManualRefunds:
     def test_maps_result_and_keeps_full_sync(self, monkeypatch):

backend/tests/test_time_parse_type_agnostic.py

@@ -88,6 +88,14 @@ class TestBackwardCompatible:
     def test_slash_date_string(self):
         assert _to_datetime("2026/6/14") == _SERIAL_46187_UTC
 
+    def test_compact_date_string(self):
+        """8 位紧凑日期串 YYYYMMDD(上游若改格式导出, 不得整列静默丢时间)"""
+        assert _to_datetime("20260614") == _SERIAL_46187_UTC
+
+    def test_compact_date_invalid_md(self):
+        """形如 8 位但月/日非法 → 不认作日期"""
+        assert _to_datetime("20261340") is None
+
     def test_datetime_naive_as_beijing(self):
         assert _to_datetime(datetime(2026, 6, 14, 0, 0)) == _SERIAL_46187_UTC
 
@@ -103,3 +111,26 @@ class TestBackwardCompatible:
         """自由文本(投诉问题等误配到时间列) → None"""
         assert _to_datetime("商家拒绝退款") is None
         assert _to_datetime("待处理") is None
+
+    def test_fractional_serial_keeps_time_of_day(self):
+        """合法小数序列号: 小数部分=当天时刻(不得被精度归一误伤)"""
+        dt = _to_datetime(46187.5)
+        assert dt is not None
+        # 46187 = 2026-06-14(北京) 12:00 → UTC 04:00
+        assert dt == datetime(2026, 6, 14, 4, 0, tzinfo=timezone.utc)
+
+    def test_numpy_scalar_types(self):
+        """numpy 标量必须与 Python 标量等价(否则原纳秒 bug 会复现)"""
+        np = pytest.importorskip("numpy")
+        assert _to_datetime(np.int64(46187)) == _SERIAL_46187_UTC
+        assert _to_datetime(np.float64(46187.0)) == _SERIAL_46187_UTC
+        assert _to_datetime(np.bool_(True)) is None
+
+    def test_unix_timestamp_not_pseudo_date(self):
+        """Unix 秒级时间戳不在序列号区间 → None(不产伪日期)"""
+        assert _to_datetime(1712345678) is None
+        assert _to_datetime("1712345678") is None
+
+    def test_whitespace_padded_serial(self):
+        """带空白文本序列号仍可解析"""
+        assert _to_datetime("  46187  ") == _SERIAL_46187_UTC

scripts/fix_complaint_handle_time.py

@@ -23,15 +23,19 @@
 from __future__ import annotations
 
 import argparse
+import logging
 import re
 import sys
 from datetime import datetime
 from pathlib import Path
+from typing import Any
 
 _ROOT = Path(__file__).resolve().parents[1]
 if str(_ROOT / "backend") not in sys.path:
     sys.path.insert(0, str(_ROOT / "backend"))
 
+logger = logging.getLogger("fix_complaint_handle_time")
+
 import pandas as pd  # noqa: E402
 from sqlalchemy import select  # noqa: E402
 
@@ -105,10 +109,13 @@ def main() -> None:
     ap = argparse.ArgumentParser(description="回填 315 handle_time 坏值(Excel 序列号解析 bug)")
     ap.add_argument("--file", required=True, help="源导出 xlsx 路径")
     ap.add_argument("--apply", action="store_true", help="实际写入(缺省仅预览)")
+    ap.add_argument("--yes", action="store_true", help="与 --apply 搭配时跳过二次确认")
+    ap.add_argument("--max-id", type=int, default=300000,
+                   help="真实主批次 id 上限(排除疑似脏副本; 不同环境 id 段不同, 需按实际调整)")
     args = ap.parse_args()
 
     df = _load_source(args.file)
-    print(f"源表: {args.file} 行数={len(df)}")
+    logger.info(f"源表: {args.file} 行数={len(df)}")
 
     # 源: 构造索引(重算 handle_time)
     src_by_no: dict[str, object] = {}
@@ -136,17 +143,17 @@ def main() -> None:
             src_by_pc[(phone, cd_day)] = ht
         if phone and sn:
             src_by_ps[(phone, sn)] = ht
-    print(f"源表可解析处理时间: {recomputed} 行; 无法解析: {unparsed} 行")
+    logger.info(f"源表可解析处理时间: {recomputed} 行; 无法解析: {unparsed} 行")
     with SessionLocal() as db:
         bad = db.scalars(
             select(ExternalComplaint)
-            .where(ExternalComplaint.id < 300000)
+            .where(ExternalComplaint.id < args.max_id)
             .where(ExternalComplaint.handle_time.isnot(None))
             .where(ExternalComplaint.handle_time < _BAD_BEFORE)
         ).all()
-        print(f"库中坏值行(< {_BAD_BEFORE}): {len(bad)}")
+        logger.info(f"库中坏值行(< {_BAD_BEFORE}, id<{args.max_id}): {len(bad)}")
 
-        matched, missed = [], []
+        matched, missed, ambiguous = [], [], []
         for row in bad:
             no = (row.complaint_no or "").replace("\n", "").strip()
             phone = (row.complainant_phone or "").strip() or None
@@ -169,12 +176,13 @@ def main() -> None:
                     new_ht, hit_by = src_by_ps[tuple(parts)], "手机+SN"
                 if new_ht is not None:
                     break
-            # 容差匹配(编号缺失/日期跨时区的行): 手机号相同 且 日期相差 ≤1 天
+            # 容差匹配(编号缺失/日期跨时区的行): 手机号相同 且 日期相差 ≤1 天。
+            # ⚠️ 安全护栏(2026-09-10 审查): 同手机号当日可能有多条不同投诉,
+            #    必须"唯一候选才回填", 多条候选宁可跳过(记入 ambiguous), 防张冠李戴。
             if new_ht is None and phone:
-                for (p_, cd_s, s_), cand in src_by_pcs.items():
-                    if p_ != phone:
-                        continue
-                    if not (cd_day and cd_s):
+                cands: list[tuple[str, Any]] = []
+                for (p_, cd_s, _s_), cand in src_by_pcs.items():
+                    if p_ != phone or not (cd_day and cd_s):
                         continue
                     try:
                         d1 = datetime.fromisoformat(cd_day)
@@ -182,31 +190,49 @@ def main() -> None:
                     except ValueError:
                         continue
                     if abs((d1 - d2).days) <= _DAY_TOL:
-                        new_ht, hit_by = cand, "手机+日期±1天"
-                        break
+                        cands.append((cd_s, cand))
+                if len(cands) == 1:
+                    new_ht, hit_by = cands[0][1], "手机+日期±1天(唯一)"
+                elif len(cands) > 1:
+                    ambiguous.append((row.id, phone, [c[0] for c in cands]))
             if new_ht is not None:
                 matched.append((row, new_ht, hit_by))
             else:
                 missed.append(row)
 
-        print(f"可回填: {len(matched)}; 未匹配: {len(missed)}")
+        logger.info(f"可回填: {len(matched)}; 未匹配: {len(missed)}; 多候选跳过: {len(ambiguous)}")
         if matched[:5]:
-            print("\n预览(前 5 条):")
+            logger.info("预览(前 5 条):")
             for row, new_ht, hit_by in matched[:5]:
-                print(f"  id={row.id} 坏值={row.handle_time:%Y-%m-%d}"
-                      f" → {new_ht.astimezone().strftime('%Y-%m-%d %H:%M')} [匹配={hit_by}]")
+                logger.info(
+                    f"  id={row.id} 坏值={row.handle_time:%Y-%m-%d}"
+                    f" → {new_ht.astimezone().strftime('%Y-%m-%d %H:%M')} [匹配={hit_by}]"
+                )
+        if ambiguous:
+            logger.warning(f"多候选无法确定 {len(ambiguous)} 行(保持原值, 需人工):")
+            for rid, ph, days in ambiguous[:20]:
+                logger.warning(f"  id={rid} 手机={ph} 候选日期={days}")
         if missed:
             # 未匹配行必须留痕可查(源表无对应/业务键缺失), 便于人工复核
-            print(f"\n未匹配 {len(missed)} 行(源表找不到对应, 保持原值):")
+            logger.warning(f"未匹配 {len(missed)} 行(源表找不到对应, 保持原值):")
             for row in missed[:20]:
-                print(f"  id={row.id} 编号={(row.complaint_no or '(空)')[:24]}"
-                      f" 手机={row.complainant_phone or '(空)'}"
-                      f" SN={row.device_sn or '(空)'}"
-                      f" 登记={(row.complaint_date.date().isoformat() if row.complaint_date else '(空)')}"
-                      f" 问题={(row.problem or '')[:16]}")
+                logger.warning(
+                    f"  id={row.id} 编号={(row.complaint_no or '(空)')[:24]}"
+                    f" 手机={row.complainant_phone or '(空)'}"
+                    f" SN={row.device_sn or '(空)'}"
+                    f" 登记={(row.complaint_date.date().isoformat() if row.complaint_date else '(空)')}"
+                    f" 问题={(row.problem or '')[:16]}"
+                )
 
         if not args.apply:
-            print("\n(预览模式; 加 --apply 才会写库)")
+            logger.info("(预览模式; 加 --apply 才会写库)")
+            return
+        # ⚠️ 写库前二次确认(2026-09-10 审查): 避免误触 --apply 直接改数。
+        if not args.yes:
+            logger.warning(
+                f"即将回填 {len(matched)} 行 handle_time(不可自动回滚)。"
+                f"确认请重跑并加 --yes。"
+            )
             return
 
         n = 0
@@ -214,8 +240,9 @@ def main() -> None:
             row.handle_time = new_ht
             n += 1
         db.commit()
-        print(f"\n✅ 已回填 {n} 行")
+        logger.info(f"✅ 已回填 {n} 行")
 
 
 if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
     main()