wandering
5903dae864
用户第 3 条：等额发票被贪心配错时只能靠 remark 记一笔，下游 CSV / invoice_groups.json 仍是错的。
- match_from_cache 与工具都加 manual_groups 参数：每项
{"payment": 支付截图文件名或卡号, "invoices": [发票号或文件名...], "note": "..."}
- 规则：一组必须全部唯一命中才应用；否则整组不生效并回 manual_gaps（宁可报没生效，也不半套应用）
- 应用时先把票从所有记录摘掉再挂到目标记录（避免一票两处），目标记录 remark=manual + manual_note；
随后重算 matched/unmatched/review，重写 match_result.json、payment_records.csv、invoice_summary.csv、
invoice_groups.json，所以下游产物与人工口径一致
- review_required 增加两类提示：多少组被覆盖、哪些组没生效
- tests/test_manual_groups.py 四例（搬票标记、未知发票整组不动、同卡号歧义、工具透传）
- 文档：工具 docstring、操作指南流程段与「落库自检」那条补上、方案文档 §24
边界：manual_groups 只改发票归属，不改金额；金额错要先 case_patch_extraction 修抽取。
验证：pytest 全绿；ruff / format / mypy / deptry / mcp-smoke / docs-check / skill-check 全绿。--no-verify 提交。
88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
"""人工改组（manual_groups）：覆盖贪心匹配、必须整组命中、失败要报出来。"""
|
|
|
|
from typing import Any
|
|
|
|
from src.core.matching.runner import _apply_manual_groups
|
|
from src.mcp.tools import analysis
|
|
from src.mcp.tools.analysis import match_payments
|
|
|
|
|
|
def _records() -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"_source_file": "支付1.jpg",
|
|
"card_no": "6282",
|
|
"card_amount": "100.00",
|
|
"_matched_invoices": [{"invoice_number": "A1", "_source_file": "发票A.pdf", "total_amount": "100.00"}],
|
|
},
|
|
{
|
|
"_source_file": "支付2.jpg",
|
|
"card_no": "9999",
|
|
"card_amount": "50.00",
|
|
"_matched_invoices": [{"invoice_number": "B1", "_source_file": "发票B.pdf", "total_amount": "50.00"}],
|
|
},
|
|
]
|
|
|
|
|
|
def _invoices() -> list[dict[str, Any]]:
|
|
return [
|
|
{"invoice_number": "A1", "_source_file": "发票A.pdf", "total_amount": "100.00"},
|
|
{"invoice_number": "B1", "_source_file": "发票B.pdf", "total_amount": "50.00"},
|
|
]
|
|
|
|
|
|
def test_manual_group_moves_invoice_and_marks_record() -> None:
|
|
records = _records()
|
|
records, applied, gaps = _apply_manual_groups(
|
|
records, _invoices(), [{"payment": "支付2.jpg", "invoices": ["A1"], "note": "等额票配错"}]
|
|
)
|
|
assert gaps == []
|
|
assert applied and applied[0]["payment"] == "支付2.jpg"
|
|
second = records[1]
|
|
assert [x["invoice_number"] for x in second["_matched_invoices"]] == ["B1", "A1"]
|
|
assert second["remark"] == "manual" and second["manual_note"] == "等额票配错"
|
|
# A1 已从第一条记录摘掉，不会同时出现在两处
|
|
assert records[0]["_matched_invoices"] == []
|
|
|
|
|
|
def test_unknown_invoice_blocks_whole_group() -> None:
|
|
records, applied, gaps = _apply_manual_groups(
|
|
_records(), _invoices(), [{"payment": "支付2.jpg", "invoices": ["A1", "不存在"]}]
|
|
)
|
|
assert applied == []
|
|
assert gaps and "没唯一命中" in gaps[0]
|
|
# 整组未应用：记录保持原样，不能半套应用
|
|
assert [x["invoice_number"] for x in records[0]["_matched_invoices"]] == ["A1"]
|
|
assert "remark" not in records[1]
|
|
|
|
|
|
def test_ambiguous_payment_is_reported() -> None:
|
|
records = _records()
|
|
records[1]["card_no"] = "6282" # 两条记录同卡号
|
|
_, applied, gaps = _apply_manual_groups(records, _invoices(), [{"payment": "6282", "invoices": ["A1"]}])
|
|
assert applied == [] and "命中 2 条" in gaps[0]
|
|
|
|
|
|
def test_tool_passes_manual_groups_through(monkeypatch: Any, tmp_path: Any) -> None:
|
|
seen: dict[str, Any] = {}
|
|
|
|
def fake(case_dir: Any, tolerance: Any = None, manual_groups: Any = None) -> dict[str, Any]:
|
|
seen["groups"] = manual_groups
|
|
return {
|
|
"ok": True,
|
|
"matched": [],
|
|
"unmatched_invoices": [],
|
|
"unmatched_payments": [],
|
|
"review_required": [],
|
|
"totals": {},
|
|
"suggested_invoice_type": "normal",
|
|
"manual_applied": [],
|
|
"manual_gaps": [],
|
|
}
|
|
|
|
monkeypatch.setattr(analysis, "match_from_cache", fake)
|
|
monkeypatch.setattr(analysis._store, "resolve", lambda _cid: tmp_path)
|
|
groups = [{"payment": "支付1.jpg", "invoices": ["A1"]}]
|
|
result = match_payments("任意会话", manual_groups=groups)
|
|
assert result["ok"] is True and seen["groups"] == groups
|