fix(cash): 智享通凭证行级幂等键剔除派生period + period_conflict留痕 + 批内行序防吞行 (dama-zxt-idemkey-period-20260911)

缺陷: 行级幂等键含 period(日期派生属性)，summary口径切主表VoucherDate后判重失效→重复入库(实测V17756 6行)。修: 幂等键=(entity_id,voucher_no,subject_code,debit,credit,summary)；命中且period不同→period_conflict计数仅留痕不覆盖；同键多行按载荷内出现次序编号防跨批次吞行；返回体/日志补计数。回归测试10项 + 真实快照41830行复推实证(测试库)。

backend/app/api/cash_vouchers_json.py

@@ -23,9 +23,12 @@
        b) 无 → 查请求体 subject_map {CwAccountID: {subject_code, subject_name}}
            或 DB 表 zxt_account_dict(CwAccountID, subject_code, subject_name)（若已建）
        c) 未命中 → 该行拒收 reason="科目字典未命中 CwAccountID=x"（不写脏数据）
-  6. 幂等：快照每10分钟全量重推同一内容。batch = zxt-{日期}-{行集sha1前10}，
+  6. 幂等：快照每10分钟全量重推同一内容。batch = zxt-{行集sha1前10}，
      导入前查 import_logs 已存在同 batch → 直接返回 already_imported（不再入库）；
-     行级再以 (entity_id, voucher_no, subject_code, debit, credit, summary, period) 防重复。
+     行级再以业务键 (entity_id, voucher_no, subject_code, debit, credit, summary) 防重复
+     —— period 是 voucher_date 派生属性，**不进幂等键**（2026-09-11 加固：日期口径
+     在 summary 派生/主表 VoucherDate 之间切换时，含 period 会判重失效导致重复入库）。
+     命中既有行即跳过；仅当命中行 period 与新行不一致时计入 period_conflict 留痕（不覆盖）。
   7. entity_id：默认 1（陕西酣客，source=zxt-sxzxb20210228），Bot 通道可传 X-Entity-Id 或 body entity_id。
   8. 入库后不触发现金流/ledger 联动（凭证完整性未证实前不作为权威来源，见财务口径警示）；
      也不影响 Excel multipart 导入（原端点不动）。
@@ -308,6 +311,8 @@ def api_import_vouchers_json(
             "pending_undated": pending_undated,
             "pending_dict_miss": pending_dict,
             "neg_rows": neg_rows,
+            "dup_skipped": 0,
+            "period_conflict": 0,
             "hint": "科目字典：请协调智享通推送方补 subject_code/subject_name，或在 body 带 subject_map，或建 zxt_account_dict 表",
         }
 
@@ -362,40 +367,71 @@ def api_import_vouchers_json(
             "total": total,
             "success_rows": 0,
             "failed_rows": 0,
+            "dup_skipped": 0,
+            "period_conflict": 0,
+            "neg_rows": neg_rows,
+            "pending_undated": pending_undated,
+            "pending_dict_miss": pending_dict,
             "batch": batch,
             "entity_id": entity_id,
         }
 
     # ── 入库（行级业务键防重复）──
+    # 幂等键 = (entity_id, voucher_no, subject_code, debit, credit, summary)：
+    #   period 是由 voucher_date 派生的属性，不进唯一性键（否则日期口径变化时会判重失效）。
+    # 同键多行按"载荷内出现次序"编号：第 n 次出现只与库内第 n 行比对——跨批次同键多重数
+    #   时不会因 query().first() 恒命中同一既存行而把源端合法多行静默吞掉。
     inserted = 0
     dup_skipped = 0
+    period_conflict = 0
+    period_conflicts: list = []
+
+    row_groups: dict = {}
     for r in valid_rows:
-        dup = db.query(VoucherDetail).filter(
+        k = (r["voucher_no"], r["subject_code"], r["debit_amount"],
+             r["credit_amount"], r["summary"])
+        row_groups.setdefault(k, []).append(r)
+
+    for k, group_rows in row_groups.items():
+        existing_rows = db.query(VoucherDetail).filter(
             VoucherDetail.entity_id == entity_id,
-            VoucherDetail.voucher_no == r["voucher_no"],
-            VoucherDetail.subject_code == r["subject_code"],
-            VoucherDetail.debit_amount == r["debit_amount"],
-            VoucherDetail.credit_amount == r["credit_amount"],
-            VoucherDetail.summary == r["summary"],
-            VoucherDetail.period == r["period"],
-        ).first()
-        if dup:
-            dup_skipped += 1
-            continue
-        db.add(VoucherDetail(
-            entity_id=entity_id,
-            voucher_no=r["voucher_no"],
-            voucher_date=r["date"],
-            subject_code=r["subject_code"],
-            subject_name=r["subject_name"],
-            debit_amount=r["debit_amount"],
-            credit_amount=r["credit_amount"],
-            summary=r["summary"],
-            carry_forward=r["carry_forward"],
-            period=r["period"],
-            batch=batch,
-        ))
-        inserted += 1
+            VoucherDetail.voucher_no == k[0],
+            VoucherDetail.subject_code == k[1],
+            VoucherDetail.debit_amount == k[2],
+            VoucherDetail.credit_amount == k[3],
+            VoucherDetail.summary == k[4],
+        ).order_by(VoucherDetail.id).all()
+        for pos, r in enumerate(group_rows):
+            if pos < len(existing_rows):
+                dup_skipped += 1
+                old = existing_rows[pos]
+                if (old.period or None) != (r["period"] or None):
+                    # 判重命中但日期口径不同：只留痕不覆盖（留痕优先，人工判定）
+                    period_conflict += 1
+                    if len(period_conflicts) < 50:
+                        period_conflicts.append({
+                            "voucher_no": r["voucher_no"],
+                            "existing_id": old.id,
+                            "existing_period": old.period,
+                            "incoming_period": r["period"],
+                            "existing_batch": old.batch,
+                            "incoming_batch": batch,
+                        })
+                continue
+            db.add(VoucherDetail(
+                entity_id=entity_id,
+                voucher_no=r["voucher_no"],
+                voucher_date=r["date"],
+                subject_code=r["subject_code"],
+                subject_name=r["subject_name"],
+                debit_amount=r["debit_amount"],
+                credit_amount=r["credit_amount"],
+                summary=r["summary"],
+                carry_forward=r["carry_forward"],
+                period=r["period"],
+                batch=batch,
+            ))
+            inserted += 1
 
     period_totals = {}
     for r in valid_rows:
@@ -418,9 +454,14 @@ def api_import_vouchers_json(
     ))
     db.commit()
     logger.info(
-        f"智享通凭证JSON导入[{batch}] entity={entity_id}: 总{total}/入库{inserted}/行级失败{failed}/重复跳过{dup_skipped}, "
-        f"凭证级异常{len(abnormal_vouchers)}"
+        f"智享通凭证JSON导入[{batch}] entity={entity_id}: 总{total}/入库{inserted}/行级失败{failed}/重复跳过{dup_skipped}/"
+        f"期间冲突{period_conflict}, 凭证级异常{len(abnormal_vouchers)}"
     )
+    if period_conflict:
+        logger.warning(
+            f"智享通凭证JSON导入[{batch}] entity={entity_id}: 行级幂等命中但 period 不一致 {period_conflict} 行"
+            f"（未覆盖仅留痕，样例: {period_conflicts[:5]}）"
+        )
 
     return {
         "success": True,
@@ -428,6 +469,8 @@ def api_import_vouchers_json(
         "success_rows": inserted,
         "failed_rows": failed,
         "dup_skipped": dup_skipped,
+        "period_conflict": period_conflict,
+        "period_conflicts": period_conflicts or None,
         "errors": errors[:100] or None,
         "balance_check": {
             "mode": "voucher_level",

backend/tests/test_cash_vouchers_json_idemkey.py

@@ -0,0 +1,183 @@
+"""智享通凭证 JSON 导入 — 行级幂等键加固测试 (dama-zxt-idemkey-period-20260911)
+
+验收覆盖：
+  1. 幂等键不含 period（同业务键、period 不同 → 不重复入库 + period_conflict 计数）
+  2. 同 payload 复推：dup_skipped = 载荷行数、0 新增（含 batch 指纹 already_imported 回归）
+  3. 跨批次同键多重数：「库 1 份 + 载荷 2 份」→ 入库后库内 2 份（不丢行）
+  4. 回归：负数行入库 / 行内 voucher_date 优先 / 借贷不平整张拒收 / 无日期拒收
+"""
+import logging
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import VoucherDetail
+
+BASE = "/api/cma/cash/import/vouchers-json"
+BOT_KEY = "cma-bot-admin-2026"
+H = {"X-BOT-KEY": BOT_KEY}
+
+
+def _row(vno, subject, debit, credit, summary, **kw):
+    r = {"voucher_no": vno, "subject_code": subject, "subject_name": f"科目{subject}",
+         "debit": debit, "credit": credit, "summary": summary}
+    r.update(kw)
+    return r
+
+
+def _balanced(vno, summary, amount="100.00", **kw):
+    """一张平衡凭证：借方一行 + 贷方一行（同摘要，不同科目）"""
+    return [
+        _row(vno, "1001", amount, "0", summary, **kw),
+        _row(vno, "2001", "0", amount, summary, **kw),
+    ]
+
+
+def _count(db: Session, vno: str) -> int:
+    return db.query(VoucherDetail).filter(VoucherDetail.voucher_no == vno).count()
+
+
+def _keys(resp_json: dict) -> None:
+    """返回体计数键齐全性断言"""
+    for k in ("success_rows", "failed_rows", "dup_skipped", "period_conflict",
+              "neg_rows", "pending_undated", "pending_dict_miss"):
+        assert k in resp_json, f"返回体缺计数键 {k}: {resp_json}"
+
+
+class TestIdempotencyKeyWithoutPeriod:
+    def test_same_key_diff_period_no_dup_and_conflict(self, client: TestClient, db: Session):
+        """复现 V17756 场景：首次 period 由 summary 派生(2022-07)，重推改为行内日期(2022-06)
+        → 不重复入库，period_conflict+2，库内 period 保持 2022-07（不覆盖）"""
+        rows = _balanced("17756", "固定资产折旧-2022-07-01-1")
+        r1 = client.post(BASE, json=rows, headers=H)
+        assert r1.status_code == 200, r1.text
+        j1 = r1.json()
+        _keys(j1)
+        assert j1["success_rows"] == 2 and j1["dup_skipped"] == 0
+        assert j1["period_conflict"] == 0
+        assert _count(db, "17756") == 2
+
+        rows2 = _balanced("17756", "固定资产折旧-2022-07-01-1", voucher_date="2022-06-30")
+        r2 = client.post(BASE, json=rows2, headers=H)
+        assert r2.status_code == 200, r2.text
+        j2 = r2.json()
+        _keys(j2)
+        assert j2["success_rows"] == 0, "幂等键含 period 的缺陷：重推被重复入库"
+        assert j2["dup_skipped"] == 2
+        assert j2["period_conflict"] == 2
+        assert j2["period_conflicts"] and j2["period_conflicts"][0]["incoming_period"] == "2022-06"
+        # 库内仍 2 行、period 保持首推口径（不自动覆盖）
+        assert _count(db, "17756") == 2
+        periods = {d.period for d in db.query(VoucherDetail).filter(VoucherDetail.voucher_no == "17756")}
+        assert periods == {"2022-07"}
+
+    def test_period_conflict_logged(self, client: TestClient, db: Session, caplog):
+        """period_conflict 需在日志中留痕"""
+        rows = _balanced("17757", "折旧-2022-07-01-1")
+        client.post(BASE, json=rows, headers=H)
+        with caplog.at_level(logging.WARNING, logger="cma.cash_zxt"):
+            resp = client.post(BASE, json=_balanced("17757", "折旧-2022-07-01-1",
+                                                    voucher_date="2022-06-30"), headers=H)
+        assert resp.json()["period_conflict"] == 2
+        assert any("period" in rec.message and "不一致" in rec.message for rec in caplog.records), \
+            [r.message for r in caplog.records]
+
+
+class TestRepushNoNewRows:
+    def test_exact_same_payload_already_imported(self, client: TestClient, db: Session):
+        """同内容快照重推 → batch 指纹命中 already_imported（既有机制回归）"""
+        rows = _balanced("20001", "同内容-2022-06-01-1")
+        j1 = client.post(BASE, json=rows, headers=H).json()
+        assert j1["success_rows"] == 2
+        j2 = client.post(BASE, json=rows, headers=H).json()
+        assert j2.get("already_imported") is True
+        _keys(j2)
+        assert _count(db, "20001") == 2
+
+    def test_same_rows_repush_dup_skipped_equals_payload_rows(self, client: TestClient, db: Session):
+        """同 payload（行集相同、仅顺序不同 → 指纹不同）复推：dup_skipped = 载荷行数、0 新增"""
+        rows = _balanced("20002", "复推-2022-06-01-1")
+        assert client.post(BASE, json=rows, headers=H).json()["success_rows"] == 2
+        reordered = list(reversed(rows))  # 同载荷行集，指纹不同，走行级幂等
+        j2 = client.post(BASE, json=reordered, headers=H).json()
+        _keys(j2)
+        assert j2["success_rows"] == 0
+        assert j2["dup_skipped"] == len(reordered)
+        assert j2["period_conflict"] == 0
+        assert _count(db, "20002") == 2
+
+    def test_cross_batch_multi_occurrence_no_row_loss(self, client: TestClient, db: Session):
+        """跨批次同键多重数：「库 1 份 + 载荷 2 份」→ 入库后库内 2 份（不丢行）"""
+        # 同额多行凭证：借 2 行(键相同) / 贷 2 行(键相同)
+        dup_debit = _row("20003", "1001", "100.00", "0", "同额多行-2022-06-01-1")
+        dup_credit = _row("20003", "2001", "0", "100.00", "同额多行-2022-06-01-1")
+        batch1 = [dict(dup_debit), dict(dup_credit)]
+        j1 = client.post(BASE, json=batch1, headers=H).json()
+        assert j1["success_rows"] == 2
+        assert _count(db, "20003") == 2
+
+        batch2 = [dict(dup_debit), dict(dup_debit), dict(dup_credit), dict(dup_credit)]
+        j2 = client.post(BASE, json=batch2, headers=H).json()
+        _keys(j2)
+        assert j2["dup_skipped"] == 2, "库内 1 份 + 载荷 2 份：第 2 份必须入库而非被吞"
+        assert j2["success_rows"] == 2
+        assert _count(db, "20003") == 4
+
+        # 再推一次同载荷（4 行已全部在库）→ 全跳过，不增长
+        j3 = client.post(BASE, json=list(reversed(batch2)), headers=H).json()
+        assert j3["success_rows"] == 0 and j3["dup_skipped"] == 4
+        assert _count(db, "20003") == 4
+
+
+class TestRegressions:
+    def test_negative_rows_imported(self, client: TestClient, db: Session):
+        """回归：负数（红字记账）行仍入库并计数 neg_rows"""
+        rows = [
+            _row("30001", "1001", "-100.00", "0", "红字-2022-06-01-1"),
+            _row("30001", "2001", "0", "-100.00", "红字-2022-06-01-1"),
+        ]
+        j = client.post(BASE, json=rows, headers=H).json()
+        _keys(j)
+        assert j["success_rows"] == 2
+        assert j["neg_rows"] == 2
+        assert _count(db, "30001") == 2
+
+    def test_inrow_voucher_date_priority_over_summary(self, client: TestClient, db: Session):
+        """回归：行内 voucher_date 优先于 summary 前缀日期"""
+        rows = _balanced("30002", "口径-2022-07-01-1", voucher_date="2022-06-30")
+        j = client.post(BASE, json=rows, headers=H).json()
+        assert j["success_rows"] == 2
+        periods = {d.period for d in db.query(VoucherDetail).filter(VoucherDetail.voucher_no == "30002")}
+        assert periods == {"2022-06"}, "行内日期未优先"
+
+    def test_unbalanced_voucher_rejected_whole(self, client: TestClient, db: Session):
+        """回归：借贷不平 → 整张凭证拒收"""
+        rows = [_row("30003", "1001", "100.00", "0", "不平-2022-06-01-1")]
+        j = client.post(BASE, json=rows, headers=H).json()
+        _keys(j)
+        assert j["success_rows"] == 0
+        assert j["balance_check"]["passed"] is False
+        assert j["balance_check"]["abnormal_vouchers"][0]["voucher_no"] == "30003"
+        assert _count(db, "30003") == 0
+
+    def test_undated_row_rejected(self, client: TestClient, db: Session):
+        """回归：无任何日期来源 → 拒收（不冒充当前时间）"""
+        rows = [_row("30004", "1001", "100.00", "0", "无日期摘要")]
+        resp = client.post(BASE, json=rows, headers=H)
+        assert resp.status_code == 200
+        j = resp.json()
+        _keys(j)
+        assert j["success"] is False
+        assert j["pending_undated"] == 1
+        assert j["success_rows"] == 0
+        assert _count(db, "30004") == 0
+
+    def test_dict_miss_row_rejected(self, client: TestClient, db: Session):
+        """回归：科目字典未命中 → 该行拒收并计数 pending_dict_miss"""
+        rows = [{"voucher_no": "30005", "CwAccountID": "999999",
+                 "debit": "100.00", "credit": "0", "summary": "缺字典-2022-06-01-1"}]
+        j = client.post(BASE, json=rows, headers=H).json()
+        _keys(j)
+        assert j["pending_dict_miss"] == 1
+        assert _count(db, "30005") == 0