HM_project_Viewer_Board/scripts/import_hanmac_voucher_xlsx.py

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

from openpyxl import load_workbook
from sqlalchemy import create_engine, text

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from runtime_config import DB_PATH
from wehago_compare import activate_erp_voucher_existence_snapshot, detect_file_kind, import_voucher_rows
from scripts.redownload_and_fix_hanmac_ledger_account import upsert_fix_source_file


def import_voucher_file(
    db_path: Path,
    source_path: Path,
    *,
    year_hint: int | None = None,
    activate_existence_snapshot: bool = False,
) -> dict[str, Any]:
    file_kind, header, sheet_name = detect_file_kind(source_path)
    if file_kind != "voucher":
        raise ValueError(f"전표내역조회 형식이 아닙니다: {source_path} ({file_kind})")

    workbook = load_workbook(source_path, read_only=True, data_only=True)
    engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False})
    try:
        sheet = workbook.worksheets[0]
        with engine.begin() as conn:
            conn.execute(text("PRAGMA busy_timeout = 300000"))
            source_id, _changed = upsert_fix_source_file(
                conn,
                source_path.resolve(),
                file_kind,
                int(year_hint) if year_hint else None,
                sheet_name,
                header,
            )
            deleted = int(
                conn.execute(
                    text("DELETE FROM wehago_voucher_rows WHERE source_file_id = :source_id"),
                    {"source_id": source_id},
                ).rowcount
                or 0
            )
            inserted = import_voucher_rows(
                conn,
                source_id,
                sheet_name,
                sheet.iter_rows(min_row=2, values_only=True),
                int(year_hint) if year_hint else None,
            )
            conn.execute(
                text(
                    """
                    UPDATE wehago_source_files
                    SET row_count = :row_count,
                        imported_at = CURRENT_TIMESTAMP
                    WHERE id = :source_id
                    """
                ),
                {"row_count": inserted, "source_id": source_id},
            )
            existence_snapshot = None
            if activate_existence_snapshot:
                snapshot_year = int(year_hint or 0)
                if snapshot_year <= 0:
                    year_values = [
                        int(row["fiscal_year"])
                        for row in conn.execute(
                            text(
                                """
                                SELECT DISTINCT fiscal_year
                                FROM wehago_voucher_rows
                                WHERE source_file_id = :source_id
                                  AND fiscal_year IS NOT NULL
                                ORDER BY fiscal_year
                                """
                            ),
                            {"source_id": source_id},
                        ).mappings()
                        if int(row["fiscal_year"] or 0) > 0
                    ]
                    if len(year_values) == 1:
                        snapshot_year = year_values[0]
                if snapshot_year <= 0:
                    raise ValueError("--activate-existence-snapshot requires --year-hint or a single fiscal year in the file.")
                existence_snapshot = activate_erp_voucher_existence_snapshot(
                    conn,
                    snapshot_year,
                    source_file_id=source_id,
                    source_label=source_path.name,
                    snapshot_mode="full",
                )
            year_rows = [
                dict(row)
                for row in conn.execute(
                    text(
                        """
                        SELECT fiscal_year, COUNT(*) AS row_count
                        FROM wehago_voucher_rows
                        WHERE source_file_id = :source_id
                        GROUP BY fiscal_year
                        ORDER BY fiscal_year
                        """
                    ),
                    {"source_id": source_id},
                ).mappings()
            ]
    finally:
        workbook.close()
        engine.dispose()

    return {
        "db_path": str(db_path),
        "source_path": str(source_path.resolve()),
        "source_id": source_id,
        "deleted_rows": deleted,
        "inserted_rows": inserted,
        "year_rows": year_rows,
        "existence_snapshot": existence_snapshot,
        "imported_at": datetime.now().isoformat(timespec="seconds"),
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="한맥 ERP 전표내역조회 엑셀을 WEHAGO 비교 DB의 voucher 테이블에 반영합니다.")
    parser.add_argument("source", type=Path)
    parser.add_argument("--db", type=Path, default=DB_PATH)
    parser.add_argument("--year-hint", type=int)
    parser.add_argument(
        "--activate-existence-snapshot",
        action="store_true",
        help="이 파일을 해당 연도의 최신 ERP 전표 존재 목록으로 활성화합니다.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    result = import_voucher_file(
        args.db,
        args.source,
        year_hint=args.year_hint,
        activate_existence_snapshot=args.activate_existence_snapshot,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


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