#!/usr/bin/env python3 """Materialize the 2025 WEHAGO compare projection from canonical voucher keys. This script is intentionally conservative. It uses the current projection as a candidate seed, but the final WEHAGO population is the raw ledger voucher set. That prevents stale export rows, display-date variants, and partial context rows from changing card/table/export counts. """ from __future__ import annotations import json import re import sqlite3 from collections import Counter, defaultdict from pathlib import Path from typing import Any START_YEAR = 2025 END_YEAR = 2025 DB_PATH = Path("/runtime/db/data.db") WEHAGO_STATUSES = ( "voucher_matched", "voucher_unmatched", "voucher_recheck", "voucher_excepted", ) STATUS_LABELS = { "voucher_matched": "Voucher", "voucher_unmatched": "Unmatched", "voucher_recheck": "Recheck", "voucher_excepted": "Excepted", } PROMOTE_TO_MATCHED = { (2025, "04-29", "00163"): "최종 validator: proof_date와 전표일자 일치, 금액·계정·거래처 문맥 일치", (2025, "09-24", "50036"): "최종 validator: proof_date와 전표일자 일치, 음수 조정 방향·계정·금액 일치", (2025, "10-14", "00001"): "최종 validator: 자금성 전표 기타예금/단기금융상품(기타예금) 동치 및 보통예금 금액 일치", (2025, "10-27", "00002"): "사용자 지정 오타 허용 특수 승격: 금액·계정·거래처 일치", (2025, "10-28", "00001"): "최종 validator: 원천 ERP 외부 후보 탐색 대상 포함, 자금성 전표 동치 규칙 적용", (2025, "10-30", "00005"): "사용자 지정 오타 허용 특수 승격: 금액·계정 일치", (2025, "11-24", "00106"): "사용자 지정 오타 허용 특수 승격: 통신비 금액·거래처 문맥 일치", (2025, "11-25", "00009"): "최종 validator: 비용 차감 전표 금액·계정·거래 문맥 일치", } KEEP_RECHECK = { (2025, "01-01", "00063"): "전표 일부 직접 대응은 있으나 ERP 국고보조금/보통예금 잔여행 검토 필요", (2025, "02-21", "00176"): "사용자 판정: 카드 비용 전표와 ERP 미지급금 출금 전표가 전표 전체로는 동일 성격이 아님", (2025, "03-31", "00048"): "사용자 판정: 상계 특수 사례는 일부 행만 대응 표시하고 전표 전체는 recheck 유지", (2025, "04-16", "50009"): "지급수수료/매입세액 직접 대응과 ERP 조정행이 함께 있어 recheck 유지", (2025, "04-23", "00007"): "사용자 판정: 카드 비용 전표와 ERP 미지급금 출금 전표가 전표 전체로는 동일 성격이 아님", (2025, "06-30", "00156"): "사용자 판정: 상계 특수 사례는 일부 행만 대응 표시하고 전표 전체는 recheck 유지", (2025, "09-22", "50046"): "VAT/proof_date 우선 검증: 현재 ERP 후보 proof_date가 WEHAGO 전표일자와 불일치", (2025, "12-19", "50036"): "VAT/proof_date 우선 검증: 현재 ERP 후보 proof_date가 WEHAGO 전표일자와 불일치", (2025, "12-25", "50006"): "VAT/proof_date 우선 검증: 현재 ERP 후보 proof_date가 WEHAGO 전표일자와 불일치", } FORCE_UNMATCHED = { (2025, "08-07", "50005"): "취소/발행 반대방향 전표: 같은 차대 위치의 음수/양수 절대값 일치는 매칭 후보에서 제거", (2025, "06-13", "00038"): "금액 유사 오염 제거: 계정·거래처·적요가 다른 후보", (2025, "07-07", "00001"): "월/기간 불일치 후보 제거: ERP는 3~5월 합산, WEHAGO는 7월 전표", } EXCEPTED_RETAIN_RECHECK_KEYS = { (2025, "03-31", "00048"), (2025, "06-30", "00156"), } def clean(value: Any) -> str: if value is None: return "" text = str(value).strip() if text.endswith(".0") and text[:-2].isdigit(): return text[:-2] return text def amount(value: Any) -> float: if value is None or value == "": return 0.0 if isinstance(value, (int, float)): return float(value) text = str(value).replace(",", "").strip() if not text: return 0.0 try: return float(text) except ValueError: return 0.0 def display_date(value: Any) -> str: text = clean(value) match = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", text) if match: return f"{match.group(2)}-{match.group(3)}" match = re.match(r"^(\d{2})-(\d{2})$", text) if match: return text return text def date_key(year: int, mmdd: str) -> str: return f"{year}-{mmdd}" if re.match(r"^\d{2}-\d{2}$", mmdd) else mmdd def voucher_no(value: Any) -> str: text = clean(value) return text.zfill(5) if text.isdigit() else text def draft_base(value: Any) -> str: return re.sub(r"-\d+$", "", clean(value)) def row_has_wehago(row: dict[str, Any]) -> bool: return bool( clean(row.get("ledger_account_name")) or clean(row.get("ledger_vendor")) or abs(amount(row.get("ledger_debit"))) >= 0.5 or abs(amount(row.get("ledger_credit"))) >= 0.5 ) def row_has_erp(row: dict[str, Any]) -> bool: return bool( clean(row.get("draft_no")) or clean(row.get("voucher_account_name")) or clean(row.get("voucher_vendor")) or abs(amount(row.get("voucher_debit"))) >= 0.5 or abs(amount(row.get("voucher_credit"))) >= 0.5 ) def normalize_row(row: dict[str, Any]) -> dict[str, Any]: normalized = dict(row) normalized["fiscal_year"] = int(normalized.get("fiscal_year") or START_YEAR) normalized["ledger_date"] = display_date(normalized.get("ledger_date")) normalized["voucher_no"] = voucher_no(normalized.get("voucher_no")) for key in ("ledger_debit", "ledger_credit", "voucher_debit", "voucher_credit"): normalized[key] = amount(normalized.get(key)) return normalized def load_current_projection(conn: sqlite3.Connection) -> dict[tuple[int, str, str], dict[str, Any]]: by_key: dict[tuple[int, str, str], dict[str, Any]] = {} rows = conn.execute( """ SELECT * FROM wehago_status_projection_groups WHERE start_year = ? AND end_year = ? AND status_key IN ('voucher_matched', 'voucher_unmatched', 'voucher_recheck', 'voucher_excepted') ORDER BY status_key, group_index """, (START_YEAR, END_YEAR), ).fetchall() status_rank = {"voucher_matched": 4, "voucher_recheck": 3, "voucher_unmatched": 2, "voucher_excepted": 1} for db_row in rows: year = int(db_row["fiscal_year"] or START_YEAR) key = (year, display_date(db_row["ledger_date"]), voucher_no(db_row["voucher_no"])) payload = dict(db_row) try: detail_rows = json.loads(db_row["rows_json"] or "[]") except Exception: detail_rows = [] payload["rows"] = [normalize_row(row) for row in detail_rows if isinstance(row, dict)] existing = by_key.get(key) if not existing: by_key[key] = payload continue existing_iso = re.match(r"^\d{4}-", clean(existing.get("ledger_date") or "")) payload_iso = re.match(r"^\d{4}-", clean(payload.get("ledger_date") or "")) if existing_iso and not payload_iso: by_key[key] = payload elif bool(existing_iso) == bool(payload_iso) and status_rank.get(payload["status_key"], 0) > status_rank.get(existing["status_key"], 0): by_key[key] = payload return by_key def raw_wehago_row(raw: sqlite3.Row) -> dict[str, Any]: year = int(raw["fiscal_year"] or START_YEAR) mmdd = display_date(raw["ledger_date"]) return { "fiscal_year": year, "status_label": "", "ledger_date": mmdd, "ledger_date_key": date_key(year, mmdd), "proof_date": "", "voucher_no": voucher_no(raw["voucher_no"]), "draft_no": "", "ledger_account_name": clean(raw["account_name"]), "voucher_account_name": "", "ledger_vendor": clean(raw["vendor_name"]), "voucher_vendor": "", "ledger_debit": amount(raw["debit"]), "ledger_credit": amount(raw["credit"]), "voucher_debit": 0.0, "voucher_credit": 0.0, "ledger_desc": clean(raw["description"]), "voucher_desc": "", "review_reason": "WEHAGO_FULL_VOUCHER_CONTEXT", "matched_case": "WEHAGO_CONTEXT_ONLY", "ledger_row_key": f"{raw['id']}", "voucher_row_key": "", "match_identity_key": "", } def load_raw_wehago(conn: sqlite3.Connection) -> dict[tuple[int, str, str], list[dict[str, Any]]]: by_key: dict[tuple[int, str, str], list[dict[str, Any]]] = defaultdict(list) rows = conn.execute( """ SELECT id, fiscal_year, ledger_date, voucher_no, account_name, vendor_name, debit, credit, description, row_number FROM wehago_ledger_rows WHERE fiscal_year = ? AND COALESCE(compare_voucher_no, '') <> '' ORDER BY ledger_date, voucher_no, row_number, id """, (START_YEAR,), ).fetchall() for raw in rows: row = raw_wehago_row(raw) key = (int(row["fiscal_year"]), row["ledger_date"], row["voucher_no"]) by_key[key].append(row) return by_key def raw_erp_row(raw: sqlite3.Row) -> dict[str, Any]: desc = " ".join(part for part in [clean(raw["desc1"]), clean(raw["desc2"])] if part) return { "fiscal_year": int(raw["fiscal_year"] or START_YEAR), "status_label": "", "ledger_date": "", "proof_date": clean(raw["proof_date"]), "voucher_no": clean(raw["confirmed_no"]), "draft_no": clean(raw["draft_no"]), "ledger_account_name": "", "voucher_account_name": clean(raw["account_name"]), "ledger_vendor": "", "voucher_vendor": clean(raw["vendor_name"]), "ledger_debit": 0.0, "ledger_credit": 0.0, "voucher_debit": amount(raw["debit_supply"]), "voucher_credit": amount(raw["credit_supply"]), "ledger_desc": "", "voucher_desc": desc, "review_reason": "ERP_FULL_DRAFT_CONTEXT", "matched_case": "ERP_CONTEXT_ONLY", "ledger_row_key": "", "voucher_row_key": f"{raw['id']}", "match_identity_key": "", } def load_raw_erp(conn: sqlite3.Connection) -> dict[str, list[dict[str, Any]]]: by_base: dict[str, list[dict[str, Any]]] = defaultdict(list) rows = conn.execute( """ SELECT id, fiscal_year, proof_date, confirmed_no, draft_no, account_name, debit_supply, credit_supply, vendor_name, desc1, desc2, row_number FROM wehago_voucher_rows WHERE fiscal_year = ? AND COALESCE(draft_no, '') <> '' ORDER BY draft_no, row_number, id """, (START_YEAR,), ).fetchall() for raw in rows: by_base[draft_base(raw["draft_no"])].append(raw_erp_row(raw)) return by_base def identity(row: dict[str, Any], side: str) -> str: if side == "wehago": return "|".join( [ display_date(row.get("ledger_date")), voucher_no(row.get("voucher_no")), clean(row.get("ledger_account_name")), clean(row.get("ledger_vendor")), f"{amount(row.get('ledger_debit')):.2f}", f"{amount(row.get('ledger_credit')):.2f}", clean(row.get("ledger_desc")), ] ) return "|".join( [ clean(row.get("draft_no")), clean(row.get("voucher_account_name")), clean(row.get("voucher_vendor")), f"{amount(row.get('voucher_debit')):.2f}", f"{amount(row.get('voucher_credit')):.2f}", clean(row.get("voucher_desc")), ] ) def blank_row() -> dict[str, Any]: return { "fiscal_year": START_YEAR, "status_label": "", "ledger_date": "", "ledger_date_key": "", "proof_date": "", "voucher_no": "", "draft_no": "", "ledger_account_name": "", "voucher_account_name": "", "ledger_vendor": "", "voucher_vendor": "", "ledger_debit": 0.0, "ledger_credit": 0.0, "voucher_debit": 0.0, "voucher_credit": 0.0, "ledger_desc": "", "voucher_desc": "", "review_reason": "", "matched_case": "", "ledger_row_key": "", "voucher_row_key": "", "match_identity_key": "", } def combine_context_rows(wehago_row: dict[str, Any] | None, erp_row: dict[str, Any] | None) -> dict[str, Any]: left = wehago_row or blank_row() right = erp_row or blank_row() row = blank_row() row.update( { "fiscal_year": int(left.get("fiscal_year") or right.get("fiscal_year") or START_YEAR), "status_label": "", "ledger_date": display_date(left.get("ledger_date")), "ledger_date_key": clean(left.get("ledger_date_key")), "proof_date": clean(right.get("proof_date")), "voucher_no": voucher_no(left.get("voucher_no")), "draft_no": clean(right.get("draft_no")), "ledger_account_name": clean(left.get("ledger_account_name")), "voucher_account_name": clean(right.get("voucher_account_name")), "ledger_vendor": clean(left.get("ledger_vendor")), "voucher_vendor": clean(right.get("voucher_vendor")), "ledger_debit": amount(left.get("ledger_debit")), "ledger_credit": amount(left.get("ledger_credit")), "voucher_debit": amount(right.get("voucher_debit")), "voucher_credit": amount(right.get("voucher_credit")), "ledger_desc": clean(left.get("ledger_desc")), "voucher_desc": clean(right.get("voucher_desc")), "review_reason": "FULL_VOUCHER_CONTEXT", "matched_case": "FULL_CONTEXT_ALIGNED", "ledger_row_key": clean(left.get("ledger_row_key")), "voucher_row_key": clean(right.get("voucher_row_key")), "match_identity_key": "", } ) if not wehago_row: row["matched_case"] = "ERP_CONTEXT_ONLY" elif not erp_row: row["matched_case"] = "WEHAGO_CONTEXT_ONLY" return row def hydrate_rows( status_key: str, seed_rows: list[dict[str, Any]], raw_wehago_rows: list[dict[str, Any]], raw_erp_by_base: dict[str, list[dict[str, Any]]], ) -> list[dict[str, Any]]: if status_key in {"voucher_unmatched", "voucher_excepted"}: rows = [dict(row) for row in raw_wehago_rows] else: bases = sorted({draft_base(row.get("draft_no")) for row in seed_rows if clean(row.get("draft_no"))}) erp_context: list[dict[str, Any]] = [] seen_erp: set[str] = set() for base in bases: for erp_row in raw_erp_by_base.get(base, []): row_id = identity(erp_row, "erp") if row_id in seen_erp: continue erp_context.append(dict(erp_row)) seen_erp.add(row_id) row_count = max(len(raw_wehago_rows), len(erp_context)) rows = [ combine_context_rows( raw_wehago_rows[index] if index < len(raw_wehago_rows) else None, erp_context[index] if index < len(erp_context) else None, ) for index in range(row_count) ] label = STATUS_LABELS.get(status_key, status_key) for row in rows: row["status_label"] = label row["ledger_date"] = display_date(row.get("ledger_date")) return rows def determine_status(key: tuple[int, str, str], seed: dict[str, Any] | None) -> tuple[str, str]: if key in FORCE_UNMATCHED: return "voucher_unmatched", FORCE_UNMATCHED[key] if key in PROMOTE_TO_MATCHED: return "voucher_matched", PROMOTE_TO_MATCHED[key] if key in KEEP_RECHECK: return "voucher_recheck", KEEP_RECHECK[key] if seed: status = clean(seed.get("status_key")) if status in WEHAGO_STATUSES: return status, clean(seed.get("review_reason")) or "원천 WEHAGO 기준 integrity projection" return "voucher_unmatched", "원천 WEHAGO 전표 복원: 검증된 ERP 후보 없음" def revised_excepted_reason(key: tuple[int, str, str], status_key: str, rows: list[dict[str, Any]], summary: dict[str, Any]) -> str: if status_key not in {"voucher_unmatched", "voucher_recheck"}: return "" if status_key == "voucher_recheck" and key in EXCEPTED_RETAIN_RECHECK_KEYS: return "" text = " ".join( [ clean(summary.get("ledger_accounts")), clean(summary.get("ledger_vendors")), clean(summary.get("review_reason")), " ".join(clean(row.get("ledger_desc")) for row in rows), " ".join(clean(row.get("ledger_account_name")) for row in rows), ] ) compact = text.replace(" ", "") if re.search(r"상계\s*\d+(?:-\d+)?\s*소구", text) or re.search(r"상계\d+(?:-\d+)?소구", compact): return "" tax_work_tokens = ( "부가세예수금과상계", "부가세대급금과상계", "부가예수금과상계", "부가대급금과상계", ) if any(token in compact for token in tax_work_tokens): return "" rules = ( ("WEHAGO_EXCEPTED_COST_SUBSTITUTION", (r"원가.*대체",)), ("WEHAGO_EXCEPTED_GOV_GRANT_OFFSET", (r"국고보조금과상계",)), ("WEHAGO_EXCEPTED_EMPLOYEE_PORTION_SUBSTITUTION", (r"본인분대체",)), ("WEHAGO_EXCEPTED_OTHER_ACCOUNT_SUBSTITUTION", (r"타계정.*대체",)), ( "WEHAGO_EXCEPTED_ACCOUNT_SUBSTITUTION", ( r"계정대체", r"계정.*대체", r"거래처대체", r"거래처.*대체", r"대체분개", r"전기대체", r"전년대체", r"전년도대체", r"결산대체", r"감사대체", r"환입대체", r"계상분대체", ), ), ( "WEHAGO_EXCEPTED_RETAINED_EARNINGS", (r"잉여금.*대체", r"미처분이익잉여금", r"이월이익잉여금", r"차기이월"), ), ("WEHAGO_EXCEPTED_CARRY_FORWARD", (r"전기이월", r"기초이월", r"이월")), ("WEHAGO_EXCEPTED_VALUATION_GAIN_LOSS", (r"평가이익", r"평가손실", r"외화평가")), ("WEHAGO_EXCEPTED_WIP_CLOSING", (r"미완성공사도급",)), ("WEHAGO_EXCEPTED_PROFIT_LOSS_TRANSFER", (r"손익.*대체", r"당기순손익.*대체", r"수익에서대체")), ) for reason, patterns in rules: if any(re.search(pattern, compact) for pattern in patterns): return reason return "" def summarize(key: tuple[int, str, str], status_key: str, rows: list[dict[str, Any]], reason: str) -> dict[str, Any]: ledger_accounts = sorted({clean(row.get("ledger_account_name")) for row in rows if clean(row.get("ledger_account_name"))}) voucher_accounts = sorted({clean(row.get("voucher_account_name")) for row in rows if clean(row.get("voucher_account_name"))}) ledger_vendors = sorted({clean(row.get("ledger_vendor")) for row in rows if clean(row.get("ledger_vendor"))}) voucher_vendors = sorted({clean(row.get("voucher_vendor")) for row in rows if clean(row.get("voucher_vendor"))}) drafts = sorted({clean(row.get("draft_no")) for row in rows if clean(row.get("draft_no"))}) ledger_basis = [row for row in rows if row_has_wehago(row)] erp_basis = [row for row in rows if row_has_erp(row)] return { "fiscal_year": key[0], "status_label": STATUS_LABELS.get(status_key, status_key), "ledger_date": key[1], "ledger_date_key": date_key(key[0], key[1]), "proof_date": "", "voucher_no": key[2], "draft_no": ", ".join(drafts), "ledger_row_count": len(ledger_basis), "voucher_row_count": len(erp_basis), "ledger_debit": sum(amount(row.get("ledger_debit")) for row in ledger_basis), "ledger_credit": sum(amount(row.get("ledger_credit")) for row in ledger_basis), "voucher_debit": sum(amount(row.get("voucher_debit")) for row in erp_basis), "voucher_credit": sum(amount(row.get("voucher_credit")) for row in erp_basis), "ledger_accounts": ", ".join(ledger_accounts), "voucher_accounts": ", ".join(voucher_accounts), "ledger_vendors": ", ".join(ledger_vendors), "voucher_vendors": ", ".join(voucher_vendors), "review_reason": reason, } def get_signatures(conn: sqlite3.Connection) -> dict[str, str]: result: dict[str, str] = {} for status in WEHAGO_STATUSES: row = conn.execute( """ SELECT signature FROM wehago_status_projection_groups WHERE start_year = ? AND end_year = ? AND status_key = ? LIMIT 1 """, (START_YEAR, END_YEAR, status), ).fetchone() result[status] = clean(row["signature"] if row else "") or f"integrity:{status}:{START_YEAR}:{END_YEAR}" return result def json_payload(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def store_setting(conn: sqlite3.Connection, status_key: str, signature: str, count: int) -> None: setting_key = f"wehago_active_status_projection:{status_key}:{START_YEAR}:{END_YEAR}" payload = { "signature": signature, "status_key": status_key, "start_year": START_YEAR, "end_year": END_YEAR, "total_count": count, "projection_version": "integrity-2025-20260629", } conn.execute( """ INSERT INTO wehago_compare_settings (setting_key, setting_json, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) ON CONFLICT(setting_key) DO UPDATE SET setting_json = excluded.setting_json, updated_at = CURRENT_TIMESTAMP """, (setting_key, json_payload(payload)), ) def cleanup_runtime(conn: sqlite3.Connection) -> None: conn.execute("DELETE FROM wehago_status_projection_groups WHERE start_year = ? AND end_year = ?", (START_YEAR, END_YEAR)) for table in ( "wehago_compare_query_groups", "wehago_compare_query_rows", "wehago_compare_query_metrics", "wehago_compare_query_page_cache", "wehago_compare_final_status_projection", "wehago_compare_export_jobs", ): conn.execute(f"DELETE FROM {table} WHERE start_year = ? AND end_year = ?", (START_YEAR, END_YEAR)) conn.execute("DELETE FROM wehago_compare_settings WHERE setting_key = ?", (f"wehago_active_query_projection:{START_YEAR}:{END_YEAR}",)) conn.execute("DELETE FROM wehago_compare_settings WHERE setting_key = ?", (f"wehago_active_status_projection_run:{START_YEAR}:{END_YEAR}",)) conn.execute("DELETE FROM wehago_compare_settings WHERE setting_key LIKE ?", (f"wehago_active_status_projection:%:{START_YEAR}:{END_YEAR}",)) conn.execute("DELETE FROM wehago_compare_settings WHERE setting_key = ?", (f"wehago_final_basis_projection:{START_YEAR}:{END_YEAR}",)) def materialize(conn: sqlite3.Connection) -> dict[str, Any]: seed_by_key = load_current_projection(conn) raw_wehago_by_key = load_raw_wehago(conn) raw_erp_by_base = load_raw_erp(conn) signatures = get_signatures(conn) raw_keys = sorted(raw_wehago_by_key) rows_to_insert: dict[str, list[tuple[Any, ...]]] = {status: [] for status in WEHAGO_STATUSES} diagnostics = { "raw_wehago_vouchers": len(raw_keys), "seed_canonical_vouchers": len(seed_by_key), "seed_duplicates_removed": 0, "restored_missing_vouchers": len(set(raw_keys) - set(seed_by_key)), "status_moves": [], } for key in raw_keys: seed = seed_by_key.get(key) old_status = clean(seed.get("status_key")) if seed else "" status, reason = determine_status(key, seed) seed_rows = seed.get("rows", []) if seed else [] display_rows = hydrate_rows(status, seed_rows, raw_wehago_by_key[key], raw_erp_by_base) summary = summarize(key, status, display_rows, reason) excepted_reason = revised_excepted_reason(key, status, display_rows, summary) if excepted_reason: status = "voucher_excepted" reason = excepted_reason display_rows = hydrate_rows(status, seed_rows, raw_wehago_by_key[key], raw_erp_by_base) summary = summarize(key, status, display_rows, reason) if old_status and old_status != status: diagnostics["status_moves"].append({"key": key, "from": old_status, "to": status, "reason": reason}) search_text = " ".join( [ summary["voucher_no"], summary["draft_no"], summary["ledger_accounts"], summary["voucher_accounts"], summary["ledger_vendors"], summary["voucher_vendors"], reason, " ".join(clean(row.get("ledger_desc")) for row in display_rows), " ".join(clean(row.get("voucher_desc")) for row in display_rows), ] ) rows_to_insert[status].append( ( START_YEAR, END_YEAR, status, signatures[status], 0, summary["fiscal_year"], summary["ledger_date"], summary["proof_date"], summary["voucher_no"], summary["draft_no"], int(summary["ledger_row_count"]), int(summary["voucher_row_count"]), amount(summary["ledger_debit"]), amount(summary["ledger_credit"]), amount(summary["voucher_debit"]), amount(summary["voucher_credit"]), summary["ledger_accounts"], summary["voucher_accounts"], summary["ledger_vendors"], summary["voucher_vendors"], reason, search_text, json_payload(summary), json_payload(display_rows), ) ) cleanup_runtime(conn) counts: dict[str, int] = {} for status in WEHAGO_STATUSES: payloads = [] for idx, payload in enumerate(rows_to_insert[status]): values = list(payload) values[4] = idx payloads.append(tuple(values)) conn.executemany( """ INSERT INTO wehago_status_projection_groups ( start_year, end_year, status_key, signature, 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, search_text, summary_json, rows_json, created_at, updated_at ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP ) """, payloads, ) counts[status] = len(payloads) store_setting(conn, status, signatures[status], len(payloads)) basis = { "source": "raw WEHAGO canonical voucher set; current projection only used as candidate seed", "counts": counts, "wehago_total": sum(counts.values()), "diagnostics": diagnostics, } conn.execute( """ INSERT INTO wehago_compare_settings (setting_key, setting_json, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) ON CONFLICT(setting_key) DO UPDATE SET setting_json = excluded.setting_json, updated_at = CURRENT_TIMESTAMP """, (f"wehago_final_basis_projection:{START_YEAR}:{END_YEAR}", json_payload(basis)), ) return basis def main() -> None: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row try: conn.execute("PRAGMA busy_timeout = 120000") conn.execute("BEGIN IMMEDIATE") basis = materialize(conn) conn.commit() except Exception: conn.rollback() raise finally: conn.close() print(json_payload(basis)) if __name__ == "__main__": main()