1245 lines
45 KiB
Python
Executable File
1245 lines
45 KiB
Python
Executable File
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()
|