428 lines
16 KiB
Python
428 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, PatternFill
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
import sys
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from runtime_config import DB_PATH # noqa: E402
|
|
|
|
EXPORT_DIR = ROOT / "static" / "exports" / "wehago_compare"
|
|
YEAR = 2025
|
|
WEHAGO_STATUSES = ("voucher_matched", "voucher_unmatched", "voucher_recheck")
|
|
|
|
|
|
def clean(value: Any) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
def key_from_display(year: int, ledger_date: str, voucher_no: str) -> str:
|
|
voucher = clean(voucher_no)
|
|
if re.fullmatch(r"\d{8}-\d{5}", voucher):
|
|
return voucher
|
|
match = re.fullmatch(r"(\d{4})-(\d{2})-(\d{5})", voucher)
|
|
if match:
|
|
return f"{match.group(1)}{match.group(2)}-{match.group(3)}"
|
|
date_text = clean(ledger_date)
|
|
date_match = re.search(r"(\d{1,2})[-./](\d{1,2})", date_text)
|
|
voucher_digits = re.sub(r"\D+", "", voucher)
|
|
if date_match and voucher_digits:
|
|
month = int(date_match.group(1))
|
|
day = int(date_match.group(2))
|
|
return f"{int(year):04d}{month:02d}{day:02d}-{int(voucher_digits):05d}"
|
|
return ""
|
|
|
|
|
|
def latest_projection_signature(cur: sqlite3.Cursor) -> str:
|
|
row = cur.execute(
|
|
"""
|
|
SELECT payload_json
|
|
FROM wehago_action_history
|
|
WHERE action_type = 'auto_recheck_promote'
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
"""
|
|
).fetchone()
|
|
if row:
|
|
try:
|
|
payload = json.loads(row[0] or "{}")
|
|
signature = clean(payload.get("signature"))
|
|
if signature:
|
|
exists = cur.execute(
|
|
"""
|
|
SELECT 1
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ? AND end_year = ? AND signature = ?
|
|
LIMIT 1
|
|
""",
|
|
(YEAR, YEAR, signature),
|
|
).fetchone()
|
|
if exists:
|
|
return signature
|
|
except Exception:
|
|
pass
|
|
row = cur.execute(
|
|
"""
|
|
SELECT signature, MAX(updated_at) AS max_updated_at
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ? AND end_year = ? AND signature LIKE 'compare-query-v5|%'
|
|
GROUP BY signature
|
|
ORDER BY max_updated_at DESC
|
|
LIMIT 1
|
|
""",
|
|
(YEAR, YEAR),
|
|
).fetchone()
|
|
return clean(row[0]) if row else ""
|
|
|
|
|
|
def append_rows(ws, rows: list[list[Any]]) -> None:
|
|
for row in rows:
|
|
ws.append(row)
|
|
|
|
|
|
def style_sheet(ws) -> None:
|
|
header_fill = PatternFill("solid", fgColor="D9EAF7")
|
|
for cell in ws[1]:
|
|
cell.font = Font(bold=True)
|
|
cell.fill = header_fill
|
|
ws.freeze_panes = "A2"
|
|
ws.auto_filter.ref = ws.dimensions
|
|
for col_idx, column_cells in enumerate(ws.columns, start=1):
|
|
max_len = 10
|
|
for cell in column_cells:
|
|
max_len = max(max_len, min(len(clean(cell.value)), 80))
|
|
ws.column_dimensions[get_column_letter(col_idx)].width = max_len + 2
|
|
|
|
|
|
def main() -> None:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
signature = latest_projection_signature(cur)
|
|
if not signature:
|
|
raise RuntimeError("compare query projection signature not found")
|
|
|
|
db_rows = cur.execute(
|
|
"""
|
|
SELECT fiscal_year, voucher_no, status, voucher_row_count, ledger_row_count,
|
|
voucher_debit, voucher_credit, ledger_debit, ledger_credit,
|
|
voucher_accounts, ledger_accounts, voucher_vendors, ledger_vendors, notes
|
|
FROM wehago_comparison_results
|
|
WHERE fiscal_year = ?
|
|
AND status <> 'voucher_only'
|
|
ORDER BY voucher_no
|
|
""",
|
|
(YEAR,),
|
|
).fetchall()
|
|
db_by_key = {clean(row["voucher_no"]): dict(row) for row in db_rows if clean(row["voucher_no"])}
|
|
comparison_total = cur.execute(
|
|
"""
|
|
SELECT COUNT(*) AS count_all,
|
|
SUM(CASE WHEN status = 'voucher_only' THEN 1 ELSE 0 END) AS voucher_only_count
|
|
FROM wehago_comparison_results
|
|
WHERE fiscal_year = ?
|
|
""",
|
|
(YEAR,),
|
|
).fetchone()
|
|
|
|
ledger_source = cur.execute(
|
|
"""
|
|
SELECT COUNT(*) AS row_count,
|
|
COUNT(DISTINCT fiscal_year || '|' || compare_voucher_no) AS voucher_count
|
|
FROM wehago_ledger_rows
|
|
WHERE fiscal_year = ? AND COALESCE(compare_voucher_no, '') <> ''
|
|
""",
|
|
(YEAR,),
|
|
).fetchone()
|
|
voucher_source = cur.execute(
|
|
"""
|
|
SELECT COUNT(*) AS row_count,
|
|
COUNT(DISTINCT fiscal_year || '|' || compare_voucher_no) AS voucher_count
|
|
FROM wehago_voucher_rows
|
|
WHERE fiscal_year = ? AND COALESCE(compare_voucher_no, '') <> ''
|
|
""",
|
|
(YEAR,),
|
|
).fetchone()
|
|
|
|
ui_rows = cur.execute(
|
|
f"""
|
|
SELECT status_key, group_index, fiscal_year, ledger_date, proof_date, voucher_no, draft_no,
|
|
ledger_row_count, voucher_row_count, ledger_debit, ledger_credit, voucher_debit, voucher_credit,
|
|
ledger_accounts, voucher_accounts, ledger_vendors, voucher_vendors, review_reason
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ? AND end_year = ? AND signature = ?
|
|
AND status_key IN ({','.join('?' for _ in WEHAGO_STATUSES)})
|
|
ORDER BY status_key, group_index
|
|
""",
|
|
(YEAR, YEAR, signature, *WEHAGO_STATUSES),
|
|
).fetchall()
|
|
|
|
ui_group_records: list[dict[str, Any]] = []
|
|
for row in ui_rows:
|
|
record = dict(row)
|
|
record["compare_key"] = key_from_display(record["fiscal_year"], record["ledger_date"], record["voucher_no"])
|
|
ui_group_records.append(record)
|
|
|
|
section_counts = Counter(record["status_key"] for record in ui_group_records)
|
|
ui_by_key: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
bad_key_records: list[dict[str, Any]] = []
|
|
for record in ui_group_records:
|
|
key = clean(record.get("compare_key"))
|
|
if key:
|
|
ui_by_key[key].append(record)
|
|
else:
|
|
bad_key_records.append(record)
|
|
|
|
db_keys = set(db_by_key)
|
|
ui_keys = set(ui_by_key)
|
|
duplicate_ui = {key: records for key, records in ui_by_key.items() if len(records) > 1}
|
|
ui_not_in_db = {key: records for key, records in ui_by_key.items() if key not in db_keys}
|
|
db_not_in_ui = {key: db_by_key[key] for key in sorted(db_keys - ui_keys)}
|
|
|
|
section_sum = sum(section_counts.get(status, 0) for status in WEHAGO_STATUSES)
|
|
db_count = len(db_by_key)
|
|
ui_unique_count = len(ui_keys)
|
|
duplicate_extra_count = section_sum - ui_unique_count
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "요약"
|
|
append_rows(
|
|
ws,
|
|
[
|
|
["항목", "값", "설명"],
|
|
["분석 기준 연도", YEAR, ""],
|
|
["사용 projection signature", signature, "화면 WEHAGO/Recheck 섹션에 사용된 최신 projection"],
|
|
["DB WEHAGO 전표 수", db_count, "wehago_comparison_results의 fiscal_year+voucher_no 기준"],
|
|
["DB 비교결과 전체 전표 수", int(comparison_total["count_all"] or 0), "ERP-only(voucher_only)를 포함한 전체 비교 결과"],
|
|
["DB 비교결과 중 ERP-only 수", int(comparison_total["voucher_only_count"] or 0), "WEHAGO 전표 수 비교에서는 제외"],
|
|
["DB WEHAGO 원장 전표 수", int(ledger_source["voucher_count"] or 0), "wehago_ledger_rows의 fiscal_year+compare_voucher_no distinct"],
|
|
["DB WEHAGO 원장 행 수", int(ledger_source["row_count"] or 0), "wehago_ledger_rows 행 수"],
|
|
["보조: wehago_voucher_rows 전표 수", int(voucher_source["voucher_count"] or 0), "ERP/증빙성 원천 행으로 보이는 테이블의 distinct 전표 수"],
|
|
["WEHAGO", section_counts.get("voucher_matched", 0), "화면 WEHAGO 그룹 수"],
|
|
["WEHAGO Unmatched", section_counts.get("voucher_unmatched", 0), "화면 WEHAGO Unmatched 그룹 수"],
|
|
["WEHAGO Recheck", section_counts.get("voucher_recheck", 0), "화면 WEHAGO Recheck 그룹 수"],
|
|
["화면 3개 섹션 합계", section_sum, "WEHAGO + WEHAGO Unmatched + WEHAGO Recheck"],
|
|
["차이(화면 합계 - DB WEHAGO)", section_sum - db_count, "양수면 화면 그룹 집계가 DB 전표 수보다 많음"],
|
|
["화면 고유 WEHAGO 전표 수", ui_unique_count, "ledger_date+voucher_no를 YYYYMMDD-전표번호로 환산한 distinct"],
|
|
["차이(화면 고유 전표 - DB WEHAGO)", ui_unique_count - db_count, "전표번호 기준 순수 누락/추가 차이"],
|
|
["중복 그룹 초과분", duplicate_extra_count, "같은 WEHAGO 전표가 여러 그룹으로 나뉘어 합계에 중복 반영된 수"],
|
|
["키 생성 불가 그룹", len(bad_key_records), "ledger_date/voucher_no 조합으로 DB 전표번호를 만들 수 없는 그룹"],
|
|
],
|
|
)
|
|
style_sheet(ws)
|
|
|
|
ws = wb.create_sheet("차이 사유")
|
|
reason_rows = [
|
|
["사유", "건수", "해석"],
|
|
[
|
|
"같은 WEHAGO 전표가 화면에서 복수 그룹으로 집계됨",
|
|
duplicate_extra_count,
|
|
"전표 하나가 복수 ERP 가전표/후보/라인 그룹으로 분리되면 화면 섹션 합계는 DB 전표 수보다 커집니다.",
|
|
],
|
|
[
|
|
"화면에는 있으나 DB comparison_results 전표번호와 직접 대응되지 않음",
|
|
len(ui_not_in_db),
|
|
"표시 전표번호를 YYYYMMDD-전표번호로 환산해도 DB 전표번호 집합에 없는 경우입니다.",
|
|
],
|
|
[
|
|
"DB에는 있으나 화면 3개 WEHAGO 섹션에 없음",
|
|
len(db_not_in_ui),
|
|
"DB 비교 결과에는 있으나 현재 projection의 WEHAGO/Unmatched/Recheck 그룹에는 없는 경우입니다.",
|
|
],
|
|
[
|
|
"전표번호 키 생성 불가",
|
|
len(bad_key_records),
|
|
"화면 그룹의 일자 또는 전표번호가 비어 있거나 형식이 달라 비교 키를 만들지 못한 경우입니다.",
|
|
],
|
|
]
|
|
append_rows(ws, reason_rows)
|
|
style_sheet(ws)
|
|
|
|
ws = wb.create_sheet("중복 그룹 상세")
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
"compare_key",
|
|
"그룹 수",
|
|
"status 목록",
|
|
"전표 표시값",
|
|
"일자 목록",
|
|
"가전표번호/ERP 전표",
|
|
"WEHAGO 계정",
|
|
"ERP 계정",
|
|
"검토 사유",
|
|
]],
|
|
)
|
|
for key, records in sorted(duplicate_ui.items(), key=lambda item: (-len(item[1]), item[0])):
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
key,
|
|
len(records),
|
|
", ".join(sorted({clean(r.get("status_key")) for r in records})),
|
|
", ".join(sorted({clean(r.get("voucher_no")) for r in records if clean(r.get("voucher_no"))}))[:300],
|
|
", ".join(sorted({clean(r.get("ledger_date")) for r in records if clean(r.get("ledger_date"))}))[:300],
|
|
", ".join(clean(r.get("draft_no")) for r in records if clean(r.get("draft_no")))[:1000],
|
|
" | ".join(clean(r.get("ledger_accounts")) for r in records if clean(r.get("ledger_accounts")))[:1000],
|
|
" | ".join(clean(r.get("voucher_accounts")) for r in records if clean(r.get("voucher_accounts")))[:1000],
|
|
" | ".join(clean(r.get("review_reason")) for r in records if clean(r.get("review_reason")))[:1000],
|
|
]],
|
|
)
|
|
style_sheet(ws)
|
|
|
|
ws = wb.create_sheet("화면만 있음")
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
"compare_key",
|
|
"그룹 수",
|
|
"status 목록",
|
|
"일자 목록",
|
|
"전표 표시값",
|
|
"가전표번호/ERP 전표",
|
|
"WEHAGO 계정",
|
|
"ERP 계정",
|
|
"검토 사유",
|
|
]],
|
|
)
|
|
for key, records in sorted(ui_not_in_db.items()):
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
key,
|
|
len(records),
|
|
", ".join(sorted({clean(r.get("status_key")) for r in records})),
|
|
", ".join(sorted({clean(r.get("ledger_date")) for r in records if clean(r.get("ledger_date"))}))[:300],
|
|
", ".join(sorted({clean(r.get("voucher_no")) for r in records if clean(r.get("voucher_no"))}))[:300],
|
|
", ".join(clean(r.get("draft_no")) for r in records if clean(r.get("draft_no")))[:1000],
|
|
" | ".join(clean(r.get("ledger_accounts")) for r in records if clean(r.get("ledger_accounts")))[:1000],
|
|
" | ".join(clean(r.get("voucher_accounts")) for r in records if clean(r.get("voucher_accounts")))[:1000],
|
|
" | ".join(clean(r.get("review_reason")) for r in records if clean(r.get("review_reason")))[:1000],
|
|
]],
|
|
)
|
|
style_sheet(ws)
|
|
|
|
ws = wb.create_sheet("DB만 있음")
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
"compare_key",
|
|
"DB status",
|
|
"voucher_row_count",
|
|
"ledger_row_count",
|
|
"WEHAGO 차변",
|
|
"WEHAGO 대변",
|
|
"ERP 차변",
|
|
"ERP 대변",
|
|
"WEHAGO 계정",
|
|
"ERP 계정",
|
|
"비고",
|
|
]],
|
|
)
|
|
for key, row in db_not_in_ui.items():
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
key,
|
|
clean(row.get("status")),
|
|
row.get("ledger_row_count"),
|
|
row.get("voucher_row_count"),
|
|
row.get("ledger_debit"),
|
|
row.get("ledger_credit"),
|
|
row.get("voucher_debit"),
|
|
row.get("voucher_credit"),
|
|
clean(row.get("ledger_accounts")),
|
|
clean(row.get("voucher_accounts")),
|
|
clean(row.get("notes")),
|
|
]],
|
|
)
|
|
style_sheet(ws)
|
|
|
|
ws = wb.create_sheet("섹션 원자료")
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
"status",
|
|
"compare_key",
|
|
"fiscal_year",
|
|
"ledger_date",
|
|
"voucher_no",
|
|
"draft_no",
|
|
"ledger_row_count",
|
|
"voucher_row_count",
|
|
"WEHAGO 차변",
|
|
"WEHAGO 대변",
|
|
"ERP 차변",
|
|
"ERP 대변",
|
|
"WEHAGO 계정",
|
|
"ERP 계정",
|
|
"WEHAGO 거래처",
|
|
"ERP 거래처",
|
|
"review_reason",
|
|
]],
|
|
)
|
|
for record in ui_group_records:
|
|
append_rows(
|
|
ws,
|
|
[[
|
|
record.get("status_key"),
|
|
record.get("compare_key"),
|
|
record.get("fiscal_year"),
|
|
record.get("ledger_date"),
|
|
record.get("voucher_no"),
|
|
record.get("draft_no"),
|
|
record.get("ledger_row_count"),
|
|
record.get("voucher_row_count"),
|
|
record.get("ledger_debit"),
|
|
record.get("ledger_credit"),
|
|
record.get("voucher_debit"),
|
|
record.get("voucher_credit"),
|
|
record.get("ledger_accounts"),
|
|
record.get("voucher_accounts"),
|
|
record.get("ledger_vendors"),
|
|
record.get("voucher_vendors"),
|
|
record.get("review_reason"),
|
|
]],
|
|
)
|
|
style_sheet(ws)
|
|
|
|
EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
|
file_name = f"wehago_voucher_coverage_{YEAR}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
output_path = EXPORT_DIR / file_name
|
|
wb.save(output_path)
|
|
conn.close()
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"file": str(output_path),
|
|
"download_url": f"/static/exports/wehago_compare/{file_name}",
|
|
"db_count": db_count,
|
|
"section_sum": section_sum,
|
|
"diff": section_sum - db_count,
|
|
"ui_unique_count": ui_unique_count,
|
|
"unique_diff": ui_unique_count - db_count,
|
|
"duplicate_extra_count": duplicate_extra_count,
|
|
"ui_not_in_db": len(ui_not_in_db),
|
|
"db_not_in_ui": len(db_not_in_ui),
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|