from __future__ import annotations import argparse import re import shutil from collections import defaultdict from copy import copy from dataclasses import dataclass, field from datetime import date, datetime from difflib import SequenceMatcher from pathlib import Path from typing import Iterable from openpyxl import load_workbook DEFAULT_DIR = Path("downloads") / "wehago_account_ledger_20260421" DEFAULT_LEDGER = DEFAULT_DIR / "\ud1b5\ud569_\uacc4\uc815\ubcc4\uc6d0\uc7a5_20260422_edit_ver.2.xlsx" DEFAULT_VOUCHER = DEFAULT_DIR / "\uc804\ud45c\ub0b4\uc5ed\uc870\ud68c_2025_20250101_20251231_260422.xlsx" LEDGER_OUT_NAME = "ledger_with_matched_voucher_detected_context60_v10_20260422.xlsx" VOUCHER_OUT_NAME = "voucher_with_matched_ledger_detected_context60_v10_20260422.xlsx" LEDGER_HEADERS = [ "matched_\uac00\uc804\ud45c\ubc88\ud638", "matched_\uacc4\uc815\ucf54\ub4dc", "matched_\uacc4\uc815\uba85\uce6d", "matched_\ucc28\ubcc0\uacf5\uae09\uac00", "matched_\ub300\ubcc0\uacf5\uae09\uac00", "matched_\ud655\uc815\uc804\ud45c\ubc88\ud638", "matched_\uac00\uc804\ud45c\uc791\uc131\uc77c", "matched_\uc77c\uc790\ucc28\uc774\uc77c", "matched_\uac80\uc99d\uadfc\uac70", "review_flag", "review_memo", ] VOUCHER_HEADERS = [ "matched_\uacc4\uc815\ucf54\ub4dc", "matched_\uacc4\uc815\uba85", "matched_\uc77c\uc790", "matched_\ucc28\ubcc0", "matched_\ub300\ubcc0", "matched_\uc804\ud45c\ubc88\ud638", "matched_\uc77c\uc790\ucc28\uc774\uc77c", "matched_\uac80\uc99d\uadfc\uac70", "review_flag", "review_memo", ] # Strict rules for balance-sheet, tax, revenue, and other accounts where # counterparty lines must not be treated as the same accounting line. EXACT_RULES = { "10110501": {"103"}, "10115301": {"135"}, "20110101": {"251"}, "20111103": {"253"}, "20111105": {"253"}, "20111108": {"253"}, "20111109": {"253"}, "20111111": {"253"}, "20112901": {"255"}, } PREFIX_RULES = [ ("101101", {"101"}), ("101105", {"103"}), ("101106", {"104"}), ("101107", {"106", "123"}), ("101111", {"108", "251"}), ("101115", {"108"}), ("101119", {"120"}), ("101129", {"114"}), ("101137", {"131"}), ("101139", {"133"}), ("101153", {"135"}), ("101155", {"136"}), ("101157", {"137"}), ("101159", {"138"}), ("201101", {"251"}), ("201111", {"253"}), ("20111501", {"254"}), ("20111503", {"254"}), ("20111505", {"275"}), ("20111507", {"274"}), ("20111509", {"276"}), ("20111511", {"276"}), ("201115", {"254"}), ("201129", {"255"}), ("40110101", {"411"}), ("40110102", {"412"}), ("40110103", {"415"}), ("40110301", {"413"}), ("40110401", {"414"}), ("501701", {"602"}), ("701101", {"901"}), ("701103", {"903"}), ("701105", {"905"}), ("701107", {"906"}), ("701109", {"908"}), ("701111", {"914"}), ("701115", {"930"}), ("703101", {"931"}), ("703103", {"933"}), ("703105", {"935"}), ("703107", {"937"}), ("703125", {"960"}), ("901", {"998"}), ] @dataclass class VoucherLine: id: int row: int group_id: str draft_no: str confirmed_no: str draft_date: date | None side: str amount: int account_code: str account_name: str desc: str vendor: str group_blob: str debit_supply: int credit_supply: int tax_debit: int tax_credit: int norm_desc: str norm_vendor: str norm_blob: str match_id: int | None = None score: int = 0 reason: str = "" date_diff: int | None = None match_ledger_ids: list[int] = field(default_factory=list) @dataclass class VoucherAggregate: id: int group_id: str row: int draft_no: str confirmed_no: str draft_date: date | None side: str amount: int account_code: str account_name: str desc: str vendor: str group_blob: str line_ids: list[int] norm_desc: str norm_vendor: str norm_blob: str match_id: int | None = None score: int = 0 reason: str = "" date_diff: int | None = None @dataclass class LedgerLine: id: int row: int ledger_date: date side: str amount: int account_code: str account_name: str desc: str vendor: str debit: int credit: int voucher_no: str source_row_no: str norm_desc: str norm_vendor: str norm_blob: str match_kind: str = "" match_id: int | None = None score: int = 0 reason: str = "" date_diff: int | None = None @dataclass(order=True) class Candidate: sort_key: tuple ledger_id: int = field(compare=False) voucher_kind: str = field(compare=False) voucher_id: int = field(compare=False) score: int = field(compare=False) reason: str = field(compare=False) date_diff: int = field(compare=False) def clean(value) -> str: return "" if value is None else str(value).strip() def norm(value) -> str: text = clean(value).lower() text = re.sub(r"\s+", "", text) return re.sub(r"[\(\)\[\]\{\},._\-/\\:;*×%◇]", "", text) def norm_party(value) -> str: text = clean(value).lower() text = text.replace("㈜", "") text = text.replace("(주)", "").replace("（주）", "") text = text.replace("주식회사", "").replace("주.", "") return norm(text) def amount(value) -> int: if value is None or value == "": return 0 if isinstance(value, bool): return int(value) if isinstance(value, (int, float)): return int(round(value)) try: return int(round(float(str(value).replace(",", "").strip()))) except ValueError: return 0 def ymd(value) -> date | None: if value is None or value == "": return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value text = str(value).strip().replace(".0", "") match = re.match(r"^(\d{4})(\d{2})(\d{2})$", text) if match: return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) match = re.match(r"^(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})$", text) if match: return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) return None def parse_ledger_date(value) -> date | None: if isinstance(value, datetime): return value.date() if isinstance(value, date): return value text = clean(value) match = re.match(r"^(\d{1,2})[-/.](\d{1,2})$", text) if match: return date(2025, int(match.group(1)), int(match.group(2))) return ymd(text) def date_from_voucher_no(value) -> date | None: match = re.search(r"-(\d{8})-", clean(value)) return ymd(match.group(1)) if match else None def mmdd(value: date | None) -> str: return value.strftime("%m-%d") if value else "" def group_no(value) -> str: return re.sub(r"-\d+$", "", clean(value)) def draft_group_no(value) -> str: return group_no(value) def setting_group_no(value: str) -> str: match = re.search(r"(11-\d{8}-[A-Z0-9]+)-0*(\d+)-0*\d+", clean(value)) if not match: return "" return f"{match.group(1)}-{int(match.group(2))}" def setting_group_refs(value: str) -> list[str]: refs = [] for match in re.finditer(r"(11-\d{8}-[A-Z0-9]+)-0*(\d+)-0*\d+", clean(value)): refs.append(f"{match.group(1)}-{int(match.group(2))}") return list(dict.fromkeys(refs)) def account_base(name: str) -> str: text = clean(name) if ")" in text and text.index(")") <= 3: text = text.split(")", 1)[1] text = re.sub(r"\(.*?\)|\[.*?\]", "", text) return norm(text) ACCOUNT_EQUIVALENT_GROUPS = [ {"외상매출금", "용역미수금"}, {"단기대여금", "관계회사단기대여금"}, ] def equivalent_account_names(left_name: str, right_name: str) -> bool: left = account_base(left_name) right = account_base(right_name) if not left or not right: return False if left in right or right in left: return True for group in ACCOUNT_EQUIVALENT_GROUPS: normalized = {norm(item) for item in group} if any(item in left for item in normalized) and any(item in right for item in normalized): return True return False def balance_sheet_context_accounts(voucher, ledger: LedgerLine) -> bool: voucher_code = clean(voucher.account_code) return ledger.account_code in {"108", "114", "131", "251"} and voucher_code[:3] in {"101", "201"} def economic_side_amount(side: str, line_amount: int) -> tuple[str, int, str]: if line_amount < 0: flipped = "credit" if side == "debit" else "debit" return flipped, abs(line_amount), "negative_amount_side_flip" return side, line_amount, "" def reversal_context_accounts(voucher, ledger: LedgerLine) -> bool: voucher_code = clean(voucher.account_code) receivable_ledger = ledger.account_code in {"108", "251"} payable_voucher = voucher_code.startswith("201111") or voucher_code.startswith("201101") return receivable_ledger and payable_voucher def negative_reversal_context(voucher, ledger: LedgerLine) -> bool: ledger_side, ledger_amount, ledger_reason = economic_side_amount(ledger.side, ledger.amount) voucher_side, voucher_amount, voucher_reason = economic_side_amount(voucher.side, voucher.amount) if not ledger_reason and not voucher_reason: return False if (ledger_side, ledger_amount) != (voucher_side, voucher_amount): return False return reversal_context_accounts(voucher, ledger) def source_line_for_multi_sum(line: VoucherLine) -> bool: if line.account_code.startswith("2"): return False if line.account_code.startswith("101101") or line.account_code.startswith("101105"): return False return True def allowed_codes(satis_code: str) -> set[str]: code = clean(satis_code).replace(".0", "") if code in EXACT_RULES: return EXACT_RULES[code] for prefix, allowed in PREFIX_RULES: if code.startswith(prefix): return allowed return set() def compatible_account(voucher, ledger: LedgerLine) -> tuple[bool, str]: allowed = allowed_codes(voucher.account_code) if allowed and ledger.account_code in allowed: return True, "semantic_code_ok" # Cost/expense accounts often differ only by SATIS detail and WEHAGO's # 6xx/8xx split. Use the normalized account-name base as a secondary rule. if voucher.account_code[:1] in ("5", "6") and ledger.account_code[:1] in ("6", "8"): if equivalent_account_names(voucher.account_name, ledger.account_name): return True, "account_name_base_ok" if equivalent_account_names(voucher.account_name, ledger.account_name): return True, "account_name_equivalent_ok" return False, "semantic_mismatch" def ledger_vendor_key(vendor: str) -> str: text = clean(vendor) if "/" in text: text = text.split("/")[-1] return norm_party(text) def vendor_status(ledger: LedgerLine, voucher) -> str: ledger_key = ledger_vendor_key(ledger.vendor) voucher_key = norm_party(voucher.vendor) voucher_blob = voucher.norm_blob if not ledger_key or not voucher_key: return "missing" if ledger_key in voucher_key or voucher_key in ledger_key or ledger_key in voucher_blob: return "match" if "/" in clean(ledger.vendor) and len(ledger_key) >= 2: return "mismatch" return "weak_mismatch" def date_score(diff: int) -> int: if diff == 0: return 140 if diff <= 3: return 120 if diff <= 7: return 90 if diff <= 15: return 60 if diff <= 31: return 35 if diff <= 60: return 5 if diff <= 90: return -35 if diff <= 180: return -110 return -190 def related_labor_desc(ledger: LedgerLine, voucher) -> bool: left = ledger.norm_desc right = voucher.norm_blob labor_cost = "인건비" in right labor_work = any(token in left for token in ["근무대가", "야간근무", "휴일근무", "휴일및야간근무"]) if not labor_cost or not labor_work: return False ratio = SequenceMatcher(None, ledger.norm_desc, voucher.norm_desc).ratio() if voucher.norm_desc else 0 return ratio >= 0.45 or any(token in right for token in ["책임감리", "건설사업관리", "고속도로", "터널"]) def best_voucher_date(voucher, ledger_date: date) -> tuple[int, str]: dates = [] if voucher.draft_date: dates.append(("draft", voucher.draft_date)) confirmed_date = date_from_voucher_no(voucher.confirmed_no) if confirmed_date: dates.append(("confirmed", confirmed_date)) if not dates: return 9999, "missing" basis, best = min(dates, key=lambda item: abs((item[1] - ledger_date).days)) return abs((best - ledger_date).days), basis def strong_desc_similarity(ledger: LedgerLine, voucher) -> tuple[bool, str, int]: if not ledger.norm_desc or not voucher.norm_desc: return False, "", 0 if ledger.norm_desc == voucher.norm_desc: return True, "absolute_desc_exact", 100 if ledger.norm_desc in voucher.norm_desc or voucher.norm_desc in ledger.norm_desc: return True, "absolute_desc_contains", 85 ratio = SequenceMatcher(None, ledger.norm_desc, voucher.norm_desc).ratio() if ratio >= 0.82: return True, f"absolute_desc_sim_{ratio:.2f}", int(70 * ratio) if ledger.norm_desc in voucher.norm_blob: return True, "absolute_ledger_desc_in_voucher_context", 70 return False, "", 0 def absolute_amount_desc_context(ledger: LedgerLine, voucher) -> tuple[bool, str, int]: if ledger.amount == voucher.amount: return False, "", 0 if abs(ledger.amount) != abs(voucher.amount): return False, "", 0 if ledger.side != voucher.side: return False, "", 0 desc_ok, desc_reason, desc_bonus = strong_desc_similarity(ledger, voucher) if not desc_ok: return False, "", 0 diff, _basis = best_voucher_date(voucher, ledger.ledger_date) v_status = vendor_status(ledger, voucher) if diff > 60 and v_status != "match": return False, "", 0 if v_status == "mismatch": return False, "", 0 bonus = 35 + desc_bonus reason = f"absolute_amount_desc_similarity,{desc_reason}" return True, reason, bonus def score_pair(ledger: LedgerLine, voucher) -> tuple[int, str, int]: ok, account_reason = compatible_account(voucher, ledger) if not voucher.draft_date and not date_from_voucher_no(voucher.confirmed_no): return -999, "missing_draft_date", 9999 v_status = vendor_status(ledger, voucher) diff, date_basis = best_voucher_date(voucher, ledger.ledger_date) ledger_account_base = account_base(ledger.account_name) account_context_in_blob = bool( ledger_account_base and len(ledger_account_base) >= 2 and ledger_account_base in voucher.norm_blob ) labor_desc_related = related_labor_desc(ledger, voucher) desc_strong = bool( account_context_in_blob or labor_desc_related or ( ledger.norm_desc and ( ledger.norm_desc == voucher.norm_desc or ledger.norm_desc in voucher.norm_desc or voucher.norm_desc in ledger.norm_desc or ledger.norm_desc in voucher.norm_blob ) ) ) negative_reversal = negative_reversal_context(voucher, ledger) absolute_desc_context, absolute_desc_reason, absolute_desc_bonus = absolute_amount_desc_context(ledger, voucher) context_vendor_override = ( v_status == "mismatch" and diff <= 60 and desc_strong and (ok or negative_reversal or voucher.account_code[:1] in ("5", "6")) ) account_context_override = ( not ok and diff <= 60 and ( (v_status == "match" and desc_strong) or (negative_reversal and v_status == "match" and desc_strong) or (absolute_desc_context and v_status in {"match", "missing", "weak_mismatch"}) or (v_status == "match" and diff <= 7 and balance_sheet_context_accounts(voucher, ledger)) or context_vendor_override ) ) if v_status == "mismatch" and not context_vendor_override: return -999, "vendor_mismatch", 9999 if not ok: if not account_context_override: return -999, account_reason, 9999 account_reason = "receivable_payable_reversal_context" if negative_reversal else "context_override_account_mismatch" score = 35 + date_score(diff) reasons = [account_reason, f"{date_basis}_diff_{diff}d"] if negative_reversal: score += 90 ledger_flip = economic_side_amount(ledger.side, ledger.amount)[2] voucher_flip = economic_side_amount(voucher.side, voucher.amount)[2] if ledger_flip: reasons.append(ledger_flip) if voucher_flip: reasons.append(f"voucher_{voucher_flip}") if absolute_desc_context: score += absolute_desc_bonus reasons.append(absolute_desc_reason) confirmed_date = date_from_voucher_no(voucher.confirmed_no) if confirmed_date and abs((confirmed_date - ledger.ledger_date).days) == 0: score += 12 reasons.append("confirmed_date_exact") if ledger.norm_desc and voucher.norm_desc: if ledger.norm_desc == voucher.norm_desc: score += 100 reasons.append("desc_exact") elif ledger.norm_desc in voucher.norm_desc or voucher.norm_desc in ledger.norm_desc: score += 75 reasons.append("desc_contains") else: ratio = SequenceMatcher(None, ledger.norm_desc, voucher.norm_desc).ratio() if ratio >= 0.72: score += int(55 * ratio) reasons.append(f"desc_sim_{ratio:.2f}") if ledger.norm_desc and ledger.norm_desc in voucher.norm_blob: score += 55 reasons.append("ledger_desc_in_voucher_group_context") if voucher.norm_desc and voucher.norm_desc in ledger.norm_blob: score += 30 reasons.append("voucher_desc_in_ledger_context") if account_context_in_blob: score += 55 reasons.append("ledger_account_in_voucher_context") if labor_desc_related: score += 70 reasons.append("desc_related_labor_cost_terms") if v_status == "match": score += 95 reasons.append("vendor_strong_match") elif context_vendor_override: score += 35 reasons.append("vendor_context_override_by_desc") elif v_status == "weak_mismatch": score -= 45 reasons.append("vendor_weak_mismatch") else: score -= 10 reasons.append("vendor_missing") if diff > 90 and v_status != "match": score -= 100 reasons.append("long_date_gap_without_vendor") if isinstance(voucher, VoucherAggregate): score += 35 reasons.append(f"group_sum_{len(voucher.line_ids)}lines") if voucher.group_id.startswith("multi:"): score += 45 reasons.append("multi_voucher_sum_by_settlement_refs") return score, ",".join(reasons), diff def load_vouchers(path: Path) -> tuple[list[VoucherLine], list[VoucherAggregate]]: workbook = load_workbook(path, read_only=True, data_only=True) sheet = workbook.worksheets[0] next(sheet.iter_rows(min_row=1, max_row=1, values_only=True)) raw_rows = [] groups = defaultdict(lambda: {"blob": [], "vendors": [], "descs": []}) settlement_groups = defaultdict(lambda: {"refs": [], "rows": [], "blob": [], "vendors": [], "descs": [], "date": None, "confirmed": []}) for row_number, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), 2): values = list(row) base = group_no(values[10]) draft_base = draft_group_no(values[1]) desc = clean(values[15]) or clean(values[16]) vendor = clean(values[18]) blob = " ".join( clean(values[index]) for index in [8, 9, 11, 12, 13, 14, 15, 16, 18, 21, 22] if index < len(values) and values[index] is not None ) raw_rows.append( { "row": row_number, "values": values, "group": base, "draft_group": draft_base, "draft_date": date_from_voucher_no(values[1]), "desc": desc, "vendor": vendor, "blob": blob, } ) groups[base]["blob"].append(blob) if vendor: groups[base]["vendors"].append(vendor) if desc: groups[base]["descs"].append(desc) settlement_refs = setting_group_refs(blob) if settlement_refs: settlement = settlement_groups[draft_base] settlement["refs"].extend(settlement_refs) settlement["rows"].append(row_number) settlement["blob"].append(blob) if vendor: settlement["vendors"].append(vendor) if desc: settlement["descs"].append(desc) if not settlement["date"]: settlement["date"] = date_from_voucher_no(values[1]) confirmed = clean(values[10]) if confirmed: settlement["confirmed"].append(confirmed) lines: list[VoucherLine] = [] for raw in raw_rows: values = raw["values"] debit = amount(values[4]) credit = amount(values[6]) if not debit and not credit: continue side, line_amount = ("debit", debit) if debit and (not credit or abs(debit) >= abs(credit)) else ("credit", credit) group_data = groups[raw["group"]] vendor = raw["vendor"] or next(iter(group_data["vendors"]), "") desc = raw["desc"] or next(iter(group_data["descs"]), "") group_blob = " ".join(group_data["blob"] + [desc, vendor]) lines.append( VoucherLine( id=len(lines), row=raw["row"], group_id=raw["group"], draft_no=clean(values[1]), confirmed_no=clean(values[10]), draft_date=raw["draft_date"], side=side, amount=line_amount, account_code=clean(values[2]).replace(".0", ""), account_name=clean(values[3]), desc=desc, vendor=vendor, group_blob=group_blob, debit_supply=debit, credit_supply=credit, tax_debit=amount(values[5]), tax_credit=amount(values[7]), norm_desc=norm(desc), norm_vendor=norm(vendor), norm_blob=norm(group_blob), ) ) aggregates: list[VoucherAggregate] = [] source_by_draft_group = defaultdict(list) for line in lines: source_by_draft_group[draft_group_no(line.draft_no)].append(line) by_group_key = defaultdict(list) for line in lines: if line.account_code.startswith("2"): continue key = (line.group_id, line.side, account_base(line.account_name), norm(line.vendor)) by_group_key[key].append(line) for (group_id, side, _account_base, _vendor), group_lines in by_group_key.items(): if len(group_lines) < 2: continue first = group_lines[0] total = sum(item.amount for item in group_lines) desc = " / ".join(dict.fromkeys(item.desc for item in group_lines if item.desc)) aggregates.append( VoucherAggregate( id=len(aggregates), group_id=group_id, row=first.row, draft_no="; ".join(item.draft_no for item in group_lines), confirmed_no="; ".join(dict.fromkeys(item.confirmed_no for item in group_lines if item.confirmed_no)), draft_date=first.draft_date, side=side, amount=total, account_code=first.account_code, account_name=first.account_name, desc=desc or first.desc, vendor=first.vendor, group_blob=" ".join(item.group_blob for item in group_lines), line_ids=[item.id for item in group_lines], norm_desc=norm(desc or first.desc), norm_vendor=norm(first.vendor), norm_blob=norm(" ".join(item.group_blob for item in group_lines)), ) ) for settlement_group, data in settlement_groups.items(): refs = [ref for ref in dict.fromkeys(data["refs"]) if ref != settlement_group] if len(refs) < 2: continue source_lines = [ line for ref in refs for line in source_by_draft_group.get(ref, []) if source_line_for_multi_sum(line) ] by_multi_key = defaultdict(list) for line in source_lines: by_multi_key[(line.side, norm(line.vendor))].append(line) for (side, _vendor), group_lines in by_multi_key.items(): if len(group_lines) < 2: continue first = group_lines[0] total = sum(item.amount for item in group_lines) if not total: continue desc = " / ".join(dict.fromkeys(item.desc for item in group_lines if item.desc)) vendor = first.vendor or next(iter(data["vendors"]), "") blob = " ".join(data["blob"] + [item.group_blob for item in group_lines]) aggregates.append( VoucherAggregate( id=len(aggregates), group_id=f"multi:{settlement_group}", row=min(item.row for item in group_lines), draft_no="; ".join(item.draft_no for item in group_lines), confirmed_no="; ".join(dict.fromkeys(data["confirmed"] + [item.confirmed_no for item in group_lines if item.confirmed_no])), draft_date=data["date"] or first.draft_date, side=side, amount=total, account_code="; ".join(dict.fromkeys(item.account_code for item in group_lines if item.account_code)), account_name="; ".join(dict.fromkeys(item.account_name for item in group_lines if item.account_name)), desc=desc or " / ".join(dict.fromkeys(data["descs"])), vendor=vendor, group_blob=blob, line_ids=[item.id for item in group_lines], norm_desc=norm(desc or " / ".join(dict.fromkeys(data["descs"]))), norm_vendor=norm(vendor), norm_blob=norm(blob), ) ) return lines, aggregates def load_ledgers(path: Path) -> list[LedgerLine]: workbook = load_workbook(path, read_only=True, data_only=True) sheet = workbook.worksheets[0] next(sheet.iter_rows(min_row=1, max_row=1, values_only=True)) ledgers: list[LedgerLine] = [] for row_number, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), 2): values = list(row) ledger_date = parse_ledger_date(values[5]) desc = clean(values[6]) debit = amount(values[8]) credit = amount(values[9]) if not ledger_date or desc.startswith("[") or (not debit and not credit): continue # Avoid carrying opening balance lines into matching without relying on a Korean literal. if not values[5] and not values[11]: continue side, line_amount = ("debit", debit) if debit and (not credit or abs(debit) >= abs(credit)) else ("credit", credit) vendor = clean(values[7]) blob = " ".join([desc, vendor, clean(values[1]), clean(values[11])]) ledgers.append( LedgerLine( id=len(ledgers), row=row_number, ledger_date=ledger_date, side=side, amount=line_amount, account_code=clean(values[0]).replace(".0", ""), account_name=clean(values[1]), desc=desc, vendor=vendor, debit=debit, credit=credit, voucher_no=clean(values[11]), source_row_no=clean(values[4]), norm_desc=norm(desc), norm_vendor=norm(vendor), norm_blob=norm(blob), ) ) return ledgers def build_candidates(ledgers: list[LedgerLine], voucher_lines: list[VoucherLine], aggregates: list[VoucherAggregate]) -> list[Candidate]: line_by_amount = defaultdict(list) line_by_abs_amount = defaultdict(list) line_by_economic_amount = defaultdict(list) aggregate_by_amount = defaultdict(list) for voucher in voucher_lines: line_by_amount[(voucher.side, voucher.amount)].append(voucher) line_by_abs_amount[(voucher.side, abs(voucher.amount))].append(voucher) economic_side, economic_amount, _reason = economic_side_amount(voucher.side, voucher.amount) line_by_economic_amount[(economic_side, economic_amount)].append(voucher) for aggregate in aggregates: aggregate_by_amount[(aggregate.side, aggregate.amount)].append(aggregate) candidates: list[Candidate] = [] for ledger in ledgers: for voucher in line_by_amount.get((ledger.side, ledger.amount), []): score, reason, diff = score_pair(ledger, voucher) if score >= 135: exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == voucher.norm_desc)) strong_vendor = int("vendor_strong_match" in reason) candidates.append( Candidate( sort_key=(score, strong_vendor, exact_desc, -diff), ledger_id=ledger.id, voucher_kind="line", voucher_id=voucher.id, score=score, reason=reason, date_diff=diff, ) ) for voucher in line_by_abs_amount.get((ledger.side, abs(ledger.amount)), []): if voucher.amount == ledger.amount: continue score, reason, diff = score_pair(ledger, voucher) if "absolute_amount_desc_similarity" not in reason: continue if score >= 150: exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == voucher.norm_desc)) strong_vendor = int("vendor_strong_match" in reason) candidates.append( Candidate( sort_key=(score, strong_vendor, exact_desc, -diff), ledger_id=ledger.id, voucher_kind="absline", voucher_id=voucher.id, score=score, reason=reason, date_diff=diff, ) ) ledger_economic_side, ledger_economic_amount, ledger_economic_reason = economic_side_amount(ledger.side, ledger.amount) if ledger_economic_reason: for voucher in line_by_economic_amount.get((ledger_economic_side, ledger_economic_amount), []): if (voucher.side, voucher.amount) == (ledger.side, ledger.amount): continue score, reason, diff = score_pair(ledger, voucher) if "receivable_payable_reversal_context" not in reason: continue if score >= 155: exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == voucher.norm_desc)) strong_vendor = int("vendor_strong_match" in reason) candidates.append( Candidate( sort_key=(score, strong_vendor, exact_desc, -diff), ledger_id=ledger.id, voucher_kind="line", voucher_id=voucher.id, score=score, reason=reason, date_diff=diff, ) ) for aggregate in aggregate_by_amount.get((ledger.side, ledger.amount), []): score, reason, diff = score_pair(ledger, aggregate) if score >= 155: exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == aggregate.norm_desc)) strong_vendor = int("vendor_strong_match" in reason) kind = "multi" if aggregate.group_id.startswith("multi:") else "aggregate" candidates.append( Candidate( sort_key=(score, strong_vendor, exact_desc, -diff), ledger_id=ledger.id, voucher_kind=kind, voucher_id=aggregate.id, score=score, reason=reason, date_diff=diff, ) ) candidates.sort(reverse=True) return candidates def assign_matches(ledgers: list[LedgerLine], voucher_lines: list[VoucherLine], aggregates: list[VoucherAggregate], candidates: list[Candidate]) -> None: used_ledgers: set[int] = set() used_voucher_lines: set[int] = set() ordered_candidates = ( [item for item in candidates if item.voucher_kind not in {"multi", "absline"}] + [item for item in candidates if item.voucher_kind == "multi"] + [item for item in candidates if item.voucher_kind == "absline"] ) for candidate in ordered_candidates: if candidate.ledger_id in used_ledgers: continue if candidate.voucher_kind in {"line", "absline"}: voucher = voucher_lines[candidate.voucher_id] line_ids = [voucher.id] else: voucher = aggregates[candidate.voucher_id] line_ids = voucher.line_ids if any(line_id in used_voucher_lines for line_id in line_ids): continue ledger = ledgers[candidate.ledger_id] ledger.match_kind = candidate.voucher_kind ledger.match_id = candidate.voucher_id ledger.score = candidate.score ledger.reason = candidate.reason ledger.date_diff = candidate.date_diff if candidate.voucher_kind in {"line", "absline"}: voucher.match_id = ledger.id voucher.score = candidate.score voucher.reason = candidate.reason voucher.date_diff = candidate.date_diff else: voucher.match_id = ledger.id voucher.score = candidate.score voucher.reason = candidate.reason voucher.date_diff = candidate.date_diff for line_id in voucher.line_ids: line = voucher_lines[line_id] line.match_id = ledger.id line.score = candidate.score line.reason = candidate.reason line.date_diff = candidate.date_diff used_ledgers.add(ledger.id) used_voucher_lines.update(line_ids) def compact_range(values: Iterable[str]) -> str: numbers = [] text_values = [] for value in values: text = clean(value) if not text: continue try: numbers.append(int(float(text))) except ValueError: text_values.append(text) if numbers and not text_values: numbers = sorted(dict.fromkeys(numbers)) if len(numbers) > 1 and numbers == list(range(numbers[0], numbers[-1] + 1)): return f"{numbers[0]}-{numbers[-1]}" return "; ".join(str(item) for item in numbers) return "; ".join(dict.fromkeys(text_values + [str(item) for item in numbers])) def ledger_group_key(ledger: LedgerLine) -> tuple: return ( ledger.side, ledger.ledger_date, ledger.account_code, account_base(ledger.account_name), ledger_vendor_key(ledger.vendor), ledger.norm_desc, ) def make_ledger_group_probe(group_lines: list[LedgerLine], total: int) -> LedgerLine: first = group_lines[0] voucher_numbers = "; ".join(item.voucher_no for item in group_lines if item.voucher_no) source_rows = compact_range(item.source_row_no for item in group_lines) blob = " ".join([first.desc, first.vendor, first.account_name, voucher_numbers, source_rows]) return LedgerLine( id=-1, row=min(item.row for item in group_lines), ledger_date=first.ledger_date, side=first.side, amount=total, account_code=first.account_code, account_name=first.account_name, desc=first.desc, vendor=first.vendor, debit=total if first.side == "debit" else 0, credit=total if first.side == "credit" else 0, voucher_no=voucher_numbers, source_row_no=source_rows, norm_desc=first.norm_desc, norm_vendor=first.norm_vendor, norm_blob=norm(blob), ) def assign_ledger_group_matches(ledgers: list[LedgerLine], voucher_lines: list[VoucherLine]) -> int: unmatched_groups = defaultdict(list) for ledger in ledgers: if ledger.match_id is None: unmatched_groups[ledger_group_key(ledger)].append(ledger) available_vouchers = defaultdict(list) for voucher in voucher_lines: if voucher.match_id is None: available_vouchers[(voucher.side, voucher.amount)].append(voucher) group_candidates = [] for group_lines in unmatched_groups.values(): if len(group_lines) < 2: continue total = sum(item.amount for item in group_lines) if not total: continue probe = make_ledger_group_probe(group_lines, total) for voucher in available_vouchers.get((probe.side, probe.amount), []): score, reason, diff = score_pair(probe, voucher) if score < 185: continue reason = ",".join( [ reason, f"ledger_group_sum_to_single_voucher", f"group_sum_{len(group_lines)}ledger_lines", f"group_source_rows_{probe.source_row_no}", f"group_vouchers_{probe.voucher_no}", ] ) exact_desc = int(bool(probe.norm_desc and probe.norm_desc == voucher.norm_desc)) strong_vendor = int("vendor_strong_match" in reason) group_candidates.append((score, strong_vendor, exact_desc, -diff, group_lines, voucher, reason, diff)) matched_count = 0 used_ledgers = {ledger.id for ledger in ledgers if ledger.match_id is not None} used_vouchers = {voucher.id for voucher in voucher_lines if voucher.match_id is not None} for score, _strong_vendor, _exact_desc, _neg_diff, group_lines, voucher, reason, diff in sorted( group_candidates, key=lambda item: item[:4], reverse=True ): if voucher.id in used_vouchers: continue if any(ledger.id in used_ledgers for ledger in group_lines): continue ledger_ids = [ledger.id for ledger in group_lines] for ledger in group_lines: ledger.match_kind = "ledger_group" ledger.match_id = voucher.id ledger.score = score ledger.reason = reason ledger.date_diff = diff used_ledgers.add(ledger.id) matched_count += 1 voucher.match_id = group_lines[0].id voucher.match_ledger_ids = ledger_ids voucher.score = score voucher.reason = reason voucher.date_diff = diff used_vouchers.add(voucher.id) return matched_count def review(diff: int | None) -> tuple[str, str]: if diff is None: return "", "" if diff > 90: return "REVIEW_LONG_DATE_GAP", f"draft-ledger date diff {diff}d" if diff >= 31: return "REVIEW_DATE_GAP", f"draft-ledger date diff {diff}d" return "", "" def style_like(target, source) -> None: if not source.has_style: return target.font = copy(source.font) target.fill = copy(source.fill) target.border = copy(source.border) target.alignment = copy(source.alignment) target.number_format = source.number_format target.protection = copy(source.protection) def write_ledger_output( source_path: Path, output_path: Path, ledgers: list[LedgerLine], voucher_lines: list[VoucherLine], aggregates: list[VoucherAggregate], ) -> None: ledger_by_row = {item.row: item for item in ledgers} workbook = load_workbook(source_path) sheet = workbook.worksheets[0] for offset, header in enumerate(LEDGER_HEADERS): cell = sheet.cell(row=1, column=15 + offset, value=header) style_like(cell, sheet.cell(row=1, column=14)) sheet.column_dimensions[cell.column_letter].width = 22 if offset != 8 else 60 for row_number in range(2, sheet.max_row + 1): ledger = ledger_by_row.get(row_number) values = [None] * len(LEDGER_HEADERS) if ledger and ledger.match_id is not None: if ledger.match_kind in ("line", "absline", "ledger_group"): match = voucher_lines[ledger.match_id] debit_value = match.debit_supply credit_value = match.credit_supply else: match = aggregates[ledger.match_id] debit_value = match.amount if match.side == "debit" else 0 credit_value = match.amount if match.side == "credit" else 0 flag, memo = review(ledger.date_diff) values = [ match.draft_no, match.account_code, match.account_name, debit_value, credit_value, match.confirmed_no, mmdd(match.draft_date), ledger.date_diff, ledger.reason, flag, memo, ] for offset, value in enumerate(values): sheet.cell(row=row_number, column=15 + offset, value=value) workbook.save(output_path) def write_voucher_output(source_path: Path, output_path: Path, ledgers: list[LedgerLine], voucher_lines: list[VoucherLine]) -> None: ledger_by_id = {item.id: item for item in ledgers} voucher_by_row = {item.row: item for item in voucher_lines} workbook = load_workbook(source_path) sheet = workbook.worksheets[0] for offset, header in enumerate(VOUCHER_HEADERS): cell = sheet.cell(row=1, column=24 + offset, value=header) style_like(cell, sheet.cell(row=1, column=23)) sheet.column_dimensions[cell.column_letter].width = 18 if offset != 7 else 60 for row_number in range(2, sheet.max_row + 1): voucher = voucher_by_row.get(row_number) values = [None] * len(VOUCHER_HEADERS) if voucher and voucher.match_id is not None: matched_ledgers = [ledger_by_id[item] for item in voucher.match_ledger_ids] if voucher.match_ledger_ids else [ledger_by_id[voucher.match_id]] first = matched_ledgers[0] flag, memo = review(voucher.date_diff) values = [ "; ".join(dict.fromkeys(item.account_code for item in matched_ledgers)), "; ".join(dict.fromkeys(item.account_name for item in matched_ledgers)), mmdd(first.ledger_date), sum(item.debit for item in matched_ledgers) or None, sum(item.credit for item in matched_ledgers) or None, "; ".join(item.voucher_no for item in matched_ledgers if item.voucher_no), voucher.date_diff, voucher.reason, flag, memo, ] for offset, value in enumerate(values): sheet.cell(row=row_number, column=24 + offset, value=value) workbook.save(output_path) def run(ledger_path: Path, voucher_path: Path, output_dir: Path) -> tuple[Path, Path, dict]: output_dir.mkdir(parents=True, exist_ok=True) ledger_out = output_dir / LEDGER_OUT_NAME voucher_out = output_dir / VOUCHER_OUT_NAME voucher_lines, aggregates = load_vouchers(voucher_path) ledgers = load_ledgers(ledger_path) candidates = build_candidates(ledgers, voucher_lines, aggregates) assign_matches(ledgers, voucher_lines, aggregates, candidates) ledger_group_matches = assign_ledger_group_matches(ledgers, voucher_lines) write_ledger_output(ledger_path, ledger_out, ledgers, voucher_lines, aggregates) write_voucher_output(voucher_path, voucher_out, ledgers, voucher_lines) stats = { "ledger_rows": len(ledgers), "voucher_lines": len(voucher_lines), "voucher_aggregates": len(aggregates), "candidates": len(candidates), "ledger_group_matches": ledger_group_matches, "matched_ledgers": sum(1 for item in ledgers if item.match_id is not None), "matched_voucher_lines": sum(1 for item in voucher_lines if item.match_id is not None), } return ledger_out, voucher_out, stats def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Match WEHAGO ledger rows against SATIS voucher rows.") parser.add_argument("--ledger", type=Path, default=DEFAULT_LEDGER, help="Path to ledger xlsx.") parser.add_argument("--voucher", type=Path, default=DEFAULT_VOUCHER, help="Path to voucher xlsx.") parser.add_argument("--output-dir", type=Path, default=DEFAULT_DIR, help="Directory for output xlsx files.") return parser.parse_args() def main() -> None: args = parse_args() ledger_out, voucher_out, stats = run(args.ledger, args.voucher, args.output_dir) print(f"ledger_out={ledger_out}") print(f"voucher_out={voucher_out}") for key, value in stats.items(): print(f"{key}={value}") if __name__ == "__main__": main()