HM_project_Viewer_Board/scripts/fix_hanmac_2025_137_minimal.py

from __future__ import annotations

import sqlite3
from datetime import date, datetime
from pathlib import Path
from typing import Any

from openpyxl import load_workbook

ROOT = Path(__file__).resolve().parents[1]
DB_PATH = ROOT / "data.db"
SOURCE_XLSX = ROOT / "reports" / "wehago_account_fixes" / "2025_137_20260617_092619" / "137_주.종단기채권.xlsx"
YEAR = 2025
ACCOUNT_CODE = "137"
ACCOUNT_NAME = "주.종단기채권"


def clean(value: Any) -> str:
    return "" if value is None else str(value).strip()


def parse_amount(value: Any) -> float:
    if value is None or value == "":
        return 0.0
    if isinstance(value, (int, float)):
        return float(value)
    text = clean(value).replace(",", "")
    try:
        return float(text)
    except ValueError:
        return 0.0


def parse_ledger_date(value: Any) -> str | None:
    if value is None or clean(value) == "":
        return None
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    text = clean(value).replace(".", "-").replace("/", "-")
    parts = [part for part in text.split("-") if part]
    if len(parts) == 2:
        month, day = parts
        return f"{YEAR}-{int(month):02d}-{int(day):02d}"
    if len(parts) == 3:
        year, month, day = parts
        return f"{int(year):04d}-{int(month):02d}-{int(day):02d}"
    return text


def compare_voucher_no(ledger_date: str | None, voucher_no: str) -> str:
    if not ledger_date or not voucher_no:
        return ""
    return f"{ledger_date.replace('-', '')}-{voucher_no}"


def normalize_text(value: Any) -> str:
    return " ".join(clean(value).split())


def main() -> int:
    wb = load_workbook(SOURCE_XLSX, read_only=True, data_only=True)
    try:
        ws = wb.worksheets[0]
        source_rows = list(ws.iter_rows(min_row=2, values_only=True))
    finally:
        wb.close()

    conn = sqlite3.connect(DB_PATH, timeout=300)
    try:
        conn.execute("PRAGMA busy_timeout = 300000")
        source_id_row = conn.execute(
            """
            SELECT source_file_id
            FROM wehago_ledger_rows
            WHERE fiscal_year = ? AND account_code = ?
            GROUP BY source_file_id
            ORDER BY COUNT(*) DESC
            LIMIT 1
            """,
            (YEAR, ACCOUNT_CODE),
        ).fetchone()
        source_file_id = int(source_id_row[0]) if source_id_row else 0
        if not source_file_id:
            source_file_id = int(
                conn.execute(
                    "SELECT source_file_id FROM wehago_ledger_rows WHERE fiscal_year = ? LIMIT 1",
                    (YEAR,),
                ).fetchone()[0]
            )

        with conn:
            deleted = conn.execute(
                "DELETE FROM wehago_ledger_rows WHERE fiscal_year = ? AND account_code = ?",
                (YEAR, ACCOUNT_CODE),
            ).rowcount
            inserted = 0
            for row_number, values in enumerate(source_rows, start=2):
                if not any(clean(item) for item in values):
                    continue
                ledger_date = parse_ledger_date(values[0] if len(values) > 0 else None)
                description = clean(values[1] if len(values) > 1 else "")
                vendor_name = clean(values[2] if len(values) > 2 else "")
                debit = parse_amount(values[3] if len(values) > 3 else 0)
                credit = parse_amount(values[4] if len(values) > 4 else 0)
                balance = parse_amount(values[5] if len(values) > 5 else 0)
                voucher_no = clean(values[6] if len(values) > 6 else "")
                account_code = clean(values[7] if len(values) > 7 else ACCOUNT_CODE) or ACCOUNT_CODE
                account_name = clean(values[8] if len(values) > 8 else ACCOUNT_NAME) or ACCOUNT_NAME
                conn.execute(
                    """
                    INSERT INTO wehago_ledger_rows (
                        source_file_id, sheet_name, row_number, ledger_date, description,
                        vendor_name, debit, credit, balance, voucher_no, account_code,
                        account_name, compare_voucher_no, compare_amount, compare_side,
                        compare_vendor, compare_desc, fiscal_year
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        source_file_id,
                        ws.title,
                        row_number,
                        ledger_date,
                        description,
                        vendor_name,
                        debit,
                        credit,
                        balance,
                        voucher_no,
                        account_code,
                        account_name,
                        compare_voucher_no(ledger_date, voucher_no),
                        debit if debit else credit,
                        "debit" if debit else ("credit" if credit else ""),
                        normalize_text(vendor_name),
                        normalize_text(description),
                        YEAR,
                    ),
                )
                inserted += 1
        print(f"source={SOURCE_XLSX}")
        print(f"source_file_id={source_file_id}")
        print(f"deleted={deleted}")
        print(f"inserted={inserted}")
    finally:
        conn.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())