"""发票与支付记录匹配 将提取到的发票数据与支付记录进行金额匹配， 输出以支付记录为主键的结果列表。 支付记录由统一提取模块（extractor）根据 LLM 返回的「invoice_type」字段分类而来， 不再按文件类型假设文档类型。 ## 业务约束 - 发票数 >= 付款记录数（最少一张发票对应一张付款记录） - 发票总金额 >= 付款总金额（发票只能比付款多，不能少） - 若发票数 == 付款数，走一对一匹配，无需一对多 ## 匹配流程 1. 接收分类好的发票和支付记录列表 2. 解析发票和刷卡记录的金额，进行总额校验 - 发票总额 < 刷卡总额时发出 warning 3. 按金额降序排序 4. 文件名匹配（最高优先级）：发票和刷卡记录的文件名（不含后缀）一致时直接匹配 5. 根据数量关系选择匹配策略： - 数量相等 → 一对一匹配：按金额从大到小依次配对，相对容差内即匹配 - 发票更多 → 一对多匹配：对每张刷卡记录贪心凑金额，相对容差内结束 6. 构建以支付记录为主键的结果列表 7. 未匹配的发票单独作为一条记录（无刷卡信息） 8. 清理内部字段，输出支付记录列表 ## 容差计算 使用相对容差（默认 3%），以刷卡金额为基准： - ¥2900 发票 vs ¥2850 刷卡 → 差 ¥50，容差 ¥85.5 → 匹配成功 - ¥100 发票 vs ¥95 刷卡 → 差 ¥5，容差 ¥3.0 → 不匹配（需精确匹配或调整） ## 一对多匹配细节 - 对每张刷卡记录，维护 remaining（剩余待匹配金额） - 遍历未分配的发票（按金额降序）： - 若发票金额 + 容差 >= remaining，视为最后一张，匹配后退出 - 否则发票金额不超过 remaining + 容差即可匹配 - 匹配后 remaining 为负且超出容差时回滚最后一张发票 - 每张发票只会被分配一次 ## 对外接口 match_invoices_to_cards(invoices, cards, tolerance) -> list[dict] """ from pathlib import Path from typing import Any from .. import get_logger log = get_logger("matcher") def _safe_float(value: str | None, default: float = 0.0) -> float: """安全转换为浮点数""" if value is None or str(value).strip() == "": return default try: return float(str(value).replace(",", "")) except (ValueError, TypeError): return default def _build_invoice_summary(invoices: list[dict[str, Any]]) -> str: """将多张发票信息汇总为备注字符串""" parts = [] for inv in invoices: inv_type = inv.get("invoice_type", "unknown") amount = inv.get("total_amount", "unknown") if inv_type == "train": label = inv.get("person_name") or inv.get("invoice_number", "unknown") elif inv_type == "hotel": label = "hotel" else: label = inv.get("item_name") or inv.get("invoice_number", "unknown") parts.append(f"{inv_type}[{label}]¥{amount}") return " | ".join(parts) def _relative_tolerance(base: float, rate: float = 0.03) -> float: """根据基准金额计算相对容差（默认 3%）""" return abs(base) * rate def match_invoices_to_cards( invoices: list[dict[str, Any]], cards: list[dict[str, Any]] | None = None, tolerance: float = 0.03, ) -> list[dict[str, Any]]: """将发票与支付记录按金额匹配，输出以支付记录为主键的结果列表 支付记录由统一提取模块（extractor）根据 LLM 返回的「invoice_type」字段分类而来。 业务约束： - 发票数 >= 付款记录数 - 发票总金额 >= 付款总金额（发票只能比付款多，不能少） - 若发票数 == 付款数，一对一匹配，无需一对多 Args: invoices: 发票列表，需包含 "total_amount" 字段 cards: 支付记录列表，需包含 "card_amount" 字段（由 extractor 分类提供） tolerance: 金额匹配容差比例（默认 0.03 = 3%） Returns: 以支付记录为主键的结果列表，每条记录包含： - card_date, card_no, card_amount（支付信息） - 关联发票列表（_matched_invoices） - 发票详情备注 - 未匹配发票单独作为一条无刷卡信息的记录 """ if not cards: log.warning("无支付记录可供匹配，发票将保持原状") return _invoices_to_records(invoices) # 解析金额 for card in cards: card["_amount"] = _safe_float(card.get("card_amount")) for inv in invoices: inv["_amount"] = _safe_float(inv.get("total_amount")) # 数据校验 total_invoices = sum(inv["_amount"] for inv in invoices) total_cards = sum(card["_amount"] for card in cards) log.info( f"金额校验: 发票总额 ¥{total_invoices:.2f}, 刷卡总额 ¥{total_cards:.2f}, " f"发票数 {len(invoices)}, 刷卡数 {len(cards)}" ) total_tolerance = _relative_tolerance(max(total_invoices, total_cards), tolerance) if total_invoices < total_cards - total_tolerance: log.warning( f"发票总额 (¥{total_invoices:.2f}) 小于刷卡总额 (¥{total_cards:.2f}), " f"超出容差 {tolerance * 100:.0f}%, 匹配结果可能有偏差" ) # 按金额降序排序 cards.sort(key=lambda c: c["_amount"], reverse=True) invoices.sort(key=lambda i: i["_amount"], reverse=True) # 执行匹配，返回 {card_index: [invoice_indices]} 的映射 card_to_invoices = _match(cards, invoices, tolerance) # 构建支付记录列表 records = _build_payment_records(cards, invoices, card_to_invoices) # 清理内部字段 for inv in invoices: inv.pop("_amount", None) for card in cards: card.pop("_amount", None) # 统计 matched_invoices = sum(len(inv_list) for inv_list in card_to_invoices.values()) unmatched_count = len(invoices) - matched_invoices log.info(f"匹配完成: {len(records)} 条支付记录, {matched_invoices}/{len(invoices)} 张发票已关联") if unmatched_count: log.info(f"未匹配发票: {unmatched_count} 张（已单独列为记录）") return records def _match( cards: list[dict[str, Any]], invoices: list[dict[str, Any]], tolerance: float, ) -> dict[int, list[int]]: """执行匹配，返回 {card_index: [invoice_indices]} 的映射 tolerance 为相对容差比例（如 0.03 表示 3%） 匹配优先级（从高到低）： 1. 文件名匹配：发票和刷卡记录的文件名（不含后缀）一致时直接匹配 2. 精确匹配：金额差 <= 0.01 元 3. 一对一 / 一对多贪心匹配：按金额容差匹配 """ result: dict[int, list[int]] = {} assigned: set[int] = set() # ---- 阶段 0：文件名匹配（最高优先级）---- _match_by_filename(invoices, cards, assigned, result) if len(invoices) == len(cards): _match_one_to_one(invoices, cards, tolerance, assigned, result) else: _match_one_to_many(invoices, cards, tolerance, assigned, result) return result def _match_by_filename( invoices: list[dict[str, Any]], cards: list[dict[str, Any]], assigned: set[int], result: dict[int, list[int]], ) -> None: """文件名匹配：发票和刷卡记录的文件名（不含后缀）一致时直接匹配""" for card_idx, card in enumerate(cards): card_name = card.get("_source_file", "") if not card_name: continue card_stem = Path(card_name).stem for idx, inv in enumerate(invoices): if idx in assigned: continue inv_name = inv.get("_source_file", "") if not inv_name: continue inv_stem = Path(inv_name).stem if inv_stem == card_stem: assigned.add(idx) result[card_idx] = [idx] log.info(f"[文件名匹配] {inv_name} ↔ {card_name}") break def _match_one_to_one( invoices: list[dict[str, Any]], cards: list[dict[str, Any]], tolerance: float, assigned: set[int], result: dict[int, list[int]], ) -> None: """一对一匹配：发票数等于刷卡数，对每张刷卡记录寻找金额最接近的未分配发票""" for card_idx, card in enumerate(cards): if card_idx in result: continue card_amount = card["_amount"] card_tol = _relative_tolerance(card_amount, tolerance) # 在未分配的发票中找金额最接近的 best_idx = -1 best_diff = float("inf") for idx, inv in enumerate(invoices): if idx in assigned: continue diff = abs(inv["_amount"] - card_amount) if diff < best_diff: best_diff = diff best_idx = idx if best_idx >= 0 and best_diff <= card_tol: inv = invoices[best_idx] assigned.add(best_idx) result[card_idx] = [best_idx] log.info( f"[一对一] {inv.get('invoice_number', 'unknown')} ¥{inv['_amount']:.2f} " f"↔ {card.get('_source_file', 'unknown')} ¥{card_amount:.2f}" ) elif best_idx >= 0: inv = invoices[best_idx] log.warning( f"[一对一] 金额偏差超出容差: " f"{inv.get('invoice_number', 'unknown')} ¥{inv['_amount']:.2f} " f"vs ¥{card_amount:.2f} (差 ¥{best_diff:.2f}, 容差 ¥{card_tol:.2f})" ) def _match_one_to_many( invoices: list[dict[str, Any]], cards: list[dict[str, Any]], tolerance: float, assigned: set[int], result: dict[int, list[int]], ) -> None: """一对多匹配：一张刷卡可能对应多张发票，按金额从大到小贪心匹配""" # ---- 阶段 1：精确匹配（金额差 <= 0.01 元视为相等）---- exact_tolerance = 0.01 for card_idx, card in enumerate(cards): card_amount = card["_amount"] if card_amount <= 0: continue for idx, inv in enumerate(invoices): if idx in assigned: continue inv_amount = inv["_amount"] if inv_amount <= 0: continue if abs(inv_amount - card_amount) <= exact_tolerance: assigned.add(idx) result[card_idx] = [idx] log.info( f"[一对多-精确] {inv.get('invoice_number', 'unknown')} ¥{inv_amount:.2f} " f"↔ {card.get('_source_file', 'unknown')} ¥{card_amount:.2f}" ) break # ---- 阶段 2：贪心匹配（仅处理未精确匹配的刷卡记录）---- for card_idx, card in enumerate(cards): if card_idx in result: continue card_amount = card["_amount"] if card_amount <= 0: continue card_tol = _relative_tolerance(card_amount, tolerance) remaining = card_amount matched_indices: list[int] = [] for idx, inv in enumerate(invoices): if idx in assigned: continue if remaining <= card_tol: break inv_amount = inv["_amount"] if inv_amount <= 0: continue if inv_amount + card_tol >= remaining: is_match = True else: is_match = inv_amount <= remaining + card_tol if is_match: assigned.add(idx) matched_indices.append(idx) remaining -= inv_amount if remaining <= card_tol: break # 回滚：如果匹配后 remaining 为负且超出容差 if remaining < -card_tol and matched_indices: last_idx = matched_indices.pop() assigned.discard(last_idx) remaining += invoices[last_idx]["_amount"] # 记录匹配结果 if matched_indices: result[card_idx] = matched_indices for idx in matched_indices: inv = invoices[idx] log.info( f"[一对多-贪心] {inv.get('invoice_number', 'unknown')} ¥{inv['_amount']:.2f} " f"→ {card.get('_source_file', 'unknown')} ¥{card['_amount']:.2f}" ) def _build_payment_records( cards: list[dict[str, Any]], invoices: list[dict[str, Any]], card_to_invoices: dict[int, list[int]], ) -> list[dict[str, Any]]: """构建以支付记录为主键的结果列表""" records: list[dict[str, Any]] = [] for card_idx, inv_indices in card_to_invoices.items(): card = cards[card_idx] matched_invs = [invoices[idx] for idx in inv_indices] record = { "card_date": card.get("card_date", ""), "card_no": card.get("card_no", ""), "card_amount": str(card["_amount"]), "relative_invoice_count": str(len(matched_invs)), "invoice_detail": _build_invoice_summary(matched_invs), "remark": "", "_source_file": card.get("_source_file", ""), "_matched_invoices": matched_invs, } records.append(record) # 未匹配的发票，单独作为记录 matched_indices = set() for inv_indices in card_to_invoices.values(): matched_indices.update(inv_indices) unmatched = [inv for idx, inv in enumerate(invoices) if idx not in matched_indices] for inv in unmatched: record = { "card_date": "", "card_no": "", "card_amount": "", "relative_invoice_count": "1", "invoice_detail": _build_invoice_summary([inv]), "remark": "unmatched", "_matched_invoices": [inv], } records.append(record) return records def _invoices_to_records(invoices: list[dict[str, Any]]) -> list[dict[str, Any]]: """无刷卡记录时，将每张发票转为独立记录""" records = [] for inv in invoices: record = { "card_date": "", "card_no": "", "card_amount": "", "relative_invoice_count": "1", "invoice_detail": _build_invoice_summary([inv]), "remark": "", "_matched_invoices": [inv], } records.append(record) return records