277 lines
10 KiB
Python
277 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Shadow-test month-token allocation for bundled ERP drafts.
|
|
|
|
This script does not modify projection tables. It focuses on cases where one
|
|
ERP draft contains several monthly rows and WEHAGO vouchers should be allocated
|
|
to the matching month row, not merely the closest amount row.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import test_wehago_row_allocator_shadow as base
|
|
|
|
DB = Path("/home/b17301/intranet-runtime/db/data.db")
|
|
YEAR = 2025
|
|
|
|
|
|
def month_tokens(*parts: Any) -> set[int]:
|
|
text = " ".join(base.clean(p) for p in parts)
|
|
found: set[int] = set()
|
|
for m in re.finditer(r"(?<!\d)(1[0-2]|0?[1-9])\s*월", text):
|
|
found.add(int(m.group(1)))
|
|
for m in re.finditer(r"(?<!\d)2025[-./](1[0-2]|0[1-9])(?:[-./]\d{1,2})?", text):
|
|
found.add(int(m.group(1)))
|
|
return found
|
|
|
|
|
|
def key_month(key: tuple[int, str, str]) -> int:
|
|
return int(key[1][:2])
|
|
|
|
|
|
def expanded_account_family(name: str) -> str:
|
|
text = re.sub(r"\s+", "", base.clean(name))
|
|
fam = base.account_family(text)
|
|
if text in {"외주비"} or "기술협력비" in text or "설계외주비" in text:
|
|
return "expense:외주비"
|
|
if "수도광열비" in text or "전력비" in text or "전기요금" in text:
|
|
return "expense:수도광열비"
|
|
if "연구개발비" in text:
|
|
return "expense:연구개발비"
|
|
return fam
|
|
|
|
|
|
def expanded_compatible(left: str, right: str) -> int:
|
|
lf = expanded_account_family(left)
|
|
rf = expanded_account_family(right)
|
|
if lf and rf and lf == rf:
|
|
return 30
|
|
score = base.compatible_account(left, right)
|
|
if score:
|
|
return score
|
|
if lf.startswith("expense:") and rf.startswith("expense:"):
|
|
return 10
|
|
return 0
|
|
|
|
|
|
def month_aware_row_score(
|
|
key: tuple[int, str, str],
|
|
left: dict[str, Any],
|
|
right: dict[str, Any],
|
|
monthly_bundle: bool,
|
|
) -> int:
|
|
score = base.amount_score(left, right)
|
|
if score < 0:
|
|
return score
|
|
acct = expanded_compatible(
|
|
base.clean(left.get("ledger_account_name")),
|
|
base.clean(right.get("voucher_account_name")),
|
|
)
|
|
if acct <= 0:
|
|
return -20
|
|
score += acct
|
|
|
|
l_months = month_tokens(left.get("ledger_desc"), left.get("ledger_date"))
|
|
r_months = month_tokens(right.get("voucher_desc"), right.get("proof_date"))
|
|
if not l_months:
|
|
l_months = {key_month(key)}
|
|
|
|
if monthly_bundle and r_months:
|
|
if l_months & r_months:
|
|
score += 55
|
|
else:
|
|
# In monthly ERP bundles, wrong-month rows must lose even when amount
|
|
# and account are close. This prevents 03월 WEHAGO rows attaching to
|
|
# 05월 ERP rows merely because the amount is nearby.
|
|
score -= 80
|
|
|
|
lv = base.vendor_key(base.clean(left.get("ledger_vendor")))
|
|
rv = base.vendor_key(base.clean(right.get("voucher_vendor")))
|
|
if lv and rv and (lv in rv or rv in lv):
|
|
score += 18
|
|
elif lv and rv:
|
|
score -= 8
|
|
|
|
lt = base.tokens(base.clean(left.get("ledger_desc")))
|
|
rt = base.tokens(base.clean(right.get("voucher_desc")))
|
|
overlap = len(lt & rt)
|
|
if overlap:
|
|
score += min(14, overlap * 3)
|
|
|
|
proof = base.mmdd(base.clean(right.get("proof_date")))
|
|
if proof and proof == f"{YEAR}-{base.mmdd(left.get('ledger_date'))}":
|
|
score += 16
|
|
return score
|
|
|
|
|
|
def is_monthly_bundle(erows: list[dict[str, Any]]) -> bool:
|
|
months: set[int] = set()
|
|
for row in erows:
|
|
months.update(month_tokens(row.get("voucher_desc"), row.get("proof_date")))
|
|
return len(months) >= 2
|
|
|
|
|
|
def allocate_month_aware(
|
|
key: tuple[int, str, str],
|
|
left_rows: list[dict[str, Any]],
|
|
right_rows: list[dict[str, Any]],
|
|
) -> tuple[list[tuple[int, int, int]], list[int], list[int]]:
|
|
monthly = is_monthly_bundle(right_rows)
|
|
scored: list[tuple[int, int, int]] = []
|
|
for i, lrow in enumerate(left_rows):
|
|
for j, rrow in enumerate(right_rows):
|
|
score = month_aware_row_score(key, lrow, rrow, monthly)
|
|
if score >= 55:
|
|
scored.append((score, i, j))
|
|
scored.sort(reverse=True)
|
|
used_l: set[int] = set()
|
|
used_r: set[int] = set()
|
|
pairs: list[tuple[int, int, int]] = []
|
|
for score, i, j in scored:
|
|
if i in used_l or j in used_r:
|
|
continue
|
|
used_l.add(i)
|
|
used_r.add(j)
|
|
pairs.append((i, j, score))
|
|
return pairs, [i for i in range(len(left_rows)) if i not in used_l], [j for j in range(len(right_rows)) if j not in used_r]
|
|
|
|
|
|
def candidate_bases_month_aware(
|
|
key: tuple[int, str, str],
|
|
left_rows: list[dict[str, Any]],
|
|
erp_by_base: dict[str, list[dict[str, Any]]],
|
|
amount_index: dict[tuple[str, int], set[str]],
|
|
) -> list[tuple[str, int, int, int, int]]:
|
|
bases: Counter[str] = Counter()
|
|
for lrow in left_rows:
|
|
for side, amt in base.signed_amounts(lrow, "wehago"):
|
|
if abs(amt) >= 0.5:
|
|
bases.update(amount_index.get((side, round(amt)), set()))
|
|
ranked: list[tuple[str, int, int, int, int]] = []
|
|
for draft_base, hits in bases.items():
|
|
if hits < 1:
|
|
continue
|
|
pairs, _, _ = allocate_month_aware(key, left_rows, erp_by_base[draft_base])
|
|
if not pairs:
|
|
continue
|
|
total = sum(score for _, _, score in pairs)
|
|
erows = erp_by_base[draft_base]
|
|
monthly = is_monthly_bundle(erows)
|
|
month_hit = 0
|
|
if monthly:
|
|
want = {key_month(key)}
|
|
for _, j, _ in pairs:
|
|
if want & month_tokens(erows[j].get("voucher_desc"), erows[j].get("proof_date")):
|
|
month_hit += 1
|
|
same_year = 1 if draft_base.startswith(f"11-{key[0]}") else 0
|
|
ranked.append((draft_base, len(pairs), total, month_hit, same_year))
|
|
ranked.sort(key=lambda x: (x[3], x[4], x[1], x[2]), reverse=True)
|
|
return ranked[:8]
|
|
|
|
|
|
def row_summary(left_rows: list[dict[str, Any]], erows: list[dict[str, Any]], pairs: list[tuple[int, int, int]]) -> list[dict[str, Any]]:
|
|
out = []
|
|
for i, j, score in sorted(pairs, key=lambda x: (x[0], x[1])):
|
|
l = left_rows[i]
|
|
r = erows[j]
|
|
out.append(
|
|
{
|
|
"wehago_account": l.get("ledger_account_name"),
|
|
"wehago_amount": base.money(l.get("ledger_debit")) or base.money(l.get("ledger_credit")),
|
|
"wehago_desc": l.get("ledger_desc"),
|
|
"erp_row": r.get("draft_no"),
|
|
"erp_account": r.get("voucher_account_name"),
|
|
"erp_amount": base.money(r.get("voucher_debit")) or base.money(r.get("voucher_credit")),
|
|
"erp_desc": r.get("voucher_desc"),
|
|
"score": score,
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def refined_ok(
|
|
key: tuple[int, str, str],
|
|
pairs: list[tuple[int, int, int]],
|
|
left_unmatched: list[int],
|
|
left_rows: list[dict[str, Any]],
|
|
erows: list[dict[str, Any]],
|
|
) -> tuple[bool, str]:
|
|
if base.has_vat(left_rows) and not base.has_exact_proof_date(key, erows):
|
|
return False, "부가세 전표 proof_date 불일치 또는 부재"
|
|
residual = {expanded_account_family(left_rows[i].get("ledger_account_name", "")) for i in left_unmatched}
|
|
settlement_only = bool(left_unmatched) and residual <= {"payable_ap", "payable_accrued", "cash", "receivable"}
|
|
if len(pairs) >= max(1, min(2, len(left_rows))):
|
|
return True, "월 토큰 기반 row 배정 가능"
|
|
if pairs and settlement_only:
|
|
return True, "핵심 row + 결제/상대계정 잔여"
|
|
return False, "월 토큰 기준 매칭 row 부족"
|
|
|
|
|
|
def main() -> None:
|
|
conn = sqlite3.connect(DB)
|
|
conn.row_factory = sqlite3.Row
|
|
groups = base.load_groups(conn)
|
|
raw_wehago = base.load_raw_wehago(conn)
|
|
erp_by_base = base.load_erp(conn)
|
|
amount_index = base.build_amount_index(erp_by_base)
|
|
|
|
keys = [
|
|
(2025, "03-05", "00001"),
|
|
(2025, "04-07", "00001"),
|
|
(2025, "05-07", "00001"),
|
|
(2025, "06-05", "00002"),
|
|
]
|
|
target_base = "11-20250530-B0100-39"
|
|
report: dict[str, Any] = {
|
|
"current_counts": dict(sorted(Counter(g.status for g in groups.values()).items())),
|
|
"target_base": target_base,
|
|
"target_base_monthly_bundle": is_monthly_bundle(erp_by_base[target_base]),
|
|
"cases": {},
|
|
}
|
|
|
|
promotions = []
|
|
detach_reviews = []
|
|
for key in keys:
|
|
group = groups.get(key)
|
|
left = raw_wehago.get(key, [])
|
|
erows = erp_by_base[target_base]
|
|
pairs, left_unmatched, right_unmatched = allocate_month_aware(key, left, erows)
|
|
ok, reason = refined_ok(key, pairs, left_unmatched, left, erows)
|
|
ranked = candidate_bases_month_aware(key, left, erp_by_base, amount_index)
|
|
status = group.status if group else "missing"
|
|
current_bases = sorted(group.draft_bases) if group else []
|
|
if ok and status in {"voucher_unmatched", "voucher_recheck"}:
|
|
promotions.append(key)
|
|
if target_base in current_bases and not pairs:
|
|
detach_reviews.append(key)
|
|
report["cases"][f"{key[1]} {key[2]}"] = {
|
|
"current_status": status,
|
|
"current_bases": current_bases,
|
|
"target_pairs": row_summary(left, erows, pairs),
|
|
"left_unmatched_count": len(left_unmatched),
|
|
"right_unmatched_count": len(right_unmatched),
|
|
"decision": reason,
|
|
"promotion_candidate": ok and status in {"voucher_unmatched", "voucher_recheck"},
|
|
"top_candidates": ranked[:5],
|
|
}
|
|
|
|
shadow_counts = Counter(g.status for g in groups.values())
|
|
for key in promotions:
|
|
old = groups[key].status
|
|
shadow_counts[old] -= 1
|
|
shadow_counts["voucher_matched"] += 1
|
|
report["month_bundle_promotions"] = [f"{k[1]} {k[2]}" for k in promotions]
|
|
report["detach_reviews"] = [f"{k[1]} {k[2]}" for k in detach_reviews]
|
|
report["shadow_counts_if_month_bundle_applied"] = dict(sorted(shadow_counts.items()))
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|