diff --git a/backend/app/api/cash_vouchers_json.py b/backend/app/api/cash_vouchers_json.py index c7fecbf9..b207d455 100644 --- a/backend/app/api/cash_vouchers_json.py +++ b/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", diff --git a/backend/tests/test_cash_vouchers_json_idemkey.py b/backend/tests/test_cash_vouchers_json_idemkey.py new file mode 100644 index 00000000..7310438d --- /dev/null +++ b/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