317 lines
13 KiB
Python
317 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from collections import Counter, defaultdict, deque
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from openpyxl import Workbook
|
|
from openpyxl.cell import WriteOnlyCell
|
|
from openpyxl.styles import Alignment, Font, PatternFill
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
if str(BASE_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BASE_DIR))
|
|
|
|
from main import engine
|
|
from wehago_compare import _get_cached_status_rows_by_range, _get_cached_voucher_sections_by_range
|
|
|
|
EXPORT_DIR = BASE_DIR / "static" / "exports"
|
|
HEADER_FILL = PatternFill(fill_type="solid", fgColor="111827")
|
|
HEADER_FONT = Font(color="FFFFFF", bold=True)
|
|
POS_FILL = PatternFill(fill_type="solid", fgColor="DCFCE7")
|
|
NEG_FILL = PatternFill(fill_type="solid", fgColor="DBEAFE")
|
|
|
|
|
|
def clean(value: Any) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
def amount(value: Any) -> float:
|
|
text = clean(value).replace(",", "")
|
|
if not text:
|
|
return 0.0
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def write_row(ws, values: list[Any], fill: PatternFill | None = None, header: bool = False) -> None:
|
|
row = []
|
|
for value in values:
|
|
cell = WriteOnlyCell(ws, value=value)
|
|
cell.alignment = Alignment(vertical="top", wrap_text=True)
|
|
if header:
|
|
cell.fill = HEADER_FILL
|
|
cell.font = HEADER_FONT
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
elif fill:
|
|
cell.fill = fill
|
|
row.append(cell)
|
|
ws.append(row)
|
|
|
|
|
|
def summarize_group(summary: dict[str, Any], side: str) -> dict[str, Any]:
|
|
if side == "w":
|
|
return {
|
|
"date": clean(summary.get("ledger_date")),
|
|
"voucher_no": clean(summary.get("voucher_no")),
|
|
"draft_no": clean(summary.get("draft_no")),
|
|
"accounts": clean(summary.get("ledger_accounts")),
|
|
"vendors": clean(summary.get("ledger_vendors")),
|
|
"debit": amount(summary.get("ledger_debit")),
|
|
"credit": amount(summary.get("ledger_credit")),
|
|
"links": len(summary.get("erp_group_keys") or []),
|
|
"reason": clean(summary.get("review_reason")),
|
|
}
|
|
return {
|
|
"date": clean(summary.get("proof_date")) or clean(summary.get("ledger_date")),
|
|
"voucher_no": clean(summary.get("voucher_no")),
|
|
"draft_no": clean(summary.get("draft_no")),
|
|
"accounts": clean(summary.get("voucher_accounts")),
|
|
"vendors": clean(summary.get("voucher_vendors")),
|
|
"debit": amount(summary.get("voucher_debit")),
|
|
"credit": amount(summary.get("voucher_credit")),
|
|
"links": len(summary.get("wehago_group_keys") or []),
|
|
"reason": clean(summary.get("review_reason")),
|
|
}
|
|
|
|
|
|
def build_gap_report(year: int) -> Path:
|
|
rows = _get_cached_status_rows_by_range(engine, year, year)
|
|
sections = _get_cached_voucher_sections_by_range(engine, year, year, rows)
|
|
wehago_groups = sections["voucher_matched"]
|
|
voucher_groups = sections["erp_voucher_matched"]
|
|
|
|
wehago_by_key: dict[Any, dict[str, Any]] = {}
|
|
voucher_by_key: dict[Any, dict[str, Any]] = {}
|
|
adj: dict[tuple[str, Any], set[tuple[str, Any]]] = defaultdict(set)
|
|
|
|
for group in wehago_groups:
|
|
summary = group["summary"]
|
|
w_key = summary.get("wehago_group_key")
|
|
wehago_by_key[w_key] = group
|
|
for e_key in summary.get("erp_group_keys") or set():
|
|
adj[("w", w_key)].add(("e", e_key))
|
|
adj[("e", e_key)].add(("w", w_key))
|
|
for group in voucher_groups:
|
|
summary = group["summary"]
|
|
e_key = summary.get("erp_group_key")
|
|
voucher_by_key[e_key] = group
|
|
for w_key in summary.get("wehago_group_keys") or set():
|
|
adj[("e", e_key)].add(("w", w_key))
|
|
adj[("w", w_key)].add(("e", e_key))
|
|
|
|
components: list[dict[str, Any]] = []
|
|
visited: set[tuple[str, Any]] = set()
|
|
for node in list(adj):
|
|
if node in visited:
|
|
continue
|
|
q = deque([node])
|
|
visited.add(node)
|
|
w_keys: list[Any] = []
|
|
e_keys: list[Any] = []
|
|
while q:
|
|
side, key = q.popleft()
|
|
if side == "w":
|
|
w_keys.append(key)
|
|
else:
|
|
e_keys.append(key)
|
|
for nxt in adj[node if False else (side, key)]:
|
|
if nxt not in visited:
|
|
visited.add(nxt)
|
|
q.append(nxt)
|
|
components.append({"w_keys": w_keys, "e_keys": e_keys})
|
|
|
|
shape_counter = Counter((len(item["w_keys"]), len(item["e_keys"])) for item in components)
|
|
gap_components = [item for item in components if len(item["w_keys"]) != len(item["e_keys"])]
|
|
gap_components.sort(key=lambda item: (abs(len(item["w_keys"]) - len(item["e_keys"])) * -1, len(item["w_keys"]) * -1, len(item["e_keys"]) * -1))
|
|
|
|
EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
|
wb = Workbook(write_only=True)
|
|
|
|
ws_summary = wb.create_sheet("summary")
|
|
write_row(ws_summary, ["항목", "값", "설명"], header=True)
|
|
write_row(ws_summary, ["연도", year, "분석 기준 연도"])
|
|
write_row(ws_summary, ["WEHAGO 매치 그룹 수", len(wehago_groups), "WEHAGO 카드에 표시되는 매치 전표 그룹 수"])
|
|
write_row(ws_summary, ["Voucher 매치 그룹 수", len(voucher_groups), "Voucher 카드에 표시되는 매치 전표 그룹 수"])
|
|
write_row(ws_summary, ["그룹 수 차이", len(wehago_groups) - len(voucher_groups), "WEHAGO 그룹 수 - Voucher 그룹 수"])
|
|
write_row(ws_summary, ["전체 연결 성분 수", len(components), "WEHAGO/ERP 매치 관계를 그래프로 묶은 연결 단위 수"])
|
|
write_row(ws_summary, ["1:1 성분 수", shape_counter.get((1, 1), 0), "양쪽 그룹 수가 같은 일반적인 경우"])
|
|
write_row(ws_summary, ["차이 발생 성분 수", len(gap_components), "WEHAGO 그룹 수와 Voucher 그룹 수가 다른 연결 성분 수"])
|
|
write_row(ws_summary, ["WEHAGO 쪽 초과 합계", sum(max(len(item['w_keys']) - len(item['e_keys']), 0) for item in gap_components), "WEHAGO 그룹 수가 더 많은 만큼의 합계"])
|
|
write_row(ws_summary, ["Voucher 쪽 초과 합계", sum(max(len(item['e_keys']) - len(item['w_keys']), 0) for item in gap_components), "Voucher 그룹 수가 더 많은 만큼의 합계"])
|
|
write_row(ws_summary, [])
|
|
write_row(ws_summary, ["형태", "건수", "설명"], header=True)
|
|
for (w_count, e_count), count in shape_counter.most_common(20):
|
|
desc = "정상 1:1" if (w_count, e_count) == (1, 1) else ("WEHAGO가 더 많이 쪼개짐" if w_count > e_count else "Voucher가 더 많이 쪼개짐")
|
|
write_row(ws_summary, [f"{w_count}:{e_count}", count, desc])
|
|
|
|
ws_gap = wb.create_sheet("gap_components")
|
|
write_row(
|
|
ws_gap,
|
|
[
|
|
"component_id",
|
|
"형태",
|
|
"WEHAGO 그룹 수",
|
|
"Voucher 그룹 수",
|
|
"차이(WEHAGO-Voucher)",
|
|
"WEHAGO 대표 전표",
|
|
"Voucher 대표 전표",
|
|
"WEHAGO 전표 목록",
|
|
"Voucher 전표 목록",
|
|
"WEHAGO 계정 요약",
|
|
"Voucher 계정 요약",
|
|
],
|
|
header=True,
|
|
)
|
|
for idx, item in enumerate(gap_components, start=1):
|
|
w_summaries = [summarize_group(wehago_by_key[key]["summary"], "w") for key in item["w_keys"] if key in wehago_by_key]
|
|
e_summaries = [summarize_group(voucher_by_key[key]["summary"], "e") for key in item["e_keys"] if key in voucher_by_key]
|
|
fill = POS_FILL if len(item["w_keys"]) > len(item["e_keys"]) else NEG_FILL
|
|
write_row(
|
|
ws_gap,
|
|
[
|
|
idx,
|
|
f"{len(item['w_keys'])}:{len(item['e_keys'])}",
|
|
len(item["w_keys"]),
|
|
len(item["e_keys"]),
|
|
len(item["w_keys"]) - len(item["e_keys"]),
|
|
" | ".join(f"{row['date']} {row['voucher_no']}" for row in w_summaries[:3]),
|
|
" | ".join(f"{row['date']} {row['voucher_no']}/{row['draft_no']}" for row in e_summaries[:3]),
|
|
"\n".join(f"{row['date']} / {row['voucher_no']} / {row['accounts']}" for row in w_summaries),
|
|
"\n".join(f"{row['date']} / {row['voucher_no']} / {row['draft_no']} / {row['accounts']}" for row in e_summaries),
|
|
"\n".join(row["accounts"] for row in w_summaries),
|
|
"\n".join(row["accounts"] for row in e_summaries),
|
|
],
|
|
fill=fill,
|
|
)
|
|
|
|
ws_wehago = wb.create_sheet("wehago_extra_groups")
|
|
write_row(
|
|
ws_wehago,
|
|
[
|
|
"component_id",
|
|
"형태",
|
|
"WEHAGO 일자",
|
|
"WEHAGO 전표번호",
|
|
"WEHAGO 계정",
|
|
"WEHAGO 거래처",
|
|
"WEHAGO 차변",
|
|
"WEHAGO 대변",
|
|
"연결 ERP 그룹 수",
|
|
"연결 ERP 전표 목록",
|
|
"검증 근거",
|
|
],
|
|
header=True,
|
|
)
|
|
for idx, item in enumerate(gap_components, start=1):
|
|
if len(item["w_keys"]) <= len(item["e_keys"]):
|
|
continue
|
|
linked_erp = [summarize_group(voucher_by_key[key]["summary"], "e") for key in item["e_keys"] if key in voucher_by_key]
|
|
linked_erp_text = "\n".join(f"{row['date']} / {row['voucher_no']} / {row['draft_no']}" for row in linked_erp)
|
|
for w_key in item["w_keys"]:
|
|
group = wehago_by_key.get(w_key)
|
|
if not group:
|
|
continue
|
|
s = summarize_group(group["summary"], "w")
|
|
write_row(
|
|
ws_wehago,
|
|
[
|
|
idx,
|
|
f"{len(item['w_keys'])}:{len(item['e_keys'])}",
|
|
s["date"],
|
|
s["voucher_no"],
|
|
s["accounts"],
|
|
s["vendors"],
|
|
s["debit"],
|
|
s["credit"],
|
|
s["links"],
|
|
linked_erp_text,
|
|
s["reason"],
|
|
],
|
|
fill=POS_FILL,
|
|
)
|
|
|
|
ws_voucher = wb.create_sheet("voucher_extra_groups")
|
|
write_row(
|
|
ws_voucher,
|
|
[
|
|
"component_id",
|
|
"형태",
|
|
"ERP 일자",
|
|
"ERP 전표번호",
|
|
"ERP 가전표번호",
|
|
"ERP 계정",
|
|
"ERP 거래처",
|
|
"ERP 차변",
|
|
"ERP 대변",
|
|
"연결 WEHAGO 그룹 수",
|
|
"연결 WEHAGO 전표 목록",
|
|
"검증 근거",
|
|
],
|
|
header=True,
|
|
)
|
|
for idx, item in enumerate(gap_components, start=1):
|
|
if len(item["e_keys"]) <= len(item["w_keys"]):
|
|
continue
|
|
linked_wehago = [summarize_group(wehago_by_key[key]["summary"], "w") for key in item["w_keys"] if key in wehago_by_key]
|
|
linked_wehago_text = "\n".join(f"{row['date']} / {row['voucher_no']}" for row in linked_wehago)
|
|
for e_key in item["e_keys"]:
|
|
group = voucher_by_key.get(e_key)
|
|
if not group:
|
|
continue
|
|
s = summarize_group(group["summary"], "e")
|
|
write_row(
|
|
ws_voucher,
|
|
[
|
|
idx,
|
|
f"{len(item['w_keys'])}:{len(item['e_keys'])}",
|
|
s["date"],
|
|
s["voucher_no"],
|
|
s["draft_no"],
|
|
s["accounts"],
|
|
s["vendors"],
|
|
s["debit"],
|
|
s["credit"],
|
|
s["links"],
|
|
linked_wehago_text,
|
|
s["reason"],
|
|
],
|
|
fill=NEG_FILL,
|
|
)
|
|
|
|
ws_examples = wb.create_sheet("top_patterns")
|
|
write_row(ws_examples, ["형태", "건수", "설명"], header=True)
|
|
for (w_count, e_count), count in shape_counter.most_common(50):
|
|
if (w_count, e_count) == (1, 1):
|
|
continue
|
|
write_row(
|
|
ws_examples,
|
|
[
|
|
f"{w_count}:{e_count}",
|
|
count,
|
|
"WEHAGO 전표 그룹 수가 더 많음" if w_count > e_count else "Voucher 전표 그룹 수가 더 많음",
|
|
],
|
|
fill=POS_FILL if w_count > e_count else NEG_FILL,
|
|
)
|
|
|
|
output = EXPORT_DIR / f"voucher_count_gap_{year}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
wb.save(output)
|
|
return output
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="WEHAGO/Voucher 매치 그룹 수 차이 분석 엑셀 생성")
|
|
parser.add_argument("--year", type=int, default=2025)
|
|
args = parser.parse_args()
|
|
output = build_gap_report(args.year)
|
|
print(output)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|