feat(match): match_payments 支持 manual_groups 人工改组

用户第 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 提交。
This commit is contained in:
wandering
2026-10-21 13:54:24 +08:00
parent 35603837e4
commit 5272314de7
6 changed files with 224 additions and 3 deletions
+101 -1
View File
@@ -59,11 +59,99 @@ def save_invoice_groups(session_dir: Path, groups: dict[str, list[dict[str, Any]
save_json(session_dir / "invoice_groups.json", data)
def match_from_cache(case_dir: Path, tolerance: float | None = None) -> dict[str, Any]:
def _invoice_keys(inv: dict[str, Any]) -> set[str]:
keys = set()
for key in ("invoice_number", "_source_file"):
value = str(inv.get(key) or "").strip()
if value:
keys.add(value)
return keys
def _payment_keys(rec: dict[str, Any]) -> set[str]:
keys = set()
for key in ("_source_file", "card_no"):
value = str(rec.get(key) or "").strip()
if value:
keys.add(value)
return keys
def _apply_manual_groups(
records: list[dict[str, Any]],
invoices: list[dict[str, Any]],
manual_groups: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str]]:
"""按人工指定的分组覆盖贪心匹配结果。
manual_groups 每项形如 {"invoices": [发票号或文件名, ...], "payment": 支付截图文件名或卡号, "note": "..."}。
规则:**一组必须全部唯一命中才应用**(payment 唯一 + 每张发票都能找到),否则整组不应用并进 gaps——
宁可报"没生效"也不能半套应用,那会把账改得更乱。
"""
applied: list[dict[str, Any]] = []
gaps: list[str] = []
for idx, group in enumerate(manual_groups, start=1):
if not isinstance(group, dict):
gaps.append(f"第 {idx} 组不是对象,已跳过")
continue
payment_id = str(group.get("payment") or "").strip()
note = str(group.get("note") or "").strip()
raw_ids = group.get("invoices") or []
if not payment_id or not isinstance(raw_ids, list) or not raw_ids:
gaps.append(f"第 {idx} 组缺少 payment 或 invoices,已跳过")
continue
targets = [rec for rec in records if payment_id in _payment_keys(rec)]
if len(targets) != 1:
reason = "没命中任何支付记录" if not targets else f"命中 {len(targets)} 条支付记录,无法确定是哪一条"
gaps.append(f"第 {idx} 组 payment={payment_id!r} {reason},整组未应用")
continue
resolved: list[dict[str, Any]] = []
unresolved: list[str] = []
for raw in raw_ids:
wanted = str(raw or "").strip()
hits = [inv for inv in invoices if wanted and wanted in _invoice_keys(inv)]
if len(hits) != 1:
unresolved.append(wanted or "(空)")
continue
resolved.append(hits[0])
if unresolved:
gaps.append(f"第 {idx} 组有发票没唯一命中:{'、'.join(unresolved)},整组未应用")
continue
# 先从所有记录里摘掉这些发票,再挂到目标记录(避免同一张票出现在两条记录里)
move_keys: set[str] = set()
for inv in resolved:
move_keys |= _invoice_keys(inv)
for rec in records:
kept = [x for x in rec.get("_matched_invoices", []) if not (_invoice_keys(x) & move_keys)]
if len(kept) != len(rec.get("_matched_invoices", [])):
rec["_matched_invoices"] = kept
target = targets[0]
existing = target.setdefault("_matched_invoices", [])
existing.extend(resolved)
target["remark"] = "manual"
if note:
target["manual_note"] = note
applied.append(
{
"payment": str(target.get("_source_file") or payment_id),
"invoices": [str(inv.get("invoice_number") or inv.get("_source_file") or "") for inv in resolved],
"note": note,
}
)
return records, applied, gaps
def match_from_cache(
case_dir: Path,
tolerance: float | None = None,
manual_groups: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""按缓存内容执行金额匹配,返回结构化结果。
返回字段里 review_required 是必须人工/模型复核的可疑点,
不要当成错误——贪心匹配本来就可能在等额发票上给出错误分配。
manual_groups 见 _apply_manual_groups:人工指定「这几张票就是这一笔支付」,覆盖贪心结果;
没全部唯一命中的组不会应用,原因回在 manual_gaps 里(不许静默失败)。
"""
cache_map = load_cache(case_dir)
buckets = split_cache(cache_map)
@@ -79,6 +167,10 @@ def match_from_cache(case_dir: Path, tolerance: float | None = None) -> dict[str
kwargs: dict[str, Any] = {} if tolerance is None else {"tolerance": tolerance}
records = match_invoices_to_cards(invoices, cards, **kwargs)
manual_applied: list[dict[str, Any]] = []
manual_gaps: list[str] = []
if manual_groups:
records, manual_applied, manual_gaps = _apply_manual_groups(records, invoices, manual_groups)
cache_dir = case_dir / ".invoice_cache"
save_match_result(records, cache_dir)
@@ -143,6 +235,12 @@ def match_from_cache(case_dir: Path, tolerance: float | None = None) -> dict[str
if multi:
review.append(f"{len(multi)} 笔支付走了一对多贪心凑额,分配结果需逐笔核对(等额发票尤其容易配错)")
if manual_applied:
review.append(
f"{len(manual_applied)} 组按人工指定覆盖了贪心匹配(对应记录 remark=manual),下游 CSV 已按新分组重算"
)
if manual_gaps:
review.append(f"人工分组有 {len(manual_gaps)} 条没生效:" + ";".join(manual_gaps))
has_travel = bool(groups.get("travel"))
has_general = bool(groups.get("general"))
if has_travel and not has_general:
@@ -169,4 +267,6 @@ def match_from_cache(case_dir: Path, tolerance: float | None = None) -> dict[str
"attachments": len(groups.get("attachments", [])),
},
"suggested_invoice_type": suggested,
"manual_applied": manual_applied,
"manual_gaps": manual_gaps,
}