438 lines
17 KiB
Python
438 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
from openpyxl import Workbook
|
|
from openpyxl.cell import WriteOnlyCell
|
|
from openpyxl.styles import Alignment, Font, PatternFill
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
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,
|
|
discover_available_years,
|
|
)
|
|
|
|
EXPORT_DIR = BASE_DIR / "static" / "exports"
|
|
HEADER_FILL = PatternFill(fill_type="solid", fgColor="1F2937")
|
|
HEADER_FONT = Font(color="FFFFFF", bold=True)
|
|
SUMMARY_FILL = PatternFill(fill_type="solid", fgColor="E5E7EB")
|
|
RECHECK_FILL = PatternFill(fill_type="solid", fgColor="FEF3C7")
|
|
UNMATCHED_FILL = PatternFill(fill_type="solid", fgColor="FDE68A")
|
|
ERP_UNMATCHED_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 autosize(ws) -> None:
|
|
widths: dict[int, int] = {}
|
|
for row in ws.iter_rows():
|
|
for cell in row:
|
|
value = "" if cell.value is None else str(cell.value)
|
|
widths[cell.column] = min(max(widths.get(cell.column, 0), len(value) + 2), 42)
|
|
for col_idx, width in widths.items():
|
|
ws.column_dimensions[get_column_letter(col_idx)].width = width
|
|
|
|
|
|
def style_sheet(ws) -> None:
|
|
ws.freeze_panes = "A2"
|
|
ws.auto_filter.ref = ws.dimensions
|
|
for cell in ws[1]:
|
|
cell.fill = HEADER_FILL
|
|
cell.font = HEADER_FONT
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
for row in ws.iter_rows(min_row=2):
|
|
status_value = clean(row[0].value)
|
|
if status_value == "Recheck":
|
|
fill = RECHECK_FILL
|
|
elif status_value == "Unmatched":
|
|
fill = UNMATCHED_FILL
|
|
elif status_value == "ERP Unmatched":
|
|
fill = ERP_UNMATCHED_FILL
|
|
else:
|
|
fill = None
|
|
if fill:
|
|
for cell in row:
|
|
cell.fill = fill
|
|
for cell in row:
|
|
cell.alignment = Alignment(vertical="top", wrap_text=True)
|
|
autosize(ws)
|
|
|
|
|
|
def append_rows(ws, headers: list[str], rows: Iterable[list[Any]]) -> None:
|
|
ws.append(headers)
|
|
for row in rows:
|
|
ws.append(row)
|
|
style_sheet(ws)
|
|
|
|
|
|
def append_rows_write_only(workbook: Workbook, title: str, headers: list[str], rows: Iterable[list[Any]]) -> None:
|
|
ws = workbook.create_sheet(title)
|
|
header_row: list[WriteOnlyCell] = []
|
|
for value in headers:
|
|
cell = WriteOnlyCell(ws, value=value)
|
|
cell.fill = HEADER_FILL
|
|
cell.font = HEADER_FONT
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
header_row.append(cell)
|
|
ws.append(header_row)
|
|
for row in rows:
|
|
ws.append(list(row))
|
|
|
|
|
|
def flatten_groups(
|
|
groups: list[dict[str, Any]],
|
|
card_label: str,
|
|
only_statuses: set[str] | None = None,
|
|
) -> list[list[Any]]:
|
|
rows: list[list[Any]] = []
|
|
for group_index, group in enumerate(groups, start=1):
|
|
summary = group.get("summary", {})
|
|
link_count = len(summary.get("erp_group_keys") or summary.get("wehago_group_keys") or [])
|
|
group_rows = group.get("rows", [])
|
|
for row_index, row in enumerate(group_rows, start=1):
|
|
status_label = clean(row.get("status_label"))
|
|
if only_statuses and status_label not in only_statuses:
|
|
continue
|
|
rows.append(
|
|
[
|
|
card_label,
|
|
group_index,
|
|
row_index,
|
|
status_label,
|
|
summary.get("fiscal_year"),
|
|
clean(summary.get("ledger_date")),
|
|
clean(summary.get("proof_date")),
|
|
clean(summary.get("voucher_no")),
|
|
clean(summary.get("draft_no")),
|
|
clean(summary.get("ledger_accounts")),
|
|
clean(summary.get("voucher_accounts")),
|
|
clean(summary.get("ledger_vendors")),
|
|
clean(summary.get("voucher_vendors")),
|
|
clean(summary.get("review_reason")),
|
|
len(group_rows),
|
|
link_count,
|
|
clean(row.get("ledger_date")),
|
|
clean(row.get("proof_date")),
|
|
clean(row.get("voucher_no")),
|
|
clean(row.get("draft_no")),
|
|
clean(row.get("ledger_account_name")),
|
|
clean(row.get("voucher_account_name")),
|
|
clean(row.get("ledger_vendor")),
|
|
clean(row.get("voucher_vendor")),
|
|
amount(row.get("ledger_debit")),
|
|
amount(row.get("ledger_credit")),
|
|
amount(row.get("voucher_debit")),
|
|
amount(row.get("voucher_credit")),
|
|
clean(row.get("ledger_desc")),
|
|
clean(row.get("voucher_desc")),
|
|
]
|
|
)
|
|
return rows
|
|
|
|
|
|
def iter_card_issue_rows(groups: list[dict[str, Any]], card_kind: str) -> list[list[Any]]:
|
|
issues: list[list[Any]] = []
|
|
for group in groups:
|
|
summary = group.get("summary", {})
|
|
parent_year = summary.get("fiscal_year")
|
|
parent_date = clean(summary.get("ledger_date") or summary.get("proof_date"))
|
|
parent_voucher_no = clean(summary.get("voucher_no"))
|
|
parent_draft_no = clean(summary.get("draft_no"))
|
|
for row in group.get("rows", []):
|
|
status_label = clean(row.get("status_label"))
|
|
if card_kind == "wehago" and status_label not in {"ERP Unmatched", "Recheck"}:
|
|
continue
|
|
if card_kind == "voucher" and status_label not in {"Unmatched", "Recheck"}:
|
|
continue
|
|
issues.append(
|
|
[
|
|
status_label,
|
|
parent_year,
|
|
parent_date,
|
|
parent_voucher_no,
|
|
parent_draft_no,
|
|
clean(summary.get("ledger_accounts")),
|
|
clean(summary.get("voucher_accounts")),
|
|
clean(summary.get("ledger_vendors")),
|
|
clean(summary.get("voucher_vendors")),
|
|
clean(summary.get("review_reason")),
|
|
clean(row.get("ledger_date")),
|
|
clean(row.get("proof_date")),
|
|
clean(row.get("voucher_no")),
|
|
clean(row.get("draft_no")),
|
|
clean(row.get("ledger_account_name")),
|
|
clean(row.get("voucher_account_name")),
|
|
clean(row.get("ledger_vendor")),
|
|
clean(row.get("voucher_vendor")),
|
|
amount(row.get("ledger_debit")),
|
|
amount(row.get("ledger_credit")),
|
|
amount(row.get("voucher_debit")),
|
|
amount(row.get("voucher_credit")),
|
|
clean(row.get("ledger_desc")),
|
|
clean(row.get("voucher_desc")),
|
|
]
|
|
)
|
|
return issues
|
|
|
|
|
|
def build_relation_rows(
|
|
wehago_groups: list[dict[str, Any]],
|
|
voucher_groups: list[dict[str, Any]],
|
|
) -> list[list[Any]]:
|
|
rows: list[list[Any]] = []
|
|
for group in wehago_groups:
|
|
summary = group.get("summary", {})
|
|
erp_keys = summary.get("erp_group_keys") or set()
|
|
if len(erp_keys) <= 1:
|
|
continue
|
|
rows.append(
|
|
[
|
|
"WEHAGO 1 : ERP N",
|
|
summary.get("fiscal_year"),
|
|
clean(summary.get("ledger_date")),
|
|
clean(summary.get("proof_date")),
|
|
clean(summary.get("voucher_no")),
|
|
clean(summary.get("draft_no")),
|
|
clean(summary.get("ledger_accounts")),
|
|
clean(summary.get("voucher_accounts")),
|
|
len(erp_keys),
|
|
len(group.get("rows", [])),
|
|
clean(summary.get("review_reason")),
|
|
" | ".join(sorted(clean(key) for key in erp_keys if clean(key))),
|
|
]
|
|
)
|
|
for group in voucher_groups:
|
|
summary = group.get("summary", {})
|
|
wehago_keys = summary.get("wehago_group_keys") or set()
|
|
if len(wehago_keys) <= 1:
|
|
continue
|
|
rows.append(
|
|
[
|
|
"ERP 1 : WEHAGO N",
|
|
summary.get("fiscal_year"),
|
|
clean(summary.get("ledger_date")),
|
|
clean(summary.get("proof_date")),
|
|
clean(summary.get("voucher_no")),
|
|
clean(summary.get("draft_no")),
|
|
clean(summary.get("ledger_accounts")),
|
|
clean(summary.get("voucher_accounts")),
|
|
len(wehago_keys),
|
|
len(group.get("rows", [])),
|
|
clean(summary.get("review_reason")),
|
|
" | ".join(sorted(clean(key) for key in wehago_keys if clean(key))),
|
|
]
|
|
)
|
|
return rows
|
|
|
|
|
|
def export_voucher_card_diff(start_year: int, end_year: int) -> Path:
|
|
EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
|
workbook = Workbook(write_only=True)
|
|
|
|
detail_headers = [
|
|
"카드",
|
|
"그룹순번",
|
|
"행순번",
|
|
"행상태",
|
|
"연도",
|
|
"부모 WEHAGO 일자",
|
|
"부모 ERP 일자",
|
|
"부모 전표번호",
|
|
"부모 가전표번호",
|
|
"부모 WEHAGO 계정",
|
|
"부모 ERP 계정",
|
|
"부모 WEHAGO 거래처",
|
|
"부모 ERP 거래처",
|
|
"부모 검증근거",
|
|
"부모 그룹 행수",
|
|
"연결된 반대 그룹 수",
|
|
"행 WEHAGO 일자",
|
|
"행 ERP 일자",
|
|
"행 전표번호",
|
|
"행 가전표번호",
|
|
"행 WEHAGO 계정",
|
|
"행 ERP 계정",
|
|
"행 WEHAGO 거래처",
|
|
"행 ERP 거래처",
|
|
"행 WEHAGO 차변",
|
|
"행 WEHAGO 대변",
|
|
"행 ERP 차변",
|
|
"행 ERP 대변",
|
|
"행 WEHAGO 적요",
|
|
"행 ERP 적요",
|
|
]
|
|
|
|
all_wehago_match_rows: list[list[Any]] = []
|
|
all_wehago_unmatched_rows: list[list[Any]] = []
|
|
all_voucher_match_rows: list[list[Any]] = []
|
|
all_voucher_unmatched_rows: list[list[Any]] = []
|
|
diff_summary_rows: list[list[Any]] = []
|
|
relation_rows_all: list[list[Any]] = []
|
|
|
|
for year in range(start_year, end_year + 1):
|
|
rows_by_status = _get_cached_status_rows_by_range(engine, year, year)
|
|
sections = _get_cached_voucher_sections_by_range(engine, year, year, rows_by_status)
|
|
wehago_groups = sections.get("voucher_matched", [])
|
|
voucher_groups = sections.get("erp_voucher_matched", [])
|
|
wehago_unmatched_groups = sections.get("voucher_unmatched", [])
|
|
wehago_issue_rows = iter_card_issue_rows(wehago_groups, "wehago")
|
|
voucher_issue_rows = iter_card_issue_rows(voucher_groups, "voucher")
|
|
relation_rows = build_relation_rows(wehago_groups, voucher_groups)
|
|
wehago_multi = sum(1 for row in relation_rows if row[0] == "WEHAGO 1 : ERP N")
|
|
voucher_multi = sum(1 for row in relation_rows if row[0] == "ERP 1 : WEHAGO N")
|
|
all_wehago_match_rows.extend(flatten_groups(wehago_groups, "WEHAGO"))
|
|
all_wehago_unmatched_rows.extend(flatten_groups(wehago_unmatched_groups, "WEHAGO Unmatched"))
|
|
all_voucher_match_rows.extend(flatten_groups(voucher_groups, "Voucher"))
|
|
all_voucher_unmatched_rows.extend(flatten_groups(voucher_groups, "Voucher", {"Unmatched", "Recheck"}))
|
|
relation_rows_all.extend(relation_rows)
|
|
diff_summary_rows.append(
|
|
[
|
|
"요약",
|
|
year,
|
|
len(wehago_groups),
|
|
len(wehago_unmatched_groups),
|
|
len(voucher_groups),
|
|
len(voucher_issue_rows),
|
|
len(wehago_groups) - len(voucher_groups),
|
|
wehago_multi,
|
|
voucher_multi,
|
|
len(wehago_issue_rows),
|
|
len(voucher_issue_rows),
|
|
"WEHAGO 카드는 WEHAGO 전표 기준 그룹 수, Voucher 카드는 ERP 전표 기준 그룹 수입니다.",
|
|
]
|
|
)
|
|
|
|
append_rows_write_only(
|
|
workbook,
|
|
"1_WEHAGO_Match",
|
|
detail_headers,
|
|
all_wehago_match_rows or [["WEHAGO", "", "", "", "", "", "", "", "", "", "", "", "", "데이터가 없습니다.", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]],
|
|
)
|
|
append_rows_write_only(
|
|
workbook,
|
|
"2_WEHAGO_Unmatched",
|
|
detail_headers,
|
|
all_wehago_unmatched_rows or [["WEHAGO Unmatched", "", "", "", "", "", "", "", "", "", "", "", "", "데이터가 없습니다.", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]],
|
|
)
|
|
append_rows_write_only(
|
|
workbook,
|
|
"3_Voucher_Match",
|
|
detail_headers,
|
|
all_voucher_match_rows or [["Voucher", "", "", "", "", "", "", "", "", "", "", "", "", "데이터가 없습니다.", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]],
|
|
)
|
|
append_rows_write_only(
|
|
workbook,
|
|
"4_Voucher_Unmatched",
|
|
detail_headers,
|
|
all_voucher_unmatched_rows or [["Voucher", "", "", "", "", "", "", "", "", "", "", "", "", "Voucher 카드 안에서 Unmatched/Recheck로 남은 행이 없습니다.", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]],
|
|
)
|
|
|
|
diff_headers = [
|
|
"구분",
|
|
"연도",
|
|
"WEHAGO 매치 그룹 수",
|
|
"WEHAGO Unmatched 그룹 수",
|
|
"Voucher 매치 그룹 수",
|
|
"Voucher unmatched 행 수",
|
|
"WEHAGO-Voucher 그룹 수 차이",
|
|
"WEHAGO 1:ERP N 건수",
|
|
"ERP 1:WEHAGO N 건수",
|
|
"WEHAGO 카드 내 ERP Unmatched/Recheck 행 수",
|
|
"Voucher 카드 내 Unmatched/Recheck 행 수",
|
|
"설명",
|
|
]
|
|
diff_rows: list[list[Any]] = []
|
|
diff_rows.append(diff_headers)
|
|
diff_rows.extend(diff_summary_rows)
|
|
if diff_summary_rows:
|
|
total_wehago = sum(int(row[2]) for row in diff_summary_rows)
|
|
total_wehago_unmatched = sum(int(row[3]) for row in diff_summary_rows)
|
|
total_voucher = sum(int(row[4]) for row in diff_summary_rows)
|
|
total_voucher_unmatched = sum(int(row[5]) for row in diff_summary_rows)
|
|
total_diff = sum(int(row[6]) for row in diff_summary_rows)
|
|
total_wehago_multi = sum(int(row[7]) for row in diff_summary_rows)
|
|
total_voucher_multi = sum(int(row[8]) for row in diff_summary_rows)
|
|
total_wehago_issues = sum(int(row[9]) for row in diff_summary_rows)
|
|
total_voucher_issues = sum(int(row[10]) for row in diff_summary_rows)
|
|
diff_rows.append(
|
|
[
|
|
"전체",
|
|
f"{start_year}~{end_year}",
|
|
total_wehago,
|
|
total_wehago_unmatched,
|
|
total_voucher,
|
|
total_voucher_unmatched,
|
|
total_diff,
|
|
total_wehago_multi,
|
|
total_voucher_multi,
|
|
total_wehago_issues,
|
|
total_voucher_issues,
|
|
"그룹 수 차이는 한쪽 전표 1건이 반대쪽 여러 전표와 연결되는 경우에서 주로 발생합니다.",
|
|
]
|
|
)
|
|
diff_rows.append([])
|
|
diff_rows.append(["관계 유형", "연도", "WEHAGO 일자", "ERP 일자", "전표번호", "가전표번호", "WEHAGO 계정", "ERP 계정", "연결된 반대 그룹 수", "그룹 행 수", "검증 근거", "연결된 그룹 키"])
|
|
diff_rows.extend(relation_rows_all or [["", "", "", "", "", "", "", "", "", "", "수 차이를 만든 다대일/일대다 사례가 없습니다.", ""]])
|
|
diff_ws = workbook.create_sheet("5_Count_Diff")
|
|
for index, row in enumerate(diff_rows, start=1):
|
|
if index in {1, len(diff_summary_rows) + 3}:
|
|
cells: list[WriteOnlyCell] = []
|
|
for value in row:
|
|
cell = WriteOnlyCell(diff_ws, value=value)
|
|
cell.fill = HEADER_FILL
|
|
cell.font = HEADER_FONT
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
cells.append(cell)
|
|
diff_ws.append(cells)
|
|
else:
|
|
diff_ws.append(row)
|
|
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
output_path = EXPORT_DIR / f"voucher_card_diff_{start_year}_{end_year}_{timestamp}.xlsx"
|
|
workbook.save(output_path)
|
|
return output_path
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
years = discover_available_years()
|
|
default_start = min(years) if years else datetime.now().year
|
|
default_end = max(years) if years else default_start
|
|
parser = argparse.ArgumentParser(description="WEHAGO/Voucher 카드 차이 분석 엑셀 생성")
|
|
parser.add_argument("--start-year", type=int, default=default_start)
|
|
parser.add_argument("--end-year", type=int, default=default_end)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
start_year = min(args.start_year, args.end_year)
|
|
end_year = max(args.start_year, args.end_year)
|
|
output_path = export_voucher_card_diff(start_year, end_year)
|
|
print(output_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|