"""人工改组(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