from __future__ import annotations

import hashlib
import json
import re
import threading
import time
from dataclasses import dataclass
from datetime import date, datetime
from functools import lru_cache
from pathlib import Path
from typing import Any, Iterable

from openpyxl import load_workbook
from sqlalchemy import text

WEHAGO_SOURCE_ROOT = Path(Path(__file__).resolve().parent.parent / "WEHAGO_DB")

VOUCHER_HEADERS = [
    "결재상태",
    "가전표번호",
    "계정코드",
    "계정명칭",
    "차변공급가",
    "차변부가세",
    "대변공급가",
    "대변부가세",
    "발의부서코드",
    "발의부서명칭",
    "확정전표번호",
    "지원부서코드",
    "지원부서명칭",
    "원가부서코드",
    "원가부서명칭",
    "적요1",
    "적요2",
    "거래처코드",
    "거래처명칭",
    "세무코드",
    "증빙일자",
    "전표종류",
    "관리항목",
]

LEDGER_HEADERS = [
    "일자",
    "적요",
    "거래처",
    "차변",
    "대변",
    "잔액",
    "전표번호",
    "계정코드",
    "계정명",
]

STATUS_META = [
    ("matched", "Matched", "WEHAGO 기준으로 정상 매칭된 항목"),
    ("ledger_only", "Unmatched", "WEHAGO에는 있으나 ERP와 연결되지 않은 항목"),
    ("voucher_only", "ERP Unmatched", "ERP 기준으로 매치되지 않은 항목"),
    ("amount_mismatch", "Recheck", "전표번호는 같지만 금액이 달라 다시 확인이 필요한 항목"),
]

DETAIL_COLUMN_MAP = {
    "matched": [
        ("fiscal_year", "연도"),
        ("ledger_date", "일자"),
        ("voucher_no", "전표번호"),
        ("draft_no", "가전표번호"),
        ("ledger_account_name", "계정"),
        ("voucher_account_name", "ERP 계정"),
        ("ledger_vendor", "거래처"),
        ("voucher_vendor", "ERP 거래처"),
        ("ledger_debit", "차변"),
        ("ledger_credit", "대변"),
        ("voucher_debit", "ERP 차변"),
        ("voucher_credit", "ERP 대변"),
        ("ledger_desc", "WEHAGO 적요"),
        ("voucher_desc", "ERP 적요"),
    ],
    "ledger_only": [
        ("fiscal_year", "연도"),
        ("voucher_no", "전표번호"),
        ("ledger_date", "일자"),
        ("ledger_account_name", "WEHAGO 계정"),
        ("ledger_vendor", "WEHAGO 거래처"),
        ("ledger_debit", "차변"),
        ("ledger_credit", "대변"),
        ("ledger_desc", "적요"),
    ],
    "amount_mismatch": [
        ("fiscal_year", "연도"),
        ("ledger_date", "일자"),
        ("voucher_no", "전표번호"),
        ("draft_no", "가전표번호"),
        ("ledger_account_name", "계정"),
        ("voucher_account_name", "ERP 계정"),
        ("ledger_vendor", "거래처"),
        ("voucher_vendor", "ERP 거래처"),
        ("ledger_debit", "차변"),
        ("ledger_credit", "대변"),
        ("voucher_debit", "ERP 차변"),
        ("voucher_credit", "ERP 대변"),
        ("ledger_desc", "WEHAGO 적요"),
        ("voucher_desc", "ERP 적요"),
    ],
    "voucher_only": [
        ("fiscal_year", "연도"),
        ("voucher_no", "전표번호"),
        ("proof_date", "증빙일자"),
        ("voucher_account_name", "ERP 계정"),
        ("voucher_vendor", "ERP 거래처"),
        ("voucher_debit", "차변"),
        ("voucher_credit", "대변"),
        ("voucher_desc", "적요"),
    ],
}

WEHAGO_COLUMNS = [
    ("ledger_date", "일자"),
    ("voucher_no", "전표번호"),
    ("account_code", "계정코드"),
    ("account_name", "계정명"),
    ("vendor_name", "거래처"),
    ("description", "적요"),
    ("debit", "차변"),
    ("credit", "대변"),
]

ERP_COLUMNS = [
    ("proof_date", "증빙일자"),
    ("confirmed_no", "확정전표번호"),
    ("draft_no", "가전표번호"),
    ("account_code", "계정코드"),
    ("account_name", "계정명"),
    ("vendor_name", "거래처"),
    ("desc1", "적요1"),
    ("desc2", "적요2"),
    ("debit_supply", "차변공급가"),
    ("credit_supply", "대변공급가"),
]

_DASHBOARD_CACHE: dict[str, dict[str, Any]] = {}
_DASHBOARD_CACHE_TTL_SEC = 20
_SUGGEST_CACHE: dict[str, dict[str, Any]] = {}
_SUGGEST_CACHE_TTL_SEC = 20
_STATUS_ROWS_CACHE: dict[str, dict[str, Any]] = {}
_STATUS_ROWS_CACHE_TTL_SEC = 20
_STATUS_CACHE_WARMING: set[str] = set()
_STATUS_CACHE_WARMING_LOCK = threading.Lock()


def _fast_metric_counts_from_result_files(conn: Any, start_year: int | None, end_year: int | None) -> dict[str, int]:
    counts = {"matched": 0, "ledger_only": 0, "voucher_only": 0, "amount_mismatch": 0}
    if start_year is None or end_year is None:
        return counts

    for year in range(start_year, end_year + 1):
        bundle = discover_compare_result_bundle(year)
        if not bundle:
            continue

        ledger_wb = load_workbook(bundle["ledger_result"], read_only=True, data_only=True)
        ledger_ws = ledger_wb.worksheets[0]
        ledger_header = [clean(v) for v in next(ledger_ws.iter_rows(min_row=1, max_row=1, values_only=True))]
        ledger_idx = {name: idx for idx, name in enumerate(ledger_header)}
        matched_col = ledger_idx.get("matched_확정전표번호")
        if matched_col is not None:
            for row in ledger_ws.iter_rows(min_row=2, values_only=True):
                matched_no = clean(row[matched_col]) if matched_col < len(row) else ""
                if matched_no:
                    counts["matched"] += 1
                else:
                    counts["ledger_only"] += 1

        voucher_wb = load_workbook(bundle["voucher_result"], read_only=True, data_only=True)
        voucher_ws = voucher_wb.worksheets[0]
        voucher_header = [clean(v) for v in next(voucher_ws.iter_rows(min_row=1, max_row=1, values_only=True))]
        voucher_idx = {name: idx for idx, name in enumerate(voucher_header)}
        matched_col = voucher_idx.get("matched_전표번호")
        reason_col = voucher_idx.get("matched_검증근거")
        flag_col = voucher_idx.get("review_flag")
        for row in voucher_ws.iter_rows(min_row=2, values_only=True):
            matched_no = clean(row[matched_col]) if (matched_col is not None and matched_col < len(row)) else ""
            if not matched_no:
                counts["voucher_only"] += 1
                continue
            review_reason = clean(row[reason_col]) if (reason_col is not None and reason_col < len(row)) else ""
            review_flag = row[flag_col] if (flag_col is not None and flag_col < len(row)) else ""
            if _normalize_flag(review_flag) or ("REVIEW" in review_reason):
                counts["amount_mismatch"] += 1

    reviewed_count = int(
        conn.execute(
            text(
                """
                SELECT COUNT(*)
                FROM wehago_recheck_reviews
                WHERE (:start_year IS NULL OR fiscal_year >= :start_year)
                  AND (:end_year IS NULL OR fiscal_year <= :end_year)
                """
            ),
            {"start_year": start_year, "end_year": end_year},
        ).scalar_one()
    )
    pair_count = int(
        conn.execute(
            text(
                """
                SELECT COUNT(*)
                FROM wehago_manual_pair_matches
                WHERE (:start_year IS NULL OR fiscal_year >= :start_year)
                  AND (:end_year IS NULL OR fiscal_year <= :end_year)
                """
            ),
            {"start_year": start_year, "end_year": end_year},
        ).scalar_one()
    )
    counts["matched"] = max(0, counts["matched"] + reviewed_count + pair_count)
    counts["amount_mismatch"] = max(0, counts["amount_mismatch"] - reviewed_count)
    counts["ledger_only"] = max(0, counts["ledger_only"] - pair_count)
    counts["voucher_only"] = max(0, counts["voucher_only"] - pair_count)
    return counts


@dataclass
class WehagoImportSummary:
    scanned_files: int = 0
    imported_files: int = 0
    skipped_files: int = 0
    voucher_rows: int = 0
    ledger_rows: int = 0
    comparison_rows: int = 0


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


def normalize_text(value: Any) -> str:
    text_value = clean(value).lower()
    text_value = re.sub(r"\s+", "", text_value)
    return re.sub(r"[\(\)\[\]\{\},._\-/\\:;*×%◇]", "", text_value)


def normalize_voucher_no(value: Any) -> str:
    return clean(value).replace(" ", "")


def parse_amount(value: Any) -> float:
    text_value = clean(value).replace(",", "")
    if not text_value:
        return 0.0
    try:
        return float(text_value)
    except ValueError:
        return 0.0


def format_amount(value: Any) -> str:
    amount = parse_amount(value)
    if abs(amount - round(amount)) < 0.000001:
        return f"{int(round(amount)):,}"
    return f"{amount:,.2f}"


def parse_excel_date(value: Any, default_year: int | None = None) -> str | None:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    text_value = clean(value).replace(".0", "")
    for pattern in (r"^(\d{4})(\d{2})(\d{2})$", r"^(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})$"):
        match = re.match(pattern, text_value)
        if match:
            return f"{int(match.group(1)):04d}-{int(match.group(2)):02d}-{int(match.group(3)):02d}"
    match = re.match(r"^(\d{1,2})[-/.](\d{1,2})$", text_value)
    if match and default_year:
        return f"{default_year:04d}-{int(match.group(1)):02d}-{int(match.group(2)):02d}"
    return None


def year_from_date_text(value: str | None) -> int | None:
    if not value:
        return None
    match = re.match(r"^(\d{4})-", value)
    return int(match.group(1)) if match else None


def compute_file_hash(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def detect_file_kind(path: Path) -> tuple[str | None, list[str], str]:
    workbook = load_workbook(path, read_only=True, data_only=True)
    sheet = workbook.worksheets[0]
    header = [clean(item) for item in next(sheet.iter_rows(min_row=1, max_row=1, values_only=True))]
    if header[: len(VOUCHER_HEADERS)] == VOUCHER_HEADERS:
        return "voucher", header, sheet.title
    if header[: len(LEDGER_HEADERS)] == LEDGER_HEADERS:
        return "ledger", header, sheet.title
    return None, header, sheet.title


def iter_voucher_files(source_root: Path) -> Iterable[Path]:
    for pattern in ("*.xlsx", "*.xlsm"):
        for path in sorted(source_root.glob(pattern)):
            if path.name.startswith("~$"):
                continue
            yield path


def iter_ledger_files(source_root: Path) -> Iterable[Path]:
    data_download_dir = source_root / "data_download"
    if not data_download_dir.exists():
        return
    for year_dir in sorted(
        path for path in data_download_dir.iterdir() if path.is_dir() and path.name.isdigit() and len(path.name) == 4
    ):
        for pattern in ("*.xlsx", "*.xlsm"):
            for path in sorted(year_dir.glob(pattern)):
                if path.name.startswith("~$"):
                    continue
                yield path


def iter_source_files(source_root: Path) -> Iterable[Path]:
    seen: set[Path] = set()
    for path in iter_voucher_files(source_root):
        if path not in seen:
            seen.add(path)
            yield path
    for path in iter_ledger_files(source_root):
        if path not in seen:
            seen.add(path)
            yield path


def infer_year_hint(path: Path, file_kind: str, sample_rows: list[tuple[Any, ...]]) -> int | None:
    for part in reversed(path.parts):
        if part.isdigit() and len(part) == 4:
            return int(part)
    match = re.search(r"(19|20)\d{2}", path.name)
    if match:
        return int(match.group(0))
    if file_kind == "voucher":
        for row in sample_rows[:50]:
            proof_date = parse_excel_date(row[20] if len(row) > 20 else None)
            year_value = year_from_date_text(proof_date)
            if year_value:
                return year_value
        for row in sample_rows[:50]:
            voucher_no = normalize_voucher_no(row[10] if len(row) > 10 else row[1] if len(row) > 1 else "")
            match = re.search(r"-(\d{4})\d{4}-", voucher_no)
            if match:
                return int(match.group(1))
    if file_kind == "ledger":
        for row in sample_rows[:50]:
            date_text = parse_excel_date(row[0] if row else None)
            year_value = year_from_date_text(date_text)
            if year_value:
                return year_value
    return None


def table_columns(conn: Any, table_name: str) -> set[str]:
    return {row[1] for row in conn.execute(text(f"PRAGMA table_info({table_name})")).fetchall()}


def ensure_column(conn: Any, table_name: str, column_name: str, column_sql: str) -> None:
    if column_name in table_columns(conn, table_name):
        return
    conn.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_sql}"))


def init_wehago_compare_db(engine: Any) -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_source_files (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    file_kind TEXT NOT NULL CHECK (file_kind IN ('voucher', 'ledger')),
                    file_path TEXT NOT NULL UNIQUE,
                    file_name TEXT NOT NULL,
                    relative_path TEXT NOT NULL,
                    year_hint INTEGER,
                    sheet_name TEXT,
                    file_size INTEGER NOT NULL,
                    modified_ts REAL NOT NULL,
                    file_hash TEXT NOT NULL,
                    header_json TEXT NOT NULL,
                    row_count INTEGER NOT NULL DEFAULT 0,
                    imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_voucher_rows (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    source_file_id INTEGER NOT NULL REFERENCES wehago_source_files(id) ON DELETE CASCADE,
                    sheet_name TEXT NOT NULL,
                    row_number INTEGER NOT NULL,
                    approval_status TEXT,
                    draft_no TEXT,
                    confirmed_no TEXT,
                    account_code TEXT,
                    account_name TEXT,
                    debit_supply REAL NOT NULL DEFAULT 0,
                    debit_tax REAL NOT NULL DEFAULT 0,
                    credit_supply REAL NOT NULL DEFAULT 0,
                    credit_tax REAL NOT NULL DEFAULT 0,
                    issue_dept_code TEXT,
                    issue_dept_name TEXT,
                    support_dept_code TEXT,
                    support_dept_name TEXT,
                    cost_dept_code TEXT,
                    cost_dept_name TEXT,
                    desc1 TEXT,
                    desc2 TEXT,
                    vendor_code TEXT,
                    vendor_name TEXT,
                    tax_code TEXT,
                    proof_date TEXT,
                    voucher_type TEXT,
                    management_item TEXT,
                    compare_voucher_no TEXT,
                    compare_amount REAL NOT NULL DEFAULT 0,
                    compare_side TEXT,
                    compare_vendor TEXT,
                    compare_desc TEXT,
                    fiscal_year INTEGER
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_ledger_rows (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    source_file_id INTEGER NOT NULL REFERENCES wehago_source_files(id) ON DELETE CASCADE,
                    sheet_name TEXT NOT NULL,
                    row_number INTEGER NOT NULL,
                    ledger_date TEXT,
                    description TEXT,
                    vendor_name TEXT,
                    debit REAL NOT NULL DEFAULT 0,
                    credit REAL NOT NULL DEFAULT 0,
                    balance REAL NOT NULL DEFAULT 0,
                    voucher_no TEXT,
                    account_code TEXT,
                    account_name TEXT,
                    compare_voucher_no TEXT,
                    compare_amount REAL NOT NULL DEFAULT 0,
                    compare_side TEXT,
                    compare_vendor TEXT,
                    compare_desc TEXT,
                    fiscal_year INTEGER
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_comparison_results (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    fiscal_year INTEGER,
                    voucher_no TEXT NOT NULL,
                    voucher_row_count INTEGER NOT NULL DEFAULT 0,
                    ledger_row_count INTEGER NOT NULL DEFAULT 0,
                    voucher_debit REAL NOT NULL DEFAULT 0,
                    voucher_credit REAL NOT NULL DEFAULT 0,
                    ledger_debit REAL NOT NULL DEFAULT 0,
                    ledger_credit REAL NOT NULL DEFAULT 0,
                    voucher_accounts TEXT NOT NULL DEFAULT '',
                    ledger_accounts TEXT NOT NULL DEFAULT '',
                    voucher_vendors TEXT NOT NULL DEFAULT '',
                    ledger_vendors TEXT NOT NULL DEFAULT '',
                    status TEXT NOT NULL,
                    notes TEXT NOT NULL DEFAULT ''
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_recheck_reviews (
                    review_key TEXT PRIMARY KEY,
                    fiscal_year INTEGER,
                    voucher_no TEXT NOT NULL DEFAULT '',
                    draft_no TEXT NOT NULL DEFAULT '',
                    voucher_account_code TEXT NOT NULL DEFAULT '',
                    voucher_account_name TEXT NOT NULL DEFAULT '',
                    voucher_vendor TEXT NOT NULL DEFAULT '',
                    voucher_desc TEXT NOT NULL DEFAULT '',
                    review_reason TEXT NOT NULL DEFAULT '',
                    review_memo TEXT NOT NULL DEFAULT '',
                    reviewed_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_manual_pair_matches (
                    pair_key TEXT PRIMARY KEY,
                    fiscal_year INTEGER,
                    ledger_row_key TEXT NOT NULL UNIQUE,
                    voucher_row_key TEXT NOT NULL UNIQUE,
                    ledger_voucher_no TEXT NOT NULL DEFAULT '',
                    ledger_account_code TEXT NOT NULL DEFAULT '',
                    ledger_account_name TEXT NOT NULL DEFAULT '',
                    ledger_vendor TEXT NOT NULL DEFAULT '',
                    ledger_debit REAL NOT NULL DEFAULT 0,
                    ledger_credit REAL NOT NULL DEFAULT 0,
                    ledger_desc TEXT NOT NULL DEFAULT '',
                    voucher_no TEXT NOT NULL DEFAULT '',
                    draft_no TEXT NOT NULL DEFAULT '',
                    voucher_account_code TEXT NOT NULL DEFAULT '',
                    voucher_account_name TEXT NOT NULL DEFAULT '',
                    voucher_vendor TEXT NOT NULL DEFAULT '',
                    voucher_debit REAL NOT NULL DEFAULT 0,
                    voucher_credit REAL NOT NULL DEFAULT 0,
                    voucher_desc TEXT NOT NULL DEFAULT '',
                    match_source TEXT NOT NULL DEFAULT 'manual',
                    confidence_score REAL NOT NULL DEFAULT 0,
                    confidence_level TEXT NOT NULL DEFAULT '',
                    match_reason TEXT NOT NULL DEFAULT '',
                    pair_note TEXT NOT NULL DEFAULT '',
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_action_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    action_type TEXT NOT NULL,
                    payload_json TEXT NOT NULL,
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS wehago_result_row_cache (
                    fiscal_year INTEGER NOT NULL,
                    ledger_result_path TEXT NOT NULL,
                    ledger_result_mtime REAL NOT NULL,
                    voucher_result_path TEXT NOT NULL,
                    voucher_result_mtime REAL NOT NULL,
                    payload_json TEXT NOT NULL,
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (
                        fiscal_year,
                        ledger_result_path,
                        ledger_result_mtime,
                        voucher_result_path,
                        voucher_result_mtime
                    )
                )
                """
            )
        )
        ensure_column(conn, "wehago_source_files", "source_origin", "TEXT NOT NULL DEFAULT 'filesystem'")
        ensure_column(conn, "wehago_manual_pair_matches", "match_source", "TEXT NOT NULL DEFAULT 'manual'")
        ensure_column(conn, "wehago_manual_pair_matches", "confidence_score", "REAL NOT NULL DEFAULT 0")
        ensure_column(conn, "wehago_manual_pair_matches", "confidence_level", "TEXT NOT NULL DEFAULT ''")
        ensure_column(conn, "wehago_manual_pair_matches", "match_reason", "TEXT NOT NULL DEFAULT ''")
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_voucher_source ON wehago_voucher_rows(source_file_id)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_voucher_year_vno ON wehago_voucher_rows(fiscal_year, compare_voucher_no)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_ledger_source ON wehago_ledger_rows(source_file_id)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_ledger_year_vno ON wehago_ledger_rows(fiscal_year, compare_voucher_no)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_compare_year_status ON wehago_comparison_results(fiscal_year, status)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_source_origin_kind ON wehago_source_files(source_origin, file_kind)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_recheck_reviews_year ON wehago_recheck_reviews(fiscal_year, reviewed_at)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_manual_pairs_year ON wehago_manual_pair_matches(fiscal_year, created_at)"))
        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_wehago_action_history_created ON wehago_action_history(created_at)"))


def upsert_source_file(
    conn: Any,
    path: Path,
    file_kind: str,
    year_hint: int | None,
    sheet_name: str,
    header: list[str],
) -> tuple[int, bool]:
    row = conn.execute(
        text(
            """
            SELECT id, file_hash, modified_ts
            FROM wehago_source_files
            WHERE file_path = :file_path
            """
        ),
        {"file_path": str(path)},
    ).mappings().first()
    file_hash = compute_file_hash(path)
    modified_ts = path.stat().st_mtime
    payload = {
        "file_kind": file_kind,
        "file_path": str(path),
        "file_name": path.name,
        "relative_path": str(path.relative_to(WEHAGO_SOURCE_ROOT)),
        "year_hint": year_hint,
        "sheet_name": sheet_name,
        "file_size": path.stat().st_size,
        "modified_ts": modified_ts,
        "file_hash": file_hash,
        "header_json": str(header),
        "source_origin": "filesystem",
    }
    if row is None:
        result = conn.execute(
            text(
                """
                INSERT INTO wehago_source_files (
                    file_kind, file_path, file_name, relative_path, year_hint, sheet_name,
                    file_size, modified_ts, file_hash, header_json, source_origin
                ) VALUES (
                    :file_kind, :file_path, :file_name, :relative_path, :year_hint, :sheet_name,
                    :file_size, :modified_ts, :file_hash, :header_json, :source_origin
                )
                """
            ),
            payload,
        )
        return int(result.lastrowid), True

    changed = row["file_hash"] != file_hash or float(row["modified_ts"]) != float(modified_ts)
    conn.execute(
        text(
            """
            UPDATE wehago_source_files
            SET file_kind = :file_kind,
                file_name = :file_name,
                relative_path = :relative_path,
                year_hint = :year_hint,
                sheet_name = :sheet_name,
                file_size = :file_size,
                modified_ts = :modified_ts,
                file_hash = :file_hash,
                header_json = :header_json,
                source_origin = :source_origin,
                imported_at = CURRENT_TIMESTAMP
            WHERE id = :id
            """
        ),
        {**payload, "id": row["id"]},
    )
    return int(row["id"]), changed


def build_voucher_signature(values: list[Any], year_hint: int | None) -> str:
    proof_date = parse_excel_date(values[20] if len(values) > 20 else None)
    fiscal_year = year_from_date_text(proof_date) or year_hint or ""
    parts = [
        clean(values[0] if len(values) > 0 else ""),
        clean(values[1] if len(values) > 1 else ""),
        clean(values[10] if len(values) > 10 else ""),
        clean(values[2] if len(values) > 2 else ""),
        clean(values[3] if len(values) > 3 else ""),
        f"{parse_amount(values[4] if len(values) > 4 else 0):.2f}",
        f"{parse_amount(values[5] if len(values) > 5 else 0):.2f}",
        f"{parse_amount(values[6] if len(values) > 6 else 0):.2f}",
        f"{parse_amount(values[7] if len(values) > 7 else 0):.2f}",
        clean(values[8] if len(values) > 8 else ""),
        clean(values[11] if len(values) > 11 else ""),
        clean(values[13] if len(values) > 13 else ""),
        clean(values[15] if len(values) > 15 else ""),
        clean(values[16] if len(values) > 16 else ""),
        clean(values[17] if len(values) > 17 else ""),
        clean(values[18] if len(values) > 18 else ""),
        clean(values[19] if len(values) > 19 else ""),
        proof_date or "",
        clean(values[21] if len(values) > 21 else ""),
        clean(values[22] if len(values) > 22 else ""),
        str(fiscal_year),
    ]
    return hashlib.sha1("|".join(parts).encode("utf-8", "ignore")).hexdigest()


def build_voucher_signature_from_mapping(row: dict[str, Any]) -> str:
    parts = [
        clean(row.get("approval_status")),
        clean(row.get("draft_no")),
        clean(row.get("confirmed_no")),
        clean(row.get("account_code")),
        clean(row.get("account_name")),
        f"{parse_amount(row.get('debit_supply')):.2f}",
        f"{parse_amount(row.get('debit_tax')):.2f}",
        f"{parse_amount(row.get('credit_supply')):.2f}",
        f"{parse_amount(row.get('credit_tax')):.2f}",
        clean(row.get("issue_dept_code")),
        clean(row.get("support_dept_code")),
        clean(row.get("cost_dept_code")),
        clean(row.get("desc1")),
        clean(row.get("desc2")),
        clean(row.get("vendor_code")),
        clean(row.get("vendor_name")),
        clean(row.get("tax_code")),
        clean(row.get("proof_date")),
        clean(row.get("voucher_type")),
        clean(row.get("management_item")),
        clean(row.get("fiscal_year")),
    ]
    return hashlib.sha1("|".join(parts).encode("utf-8", "ignore")).hexdigest()


def import_voucher_rows(conn: Any, source_id: int, sheet_name: str, rows: Iterable[tuple[Any, ...]], year_hint: int | None) -> int:
    inserted = 0
    insert_sql = text(
        """
        INSERT INTO wehago_voucher_rows (
            source_file_id, sheet_name, row_number, approval_status, draft_no, confirmed_no, account_code, account_name,
            debit_supply, debit_tax, credit_supply, credit_tax, issue_dept_code, issue_dept_name, support_dept_code, support_dept_name,
            cost_dept_code, cost_dept_name, desc1, desc2, vendor_code, vendor_name, tax_code, proof_date, voucher_type, management_item,
            compare_voucher_no, compare_amount, compare_side, compare_vendor, compare_desc, fiscal_year
        ) VALUES (
            :source_file_id, :sheet_name, :row_number, :approval_status, :draft_no, :confirmed_no, :account_code, :account_name,
            :debit_supply, :debit_tax, :credit_supply, :credit_tax, :issue_dept_code, :issue_dept_name, :support_dept_code, :support_dept_name,
            :cost_dept_code, :cost_dept_name, :desc1, :desc2, :vendor_code, :vendor_name, :tax_code, :proof_date, :voucher_type, :management_item,
            :compare_voucher_no, :compare_amount, :compare_side, :compare_vendor, :compare_desc, :fiscal_year
        )
        """
    )
    for row_number, row in enumerate(rows, start=2):
        values = list(row)
        if not any(item is not None and clean(item) for item in values):
            continue
        proof_date = parse_excel_date(values[20] if len(values) > 20 else None)
        fiscal_year = year_from_date_text(proof_date) or year_hint
        draft_no = clean(values[1] if len(values) > 1 else "")
        confirmed_no = clean(values[10] if len(values) > 10 else "")
        debit_supply = parse_amount(values[4] if len(values) > 4 else 0)
        credit_supply = parse_amount(values[6] if len(values) > 6 else 0)
        conn.execute(
            insert_sql,
            {
                "source_file_id": source_id,
                "sheet_name": sheet_name,
                "row_number": row_number,
                "approval_status": clean(values[0] if len(values) > 0 else ""),
                "draft_no": draft_no,
                "confirmed_no": confirmed_no,
                "account_code": clean(values[2] if len(values) > 2 else ""),
                "account_name": clean(values[3] if len(values) > 3 else ""),
                "debit_supply": debit_supply,
                "debit_tax": parse_amount(values[5] if len(values) > 5 else 0),
                "credit_supply": credit_supply,
                "credit_tax": parse_amount(values[7] if len(values) > 7 else 0),
                "issue_dept_code": clean(values[8] if len(values) > 8 else ""),
                "issue_dept_name": clean(values[9] if len(values) > 9 else ""),
                "support_dept_code": clean(values[11] if len(values) > 11 else ""),
                "support_dept_name": clean(values[12] if len(values) > 12 else ""),
                "cost_dept_code": clean(values[13] if len(values) > 13 else ""),
                "cost_dept_name": clean(values[14] if len(values) > 14 else ""),
                "desc1": clean(values[15] if len(values) > 15 else ""),
                "desc2": clean(values[16] if len(values) > 16 else ""),
                "vendor_code": clean(values[17] if len(values) > 17 else ""),
                "vendor_name": clean(values[18] if len(values) > 18 else ""),
                "tax_code": clean(values[19] if len(values) > 19 else ""),
                "proof_date": proof_date,
                "voucher_type": clean(values[21] if len(values) > 21 else ""),
                "management_item": clean(values[22] if len(values) > 22 else ""),
                "compare_voucher_no": normalize_voucher_no(confirmed_no or draft_no),
                "compare_amount": debit_supply if debit_supply else credit_supply,
                "compare_side": "debit" if debit_supply else ("credit" if credit_supply else ""),
                "compare_vendor": normalize_text(values[18] if len(values) > 18 else ""),
                "compare_desc": normalize_text(
                    " ".join(clean(values[index]) for index in (15, 16, 22) if index < len(values) and clean(values[index]))
                ),
                "fiscal_year": fiscal_year,
            },
        )
        inserted += 1
    return inserted


def import_ledger_rows(conn: Any, source_id: int, sheet_name: str, rows: Iterable[tuple[Any, ...]], year_hint: int | None) -> int:
    inserted = 0
    insert_sql = text(
        """
        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, :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
        )
        """
    )
    for row_number, row in enumerate(rows, start=2):
        values = list(row)
        if not any(item is not None and clean(item) for item in values):
            continue
        ledger_date = parse_excel_date(values[0] if len(values) > 0 else None, default_year=year_hint)
        fiscal_year = year_from_date_text(ledger_date) or year_hint
        debit = parse_amount(values[3] if len(values) > 3 else 0)
        credit = parse_amount(values[4] if len(values) > 4 else 0)
        conn.execute(
            insert_sql,
            {
                "source_file_id": source_id,
                "sheet_name": sheet_name,
                "row_number": row_number,
                "ledger_date": ledger_date,
                "description": clean(values[1] if len(values) > 1 else ""),
                "vendor_name": clean(values[2] if len(values) > 2 else ""),
                "debit": debit,
                "credit": credit,
                "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_name": clean(values[8] if len(values) > 8 else ""),
                "compare_voucher_no": normalize_voucher_no(values[6] if len(values) > 6 else ""),
                "compare_amount": debit if debit else credit,
                "compare_side": "debit" if debit else ("credit" if credit else ""),
                "compare_vendor": normalize_text(values[2] if len(values) > 2 else ""),
                "compare_desc": normalize_text(values[1] if len(values) > 1 else ""),
                "fiscal_year": fiscal_year,
            },
        )
        inserted += 1
    return inserted


def rebuild_comparison_results(conn: Any) -> None:
    conn.execute(text("DELETE FROM wehago_comparison_results"))
    conn.execute(
        text(
            """
            WITH voucher_groups AS (
                SELECT
                    COALESCE(fiscal_year, 0) AS fiscal_year,
                    compare_voucher_no AS voucher_no,
                    COUNT(*) AS voucher_row_count,
                    SUM(debit_supply) AS voucher_debit,
                    SUM(credit_supply) AS voucher_credit,
                    GROUP_CONCAT(DISTINCT account_code) AS voucher_accounts,
                    GROUP_CONCAT(DISTINCT vendor_name) AS voucher_vendors
                FROM wehago_voucher_rows
                WHERE COALESCE(compare_voucher_no, '') <> ''
                GROUP BY COALESCE(fiscal_year, 0), compare_voucher_no
            ),
            ledger_groups AS (
                SELECT
                    COALESCE(fiscal_year, 0) AS fiscal_year,
                    compare_voucher_no AS voucher_no,
                    COUNT(*) AS ledger_row_count,
                    SUM(debit) AS ledger_debit,
                    SUM(credit) AS ledger_credit,
                    GROUP_CONCAT(DISTINCT account_code) AS ledger_accounts,
                    GROUP_CONCAT(DISTINCT vendor_name) AS ledger_vendors
                FROM wehago_ledger_rows
                WHERE COALESCE(compare_voucher_no, '') <> ''
                GROUP BY COALESCE(fiscal_year, 0), compare_voucher_no
            ),
            joined AS (
                SELECT
                    COALESCE(v.fiscal_year, l.fiscal_year) AS fiscal_year,
                    COALESCE(v.voucher_no, l.voucher_no) AS voucher_no,
                    COALESCE(v.voucher_row_count, 0) AS voucher_row_count,
                    COALESCE(l.ledger_row_count, 0) AS ledger_row_count,
                    COALESCE(v.voucher_debit, 0) AS voucher_debit,
                    COALESCE(v.voucher_credit, 0) AS voucher_credit,
                    COALESCE(l.ledger_debit, 0) AS ledger_debit,
                    COALESCE(l.ledger_credit, 0) AS ledger_credit,
                    COALESCE(v.voucher_accounts, '') AS voucher_accounts,
                    COALESCE(l.ledger_accounts, '') AS ledger_accounts,
                    COALESCE(v.voucher_vendors, '') AS voucher_vendors,
                    COALESCE(l.ledger_vendors, '') AS ledger_vendors
                FROM voucher_groups v
                LEFT JOIN ledger_groups l
                  ON l.fiscal_year = v.fiscal_year AND l.voucher_no = v.voucher_no
                UNION ALL
                SELECT
                    l.fiscal_year,
                    l.voucher_no,
                    0,
                    l.ledger_row_count,
                    0,
                    0,
                    l.ledger_debit,
                    l.ledger_credit,
                    '',
                    l.ledger_accounts,
                    '',
                    l.ledger_vendors
                FROM ledger_groups l
                LEFT JOIN voucher_groups v
                  ON v.fiscal_year = l.fiscal_year AND v.voucher_no = l.voucher_no
                WHERE v.voucher_no IS NULL
            )
            INSERT INTO wehago_comparison_results (
                fiscal_year, voucher_no, voucher_row_count, ledger_row_count,
                voucher_debit, voucher_credit, ledger_debit, ledger_credit,
                voucher_accounts, ledger_accounts, voucher_vendors, ledger_vendors,
                status, notes
            )
            SELECT
                fiscal_year,
                voucher_no,
                voucher_row_count,
                ledger_row_count,
                voucher_debit,
                voucher_credit,
                ledger_debit,
                ledger_credit,
                voucher_accounts,
                ledger_accounts,
                voucher_vendors,
                ledger_vendors,
                CASE
                    WHEN voucher_row_count > 0 AND ledger_row_count > 0
                         AND ABS(voucher_debit - ledger_debit) < 0.5
                         AND ABS(voucher_credit - ledger_credit) < 0.5
                        THEN 'matched'
                    WHEN voucher_row_count > 0 AND ledger_row_count = 0
                        THEN 'voucher_only'
                    WHEN voucher_row_count = 0 AND ledger_row_count > 0
                        THEN 'ledger_only'
                    ELSE 'amount_mismatch'
                END AS status,
                CASE
                    WHEN voucher_row_count > 0 AND ledger_row_count > 0
                         AND (ABS(voucher_debit - ledger_debit) >= 0.5 OR ABS(voucher_credit - ledger_credit) >= 0.5)
                        THEN '전표번호는 같지만 차/대변 합계가 다릅니다.'
                    WHEN voucher_row_count > 0 AND ledger_row_count = 0
                        THEN 'ERP에는 있으나 WEHAGO에서 찾지 못했습니다.'
                    WHEN voucher_row_count = 0 AND ledger_row_count > 0
                        THEN 'WEHAGO에는 있으나 ERP에서 찾지 못했습니다.'
                    ELSE ''
                END AS notes
            FROM joined
            """
        )
    )


def refresh_wehago_compare_data(engine: Any, source_root: Path | None = None) -> dict[str, int]:
    source_root = source_root or WEHAGO_SOURCE_ROOT
    summary = WehagoImportSummary()
    with engine.begin() as conn:
        init_wehago_compare_db(engine)
        for path in iter_source_files(source_root):
            summary.scanned_files += 1
            file_kind, header, sheet_name = detect_file_kind(path)
            if not file_kind:
                summary.skipped_files += 1
                continue
            workbook = load_workbook(path, read_only=True, data_only=True)
            sheet = workbook.worksheets[0]
            sample_rows = list(sheet.iter_rows(min_row=2, max_row=51, values_only=True))
            year_hint = infer_year_hint(path, file_kind, sample_rows)
            source_id, changed = upsert_source_file(conn, path, file_kind, year_hint, sheet_name, header)
            if not changed:
                summary.skipped_files += 1
                continue
            table_name = "wehago_voucher_rows" if file_kind == "voucher" else "wehago_ledger_rows"
            conn.execute(text(f"DELETE FROM {table_name} WHERE source_file_id = :source_id"), {"source_id": source_id})
            rows = sheet.iter_rows(min_row=2, values_only=True)
            if file_kind == "voucher":
                inserted = import_voucher_rows(conn, source_id, sheet_name, rows, year_hint)
                summary.voucher_rows += inserted
            else:
                inserted = import_ledger_rows(conn, source_id, sheet_name, rows, year_hint)
                summary.ledger_rows += inserted
            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},
            )
            summary.imported_files += 1
        rebuild_comparison_results(conn)
        summary.comparison_rows = int(conn.execute(text("SELECT COUNT(*) FROM wehago_comparison_results")).scalar_one())
    _DASHBOARD_CACHE.clear()
    _SUGGEST_CACHE.clear()
    _STATUS_ROWS_CACHE.clear()
    return {
        "scanned_files": summary.scanned_files,
        "imported_files": summary.imported_files,
        "skipped_files": summary.skipped_files,
        "voucher_rows": summary.voucher_rows,
        "ledger_rows": summary.ledger_rows,
        "comparison_rows": summary.comparison_rows,
    }


def fetch_existing_voucher_signatures(conn: Any) -> set[str]:
    rows = conn.execute(
        text(
            """
            SELECT
                approval_status, draft_no, confirmed_no, account_code, account_name,
                debit_supply, debit_tax, credit_supply, credit_tax,
                issue_dept_code, support_dept_code, cost_dept_code,
                desc1, desc2, vendor_code, vendor_name, tax_code,
                proof_date, voucher_type, management_item, fiscal_year
            FROM wehago_voucher_rows
            """
        )
    ).mappings()
    return {build_voucher_signature_from_mapping(dict(row)) for row in rows}


def count_nonempty_rows(sheet: Any) -> int:
    count = 0
    for row in sheet.iter_rows(min_row=2, values_only=True):
        values = list(row)
        if any(item is not None and clean(item) for item in values):
            count += 1
    return count


def import_uploaded_erp_voucher_file(engine: Any, upload_path: Path, original_filename: str) -> dict[str, Any]:
    init_wehago_compare_db(engine)
    file_kind, header, sheet_name = detect_file_kind(upload_path)
    if file_kind != "voucher":
        raise ValueError("선택한 파일이 ERP 전표 형식이 아닙니다.")

    workbook = load_workbook(upload_path, read_only=True, data_only=True)
    sheet = workbook.worksheets[0]
    sample_rows = list(sheet.iter_rows(min_row=2, max_row=51, values_only=True))
    year_hint = infer_year_hint(upload_path, "voucher", sample_rows)
    file_hash = compute_file_hash(upload_path)
    pseudo_path = f"upload://voucher/{file_hash}"
    file_size = upload_path.stat().st_size
    modified_ts = upload_path.stat().st_mtime

    with engine.begin() as conn:
        existing_source = conn.execute(
            text(
                """
                SELECT id, row_count
                FROM wehago_source_files
                WHERE file_path = :file_path
                """
            ),
            {"file_path": pseudo_path},
        ).mappings().first()
        if existing_source is not None:
            return {
                "source_id": int(existing_source["id"]),
                "inserted_rows": 0,
                "duplicate_rows": count_nonempty_rows(sheet),
                "file_name": original_filename,
                "comparison_rows": int(conn.execute(text("SELECT COUNT(*) FROM wehago_comparison_results")).scalar_one()),
            }

        existing_signatures = fetch_existing_voucher_signatures(conn)
        seen_signatures: set[str] = set()
        insert_sql = text(
            """
            INSERT INTO wehago_voucher_rows (
                source_file_id, sheet_name, row_number, approval_status, draft_no, confirmed_no, account_code, account_name,
                debit_supply, debit_tax, credit_supply, credit_tax, issue_dept_code, issue_dept_name, support_dept_code, support_dept_name,
                cost_dept_code, cost_dept_name, desc1, desc2, vendor_code, vendor_name, tax_code, proof_date, voucher_type, management_item,
                compare_voucher_no, compare_amount, compare_side, compare_vendor, compare_desc, fiscal_year
            ) VALUES (
                :source_file_id, :sheet_name, :row_number, :approval_status, :draft_no, :confirmed_no, :account_code, :account_name,
                :debit_supply, :debit_tax, :credit_supply, :credit_tax, :issue_dept_code, :issue_dept_name, :support_dept_code, :support_dept_name,
                :cost_dept_code, :cost_dept_name, :desc1, :desc2, :vendor_code, :vendor_name, :tax_code, :proof_date, :voucher_type, :management_item,
                :compare_voucher_no, :compare_amount, :compare_side, :compare_vendor, :compare_desc, :fiscal_year
            )
            """
        )

        source_result = conn.execute(
            text(
                """
                INSERT INTO wehago_source_files (
                    file_kind, file_path, file_name, relative_path, year_hint, sheet_name,
                    file_size, modified_ts, file_hash, header_json, row_count, source_origin
                ) VALUES (
                    'voucher', :file_path, :file_name, :relative_path, :year_hint, :sheet_name,
                    :file_size, :modified_ts, :file_hash, :header_json, 0, 'upload'
                )
                """
            ),
            {
                "file_path": pseudo_path,
                "file_name": original_filename,
                "relative_path": f"uploaded/{original_filename}",
                "year_hint": year_hint,
                "sheet_name": sheet_name,
                "file_size": file_size,
                "modified_ts": modified_ts,
                "file_hash": file_hash,
                "header_json": str(header),
            },
        )
        source_id = int(source_result.lastrowid)
        inserted = 0
        duplicate_rows = 0

        for row_number, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
            values = list(row)
            if not any(item is not None and clean(item) for item in values):
                continue
            signature = build_voucher_signature(values, year_hint)
            if signature in existing_signatures or signature in seen_signatures:
                duplicate_rows += 1
                continue
            seen_signatures.add(signature)
            proof_date = parse_excel_date(values[20] if len(values) > 20 else None)
            fiscal_year = year_from_date_text(proof_date) or year_hint
            draft_no = clean(values[1] if len(values) > 1 else "")
            confirmed_no = clean(values[10] if len(values) > 10 else "")
            debit_supply = parse_amount(values[4] if len(values) > 4 else 0)
            credit_supply = parse_amount(values[6] if len(values) > 6 else 0)
            conn.execute(
                insert_sql,
                {
                    "source_file_id": source_id,
                    "sheet_name": sheet_name,
                    "row_number": row_number,
                    "approval_status": clean(values[0] if len(values) > 0 else ""),
                    "draft_no": draft_no,
                    "confirmed_no": confirmed_no,
                    "account_code": clean(values[2] if len(values) > 2 else ""),
                    "account_name": clean(values[3] if len(values) > 3 else ""),
                    "debit_supply": debit_supply,
                    "debit_tax": parse_amount(values[5] if len(values) > 5 else 0),
                    "credit_supply": credit_supply,
                    "credit_tax": parse_amount(values[7] if len(values) > 7 else 0),
                    "issue_dept_code": clean(values[8] if len(values) > 8 else ""),
                    "issue_dept_name": clean(values[9] if len(values) > 9 else ""),
                    "support_dept_code": clean(values[11] if len(values) > 11 else ""),
                    "support_dept_name": clean(values[12] if len(values) > 12 else ""),
                    "cost_dept_code": clean(values[13] if len(values) > 13 else ""),
                    "cost_dept_name": clean(values[14] if len(values) > 14 else ""),
                    "desc1": clean(values[15] if len(values) > 15 else ""),
                    "desc2": clean(values[16] if len(values) > 16 else ""),
                    "vendor_code": clean(values[17] if len(values) > 17 else ""),
                    "vendor_name": clean(values[18] if len(values) > 18 else ""),
                    "tax_code": clean(values[19] if len(values) > 19 else ""),
                    "proof_date": proof_date,
                    "voucher_type": clean(values[21] if len(values) > 21 else ""),
                    "management_item": clean(values[22] if len(values) > 22 else ""),
                    "compare_voucher_no": normalize_voucher_no(confirmed_no or draft_no),
                    "compare_amount": debit_supply if debit_supply else credit_supply,
                    "compare_side": "debit" if debit_supply else ("credit" if credit_supply else ""),
                    "compare_vendor": normalize_text(values[18] if len(values) > 18 else ""),
                    "compare_desc": normalize_text(
                        " ".join(clean(values[index]) for index in (15, 16, 22) if index < len(values) and clean(values[index]))
                    ),
                    "fiscal_year": fiscal_year,
                },
            )
            inserted += 1

        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},
        )
        rebuild_comparison_results(conn)
        comparison_rows = int(conn.execute(text("SELECT COUNT(*) FROM wehago_comparison_results")).scalar_one())

    return {
        "source_id": source_id,
        "inserted_rows": inserted,
        "duplicate_rows": duplicate_rows,
        "file_name": original_filename,
        "comparison_rows": comparison_rows,
    }


def build_year_filter_sql(column_name: str = "fiscal_year") -> str:
    return (
        f"(:start_year IS NULL OR {column_name} >= :start_year) "
        f"AND (:end_year IS NULL OR {column_name} <= :end_year)"
    )


def fetch_metric_sections(conn: Any, start_year: int | None, end_year: int | None) -> list[dict[str, Any]]:
    sections: list[dict[str, Any]] = []
    for status_key, label, description in STATUS_META:
        count = int(
            conn.execute(
                text(
                    """
                    SELECT COUNT(*)
                    FROM wehago_comparison_results
                    WHERE status = :status
                      AND """
                    + build_year_filter_sql()
                    + """
                    """
                ),
                {"status": status_key, "start_year": start_year, "end_year": end_year},
            ).scalar_one()
        )
        sections.append(
            {
                "key": status_key,
                "label": label,
                "description": description,
                "count": count,
                "columns": DETAIL_COLUMN_MAP[status_key],
                "rows": [],
            }
        )
    return sections


def fetch_wehago_rows(conn: Any, start_year: int | None, end_year: int | None) -> list[dict[str, Any]]:
    return [
        dict(row._mapping)
        for row in conn.execute(
            text(
                """
                SELECT
                    ledger_date, voucher_no, account_code, account_name,
                    vendor_name, description, debit, credit
                FROM wehago_ledger_rows
                WHERE """
                + build_year_filter_sql()
                + """
                ORDER BY COALESCE(ledger_date, '') DESC, voucher_no DESC, row_number DESC
                LIMIT 300
                """
            ),
            {"start_year": start_year, "end_year": end_year},
        ).fetchall()
    ]


def fetch_erp_rows(conn: Any, start_year: int | None, end_year: int | None) -> list[dict[str, Any]]:
    return [
        dict(row._mapping)
        for row in conn.execute(
            text(
                """
                SELECT
                    proof_date, confirmed_no, draft_no, account_code, account_name,
                    vendor_name, desc1, desc2, debit_supply, credit_supply
                FROM wehago_voucher_rows
                WHERE """
                + build_year_filter_sql()
                + """
                ORDER BY COALESCE(proof_date, '') DESC, confirmed_no DESC, draft_no DESC, row_number DESC
                LIMIT 300
                """
            ),
            {"start_year": start_year, "end_year": end_year},
        ).fetchall()
    ]


def fetch_latest_upload_meta(conn: Any) -> dict[str, Any] | None:
    row = conn.execute(
        text(
            """
            SELECT file_name, imported_at, row_count
            FROM wehago_source_files
            WHERE file_kind = 'voucher' AND source_origin = 'upload'
            ORDER BY imported_at DESC, id DESC
            LIMIT 1
            """
        )
    ).mappings().first()
    return dict(row) if row else None


def count_rows(conn: Any, table_name: str, start_year: int | None, end_year: int | None) -> int:
    return int(
        conn.execute(
            text(
                f"""
                SELECT COUNT(*)
                FROM {table_name}
                WHERE {build_year_filter_sql()}
                """
            ),
            {"start_year": start_year, "end_year": end_year},
        ).scalar_one()
    )


def discover_available_years(source_root: Path | None = None) -> list[int]:
    source_root = source_root or WEHAGO_SOURCE_ROOT
    data_download_dir = source_root / "data_download"
    if not data_download_dir.exists():
        return []
    years = sorted(
        [
            int(path.name)
            for path in data_download_dir.iterdir()
            if path.is_dir() and path.name.isdigit() and len(path.name) == 4
        ],
        reverse=True,
    )
    return years


def get_default_year(available_years: list[int]) -> int | None:
    return available_years[0] if available_years else None


def discover_compare_result_bundle(year: int, source_root: Path | None = None) -> dict[str, Path] | None:
    source_root = source_root or WEHAGO_SOURCE_ROOT
    data_download_dir = source_root / "data_download"
    if not data_download_dir.exists():
        return None

    candidates: list[tuple[float, dict[str, Path]]] = []
    for folder in data_download_dir.iterdir():
        if not folder.is_dir() or (folder.name.isdigit() and len(folder.name) == 4):
            continue
        ledger_raw = next(folder.glob(f"ledger_{year}_*.xlsx"), None)
        voucher_raw = next(folder.glob(f"voucher_{year}_*.xlsx"), None)
        ledger_result = folder / "ledger_with_matched_voucher_detected_context60_v10_20260422.xlsx"
        voucher_result = folder / "voucher_with_matched_ledger_detected_context60_v10_20260422.xlsx"
        if ledger_raw and voucher_raw and ledger_result.exists() and voucher_result.exists():
            candidates.append(
                (
                    max(ledger_result.stat().st_mtime, voucher_result.stat().st_mtime, folder.stat().st_mtime),
                    {
                        "ledger_result": ledger_result,
                        "voucher_result": voucher_result,
                    },
                )
            )
    if not candidates:
        return None
    candidates.sort(key=lambda item: item[0], reverse=True)
    return candidates[0][1]


def _normalize_flag(value: Any) -> bool:
    text_value = clean(value).strip().lower()
    return text_value not in {"", "0", "false", "n", "none"}


def _append_limited(bucket: list[dict[str, Any]], item: dict[str, Any], limit: int = 200) -> None:
    if len(bucket) < limit:
        bucket.append(item)


def build_review_key(row: dict[str, Any]) -> str:
    parts = [
        clean(row.get("fiscal_year")),
        clean(row.get("voucher_no")),
        clean(row.get("draft_no")),
        clean(row.get("voucher_account_code")),
        clean(row.get("voucher_account_name")),
        clean(row.get("voucher_vendor")),
        clean(row.get("voucher_desc")),
        clean(row.get("review_reason")),
        clean(row.get("review_memo")),
        f"{parse_amount(row.get('voucher_debit')):.2f}",
        f"{parse_amount(row.get('voucher_credit')):.2f}",
    ]
    return hashlib.sha1("|".join(parts).encode("utf-8", "ignore")).hexdigest()


def build_match_identity_key(row: dict[str, Any]) -> str:
    parts = [
        clean(row.get("fiscal_year")),
        clean(row.get("draft_no")),
        clean(row.get("voucher_account_code")),
        clean(row.get("voucher_account_name")),
        clean(row.get("voucher_vendor")),
        clean(row.get("voucher_desc")),
        f"{parse_amount(row.get('voucher_debit')):.2f}",
        f"{parse_amount(row.get('voucher_credit')):.2f}",
    ]
    return hashlib.sha1("|".join(parts).encode("utf-8", "ignore")).hexdigest()


def build_ledger_row_key(row: dict[str, Any]) -> str:
    parts = [
        clean(row.get("fiscal_year")),
        clean(row.get("voucher_no")),
        clean(row.get("ledger_date")),
        clean(row.get("ledger_account_code")),
        clean(row.get("ledger_account_name")),
        clean(row.get("ledger_vendor")),
        clean(row.get("ledger_desc")),
        f"{parse_amount(row.get('ledger_debit')):.2f}",
        f"{parse_amount(row.get('ledger_credit')):.2f}",
    ]
    return hashlib.sha1("|".join(parts).encode("utf-8", "ignore")).hexdigest()


def build_voucher_row_key(row: dict[str, Any]) -> str:
    parts = [
        clean(row.get("fiscal_year")),
        clean(row.get("voucher_no")),
        clean(row.get("draft_no")),
        clean(row.get("proof_date")),
        clean(row.get("voucher_account_code")),
        clean(row.get("voucher_account_name")),
        clean(row.get("voucher_vendor")),
        clean(row.get("voucher_desc")),
        f"{parse_amount(row.get('voucher_debit')):.2f}",
        f"{parse_amount(row.get('voucher_credit')):.2f}",
    ]
    return hashlib.sha1("|".join(parts).encode("utf-8", "ignore")).hexdigest()


def build_manual_pair_key(ledger_row_key: str, voucher_row_key: str) -> str:
    return hashlib.sha1(f"{clean(ledger_row_key)}|{clean(voucher_row_key)}".encode("utf-8", "ignore")).hexdigest()


@lru_cache(maxsize=16)
def parse_compare_result_bundle(
    ledger_path_str: str,
    ledger_mtime: float,
    voucher_path_str: str,
    voucher_mtime: float,
    year: int,
) -> dict[str, Any]:
    del ledger_mtime, voucher_mtime
    ledger_path = Path(ledger_path_str)
    voucher_path = Path(voucher_path_str)
    metric_map = {
        "matched": {
            "count": 0,
            "columns": [
                ("fiscal_year", "연도"),
                ("ledger_date", "일자"),
                ("voucher_no", "전표번호"),
                ("draft_no", "가전표번호"),
                ("ledger_account_name", "계정"),
                ("voucher_account_name", "ERP 계정"),
                ("ledger_vendor", "거래처"),
                ("voucher_vendor", "ERP 거래처"),
                ("ledger_debit", "차변"),
                ("ledger_credit", "대변"),
                ("voucher_debit", "ERP 차변"),
                ("voucher_credit", "ERP 대변"),
                ("ledger_desc", "WEHAGO 적요"),
                ("voucher_desc", "ERP 적요"),
            ],
            "rows": [],
        },
        "ledger_only": {
            "count": 0,
            "columns": [
                ("fiscal_year", "연도"),
                ("voucher_no", "전표번호"),
                ("ledger_date", "일자"),
                ("ledger_account_name", "WEHAGO 계정"),
                ("ledger_vendor", "WEHAGO 거래처"),
                ("ledger_debit", "차변"),
                ("ledger_credit", "대변"),
                ("ledger_desc", "적요"),
            ],
            "rows": [],
        },
        "amount_mismatch": {
            "count": 0,
            "columns": [
                ("fiscal_year", "연도"),
                ("ledger_date", "일자"),
                ("voucher_no", "전표번호"),
                ("draft_no", "가전표번호"),
                ("ledger_account_name", "계정"),
                ("voucher_account_name", "ERP 계정"),
                ("ledger_vendor", "거래처"),
                ("voucher_vendor", "ERP 거래처"),
                ("ledger_debit", "차변"),
                ("ledger_credit", "대변"),
                ("voucher_debit", "ERP 차변"),
                ("voucher_credit", "ERP 대변"),
                ("ledger_desc", "WEHAGO 적요"),
                ("voucher_desc", "ERP 적요"),
            ],
            "rows": [],
        },
        "voucher_only": {
            "count": 0,
            "columns": [
                ("fiscal_year", "연도"),
                ("voucher_no", "전표번호"),
                ("proof_date", "증빙일자"),
                ("voucher_account_name", "ERP 계정"),
                ("voucher_vendor", "ERP 거래처"),
                ("voucher_debit", "차변"),
                ("voucher_credit", "대변"),
                ("voucher_desc", "적요"),
            ],
            "rows": [],
        },
    }

    ledger_wb = load_workbook(ledger_path, read_only=True, data_only=True)
    ledger_ws = ledger_wb.worksheets[0]
    ledger_header = [clean(v) for v in next(ledger_ws.iter_rows(min_row=1, max_row=1, values_only=True))]
    ledger_idx = {name: idx for idx, name in enumerate(ledger_header)}
    ledger_rows: list[tuple[Any, ...]] = list(ledger_ws.iter_rows(min_row=2, values_only=True))
    ledger_by_matched_voucher_no: dict[str, dict[str, Any]] = {}
    for row in ledger_rows:
        matched_voucher_no = clean(row[ledger_idx["matched_확정전표번호"]]) if "matched_확정전표번호" in ledger_idx else ""
        if not matched_voucher_no or matched_voucher_no in ledger_by_matched_voucher_no:
            continue
        ledger_by_matched_voucher_no[matched_voucher_no] = {
            "voucher_no": clean(row[ledger_idx["전표번호"]]) if "전표번호" in ledger_idx else "",
            "ledger_date": clean(row[ledger_idx["일자"]]) if "일자" in ledger_idx else "",
            "ledger_account_code": clean(row[ledger_idx["계정코드"]]) if "계정코드" in ledger_idx else "",
            "ledger_account_name": clean(row[ledger_idx["계정명"]]) if "계정명" in ledger_idx else "",
            "ledger_vendor": clean(row[ledger_idx["거래처"]]) if "거래처" in ledger_idx else "",
            "ledger_desc": clean(row[ledger_idx["적요"]]) if "적요" in ledger_idx else "",
            "ledger_debit": parse_amount(row[ledger_idx["차변"]]) if "차변" in ledger_idx else 0,
            "ledger_credit": parse_amount(row[ledger_idx["대변"]]) if "대변" in ledger_idx else 0,
        }

    voucher_wb = load_workbook(voucher_path, read_only=True, data_only=True)
    voucher_ws = voucher_wb.worksheets[0]
    voucher_header = [clean(v) for v in next(voucher_ws.iter_rows(min_row=1, max_row=1, values_only=True))]
    voucher_idx = {name: idx for idx, name in enumerate(voucher_header)}
    voucher_by_confirmed_no: dict[str, dict[str, Any]] = {}
    for row in voucher_ws.iter_rows(min_row=2, values_only=True):
        matched_voucher_no = clean(row[voucher_idx["matched_전표번호"]]) if "matched_전표번호" in voucher_idx else ""
        review_reason = clean(row[voucher_idx["matched_검증근거"]]) if "matched_검증근거" in voucher_idx else ""
        review_flag = row[voucher_idx["review_flag"]] if "review_flag" in voucher_idx else ""
        confirmed_no = clean(row[voucher_idx["확정전표번호"]]) if "확정전표번호" in voucher_idx else ""
        desc1 = clean(row[voucher_idx["적요1"]]) if "적요1" in voucher_idx else ""
        matched_ledger = ledger_by_matched_voucher_no.get(confirmed_no, {})
        item = {
            "fiscal_year": year,
            "voucher_no": (
                matched_ledger.get("voucher_no", "")
                if matched_voucher_no
                else (confirmed_no or (clean(row[voucher_idx["가전표번호"]]) if "가전표번호" in voucher_idx else ""))
            ),
            "confirmed_no": confirmed_no,
            "proof_date": clean(row[voucher_idx["증빙일자"]]) if "증빙일자" in voucher_idx else "",
            "ledger_date": matched_ledger.get("ledger_date", ""),
            "draft_no": clean(row[voucher_idx["가전표번호"]]) if "가전표번호" in voucher_idx else "",
            "ledger_account_code": matched_ledger.get("ledger_account_code", ""),
            "ledger_account_name": matched_ledger.get("ledger_account_name", ""),
            "voucher_account_code": clean(row[voucher_idx["계정코드"]]) if "계정코드" in voucher_idx else "",
            "voucher_account_name": clean(row[voucher_idx["계정명칭"]]) if "계정명칭" in voucher_idx else "",
            "ledger_vendor": matched_ledger.get("ledger_vendor", ""),
            "voucher_vendor": clean(row[voucher_idx["거래처명칭"]]) if "거래처명칭" in voucher_idx else "",
            "ledger_desc": matched_ledger.get("ledger_desc", ""),
            "voucher_debit": parse_amount(row[voucher_idx["차변공급가"]]) if "차변공급가" in voucher_idx else 0,
            "voucher_credit": parse_amount(row[voucher_idx["대변공급가"]]) if "대변공급가" in voucher_idx else 0,
            "voucher_desc": desc1,
            "ledger_debit": matched_ledger.get("ledger_debit", 0),
            "ledger_credit": matched_ledger.get("ledger_credit", 0),
            "review_reason": review_reason,
            "review_memo": clean(row[voucher_idx["review_memo"]]) if "review_memo" in voucher_idx else "",
        }
        item["review_key"] = build_review_key(item)
        item["match_identity_key"] = build_match_identity_key(item)
        item["voucher_row_key"] = build_voucher_row_key(item)
        if confirmed_no and confirmed_no not in voucher_by_confirmed_no:
            voucher_by_confirmed_no[confirmed_no] = item
        if not matched_voucher_no:
            metric_map["voucher_only"]["count"] += 1
            metric_map["voucher_only"]["rows"].append(item)
        has_wehago_link = bool(
            item.get("voucher_no")
            or item.get("ledger_date")
            or item.get("ledger_account_name")
            or item.get("ledger_vendor")
        )
        if has_wehago_link and (_normalize_flag(review_flag) or ("REVIEW" in review_reason)):
            metric_map["amount_mismatch"]["count"] += 1
            metric_map["amount_mismatch"]["rows"].append(item)

    for row in ledger_rows:
        matched_voucher_no = clean(row[ledger_idx["matched_확정전표번호"]]) if "matched_확정전표번호" in ledger_idx else ""
        matched_voucher = voucher_by_confirmed_no.get(matched_voucher_no, {})
        item = {
            "fiscal_year": year,
            "voucher_no": clean(row[ledger_idx["전표번호"]]) if "전표번호" in ledger_idx else "",
            "ledger_date": clean(row[ledger_idx["일자"]]) if "일자" in ledger_idx else "",
            "draft_no": matched_voucher.get("draft_no", ""),
            "ledger_account_code": clean(row[ledger_idx["계정코드"]]) if "계정코드" in ledger_idx else "",
            "ledger_account_name": clean(row[ledger_idx["계정명"]]) if "계정명" in ledger_idx else "",
            "voucher_account_code": matched_voucher.get("voucher_account_code")
            or (clean(row[ledger_idx["matched_계정코드"]]) if "matched_계정코드" in ledger_idx else ""),
            "voucher_account_name": matched_voucher.get("voucher_account_name")
            or (clean(row[ledger_idx["matched_계정명칭"]]) if "matched_계정명칭" in ledger_idx else ""),
            "ledger_vendor": clean(row[ledger_idx["거래처"]]) if "거래처" in ledger_idx else "",
            "voucher_vendor": matched_voucher.get("voucher_vendor", ""),
            "voucher_debit": matched_voucher.get("voucher_debit", 0),
            "voucher_credit": matched_voucher.get("voucher_credit", 0),
            "ledger_desc": clean(row[ledger_idx["적요"]]) if "적요" in ledger_idx else "",
            "voucher_desc": matched_voucher.get("voucher_desc", ""),
            "ledger_debit": parse_amount(row[ledger_idx["차변"]]) if "차변" in ledger_idx else 0,
            "ledger_credit": parse_amount(row[ledger_idx["대변"]]) if "대변" in ledger_idx else 0,
        }
        item["review_key"] = build_review_key(item)
        item["match_identity_key"] = build_match_identity_key(item)
        item["ledger_row_key"] = build_ledger_row_key(item)
        if matched_voucher_no:
            metric_map["matched"]["count"] += 1
            metric_map["matched"]["rows"].append(item)
        else:
            metric_map["ledger_only"]["count"] += 1
            metric_map["ledger_only"]["rows"].append(item)

    return metric_map


def build_metric_sections_from_results(start_year: int | None, end_year: int | None) -> list[dict[str, Any]]:
    if start_year is None or end_year is None:
        return []
    aggregate = {
        "matched": {"count": 0, "columns": None, "rows": []},
        "ledger_only": {"count": 0, "columns": None, "rows": []},
        "amount_mismatch": {"count": 0, "columns": None, "rows": []},
        "voucher_only": {"count": 0, "columns": None, "rows": []},
    }
    for year in range(start_year, end_year + 1):
        bundle = discover_compare_result_bundle(year)
        if not bundle:
            continue
        parsed = parse_compare_result_bundle(
            str(bundle["ledger_result"]),
            bundle["ledger_result"].stat().st_mtime,
            str(bundle["voucher_result"]),
            bundle["voucher_result"].stat().st_mtime,
            year,
        )
        for key, section in aggregate.items():
            section["count"] += parsed[key]["count"]
            section["columns"] = parsed[key]["columns"]

    built_sections: list[dict[str, Any]] = []
    for status_key, label, description in STATUS_META:
        section = aggregate[status_key]
        built_sections.append(
            {
                "key": status_key,
                "label": label,
                "description": description,
                "count": section["count"],
                "columns": section["columns"] or DETAIL_COLUMN_MAP[status_key],
                "rows": [],
            }
        )
    return built_sections


def build_metric_sections_with_review_state(conn: Any, start_year: int | None, end_year: int | None) -> list[dict[str, Any]]:
    if start_year is None or end_year is None:
        return []
    aggregate = {
        "matched": {"count": 0, "columns": None, "rows": []},
        "ledger_only": {"count": 0, "columns": None, "rows": []},
        "amount_mismatch": {"count": 0, "columns": None, "rows": []},
        "voucher_only": {"count": 0, "columns": None, "rows": []},
    }
    for year in range(start_year, end_year + 1):
        bundle = discover_compare_result_bundle(year)
        if not bundle:
            continue
        parsed = parse_compare_result_bundle(
            str(bundle["ledger_result"]),
            bundle["ledger_result"].stat().st_mtime,
            str(bundle["voucher_result"]),
            bundle["voucher_result"].stat().st_mtime,
            year,
        )
        parsed = apply_saved_recheck_reviews(parsed, get_saved_recheck_review_keys(conn, year, year))
        parsed = apply_saved_manual_pair_matches(parsed, get_saved_manual_pair_matches(conn, year, year))
        for key, section in aggregate.items():
            section["count"] += parsed[key]["count"]
            section["columns"] = parsed[key]["columns"]

    built_sections: list[dict[str, Any]] = []
    for status_key, label, description in STATUS_META:
        section = aggregate[status_key]
        built_sections.append(
            {
                "key": status_key,
                "label": label,
                "description": description,
                "count": section["count"],
                "columns": section["columns"] or DETAIL_COLUMN_MAP[status_key],
                "rows": [],
            }
        )
    return built_sections


def build_account_options_from_results(start_year: int | None, end_year: int | None) -> dict[str, list[str]]:
    wehago_items: set[str] = set()
    erp_items: set[str] = set()
    if start_year is None or end_year is None:
        return {"wehago": [], "erp": []}
    for year in range(start_year, end_year + 1):
        bundle = discover_compare_result_bundle(year)
        if not bundle:
            continue
        parsed = parse_compare_result_bundle(
            str(bundle["ledger_result"]),
            bundle["ledger_result"].stat().st_mtime,
            str(bundle["voucher_result"]),
            bundle["voucher_result"].stat().st_mtime,
            year,
        )
        for section in parsed.values():
            for row in section["rows"]:
                wehago_code = clean(row.get("ledger_account_code"))
                wehago_name = clean(row.get("ledger_account_name"))
                erp_code = clean(row.get("voucher_account_code"))
                erp_name = clean(row.get("voucher_account_name"))
                if wehago_code or wehago_name:
                    wehago_items.add(" ".join(part for part in (wehago_code, wehago_name) if part))
                if erp_code or erp_name:
                    erp_items.add(" ".join(part for part in (erp_code, erp_name) if part))
    return {"wehago": sorted(wehago_items), "erp": sorted(erp_items)}


def build_account_options_from_rows(rows_by_status: dict[str, list[dict[str, Any]]]) -> dict[str, list[str]]:
    wehago_items: set[str] = set()
    erp_items: set[str] = set()
    for rows in rows_by_status.values():
        for row in rows:
            wehago_code = clean(row.get("ledger_account_code"))
            wehago_name = clean(row.get("ledger_account_name"))
            erp_code = clean(row.get("voucher_account_code"))
            erp_name = clean(row.get("voucher_account_name"))
            if wehago_code or wehago_name:
                wehago_items.add(" ".join(part for part in (wehago_code, wehago_name) if part))
            if erp_code or erp_name:
                erp_items.add(" ".join(part for part in (erp_code, erp_name) if part))
    return {"wehago": sorted(wehago_items), "erp": sorted(erp_items)}


def build_account_options_from_db(conn: Any, start_year: int | None, end_year: int | None, limit: int = 3000) -> dict[str, list[str]]:
    wehago_rows = conn.execute(
        text(
            f"""
            SELECT DISTINCT COALESCE(account_code, '') AS account_code, COALESCE(account_name, '') AS account_name
            FROM wehago_ledger_rows
            WHERE {build_year_filter_sql()}
              AND (COALESCE(account_code, '') <> '' OR COALESCE(account_name, '') <> '')
            ORDER BY account_code, account_name
            LIMIT :limit
            """
        ),
        {"start_year": start_year, "end_year": end_year, "limit": limit},
    ).fetchall()
    erp_rows = conn.execute(
        text(
            f"""
            SELECT DISTINCT COALESCE(account_code, '') AS account_code, COALESCE(account_name, '') AS account_name
            FROM wehago_voucher_rows
            WHERE {build_year_filter_sql()}
              AND (COALESCE(account_code, '') <> '' OR COALESCE(account_name, '') <> '')
            ORDER BY account_code, account_name
            LIMIT :limit
            """
        ),
        {"start_year": start_year, "end_year": end_year, "limit": limit},
    ).fetchall()
    wehago = sorted({" ".join(part for part in (clean(row[0]), clean(row[1])) if part) for row in wehago_rows})
    erp = sorted({" ".join(part for part in (clean(row[0]), clean(row[1])) if part) for row in erp_rows})
    return {"wehago": wehago, "erp": erp}


def get_saved_recheck_review_keys(conn: Any, start_year: int | None, end_year: int | None) -> set[str]:
    rows = conn.execute(
        text(
            """
            SELECT review_key
            FROM wehago_recheck_reviews
            WHERE (:start_year IS NULL OR fiscal_year >= :start_year)
              AND (:end_year IS NULL OR fiscal_year <= :end_year)
            """
        ),
        {"start_year": start_year, "end_year": end_year},
    ).fetchall()
    return {str(row[0]) for row in rows}


def apply_saved_recheck_reviews(
    sections: dict[str, dict[str, Any]],
    reviewed_keys: set[str],
) -> dict[str, dict[str, Any]]:
    if not reviewed_keys:
        return sections
    adjusted: dict[str, dict[str, Any]] = {}
    for key, section in sections.items():
        copied = {
            "count": section["count"],
            "columns": section["columns"],
            "rows": list(section["rows"]),
        }
        adjusted[key] = copied

    original_recheck_rows = adjusted["amount_mismatch"]["rows"]
    remaining_recheck_rows = [row for row in original_recheck_rows if row.get("review_key") not in reviewed_keys]
    reviewed_rows = [row for row in original_recheck_rows if row.get("review_key") in reviewed_keys]
    adjusted["amount_mismatch"]["rows"] = remaining_recheck_rows
    adjusted["amount_mismatch"]["count"] = len(remaining_recheck_rows)

    matched_rows = adjusted["matched"]["rows"]
    matched_identity = {row.get("match_identity_key") for row in matched_rows}
    for row in reviewed_rows:
        identity = row.get("match_identity_key")
        if identity in matched_identity:
            continue
        matched_identity.add(identity)
        matched_rows.append(row)
    adjusted["matched"]["count"] = len(matched_rows)
    return adjusted


def get_saved_manual_pair_matches(conn: Any, start_year: int | None, end_year: int | None) -> list[dict[str, Any]]:
    rows = conn.execute(
        text(
            """
            SELECT
                pair_key, fiscal_year, ledger_row_key, voucher_row_key, pair_note
            FROM wehago_manual_pair_matches
            WHERE (:start_year IS NULL OR fiscal_year >= :start_year)
              AND (:end_year IS NULL OR fiscal_year <= :end_year)
            """
        ),
        {"start_year": start_year, "end_year": end_year},
    ).mappings()
    return [dict(row) for row in rows]


def _build_bundle_signature(start_year: int | None, end_year: int | None) -> str:
    if start_year is None or end_year is None:
        return ""
    parts: list[str] = []
    for year in range(start_year, end_year + 1):
        bundle = discover_compare_result_bundle(year)
        if not bundle:
            parts.append(f"{year}:none")
            continue
        ledger_mtime = f"{bundle['ledger_result'].stat().st_mtime:.3f}"
        voucher_mtime = f"{bundle['voucher_result'].stat().st_mtime:.3f}"
        parts.append(f"{year}:{ledger_mtime}:{voucher_mtime}")
    return "|".join(parts)


def _get_or_create_year_cached_sections(conn: Any, year: int) -> dict[str, dict[str, Any]] | None:
    bundle = discover_compare_result_bundle(year)
    if not bundle:
        return None
    ledger_path = str(bundle["ledger_result"])
    voucher_path = str(bundle["voucher_result"])
    ledger_mtime = float(bundle["ledger_result"].stat().st_mtime)
    voucher_mtime = float(bundle["voucher_result"].stat().st_mtime)

    cached = conn.execute(
        text(
            """
            SELECT payload_json
            FROM wehago_result_row_cache
            WHERE fiscal_year = :fiscal_year
              AND ledger_result_path = :ledger_result_path
              AND ledger_result_mtime = :ledger_result_mtime
              AND voucher_result_path = :voucher_result_path
              AND voucher_result_mtime = :voucher_result_mtime
            LIMIT 1
            """
        ),
        {
            "fiscal_year": year,
            "ledger_result_path": ledger_path,
            "ledger_result_mtime": ledger_mtime,
            "voucher_result_path": voucher_path,
            "voucher_result_mtime": voucher_mtime,
        },
    ).first()
    if cached and cached[0]:
        try:
            payload = json.loads(str(cached[0]))
            if isinstance(payload, dict):
                return payload
        except Exception:
            pass

    parsed = parse_compare_result_bundle(
        ledger_path,
        ledger_mtime,
        voucher_path,
        voucher_mtime,
        year,
    )
    conn.execute(
        text(
            """
            INSERT OR REPLACE INTO wehago_result_row_cache (
                fiscal_year,
                ledger_result_path,
                ledger_result_mtime,
                voucher_result_path,
                voucher_result_mtime,
                payload_json,
                created_at
            ) VALUES (
                :fiscal_year,
                :ledger_result_path,
                :ledger_result_mtime,
                :voucher_result_path,
                :voucher_result_mtime,
                :payload_json,
                CURRENT_TIMESTAMP
            )
            """
        ),
        {
            "fiscal_year": year,
            "ledger_result_path": ledger_path,
            "ledger_result_mtime": ledger_mtime,
            "voucher_result_path": voucher_path,
            "voucher_result_mtime": voucher_mtime,
            "payload_json": json.dumps(parsed, ensure_ascii=False),
        },
    )
    conn.execute(
        text(
            """
            DELETE FROM wehago_result_row_cache
            WHERE fiscal_year = :fiscal_year
              AND NOT (
                  ledger_result_path = :ledger_result_path
                  AND ledger_result_mtime = :ledger_result_mtime
                  AND voucher_result_path = :voucher_result_path
                  AND voucher_result_mtime = :voucher_result_mtime
              )
            """
        ),
        {
            "fiscal_year": year,
            "ledger_result_path": ledger_path,
            "ledger_result_mtime": ledger_mtime,
            "voucher_result_path": voucher_path,
            "voucher_result_mtime": voucher_mtime,
        },
    )
    return parsed


def _build_db_state_signature(conn: Any, start_year: int | None, end_year: int | None) -> str:
    recheck = conn.execute(
        text(
            """
            SELECT COALESCE(MAX(reviewed_at), ''), COUNT(*)
            FROM wehago_recheck_reviews
            WHERE (:start_year IS NULL OR fiscal_year >= :start_year)
              AND (:end_year IS NULL OR fiscal_year <= :end_year)
            """
        ),
        {"start_year": start_year, "end_year": end_year},
    ).first()
    pair = conn.execute(
        text(
            """
            SELECT COALESCE(MAX(created_at), ''), COUNT(*)
            FROM wehago_manual_pair_matches
            WHERE (:start_year IS NULL OR fiscal_year >= :start_year)
              AND (:end_year IS NULL OR fiscal_year <= :end_year)
            """
        ),
        {"start_year": start_year, "end_year": end_year},
    ).first()
    return f"r:{recheck[0]}:{recheck[1]}|p:{pair[0]}:{pair[1]}"


def _get_cached_status_rows_by_range(engine: Any, start_year: int | None, end_year: int | None) -> dict[str, list[dict[str, Any]]]:
    if start_year is None or end_year is None:
        return {"matched": [], "ledger_only": [], "amount_mismatch": [], "voucher_only": []}
    init_wehago_compare_db(engine)
    with engine.begin() as conn:
        cache_key = "|".join(
            [
                str(start_year),
                str(end_year),
                _build_bundle_signature(start_year, end_year),
                _build_db_state_signature(conn, start_year, end_year),
            ]
        )
        now = time.time()
        cached = _STATUS_ROWS_CACHE.get(cache_key)
        if cached and (now - float(cached.get("ts", 0))) <= _STATUS_ROWS_CACHE_TTL_SEC:
            return cached["rows_by_status"]

        reviewed_keys = get_saved_recheck_review_keys(conn, start_year, end_year)
        manual_pair_matches = get_saved_manual_pair_matches(conn, start_year, end_year)
        rows_by_status: dict[str, list[dict[str, Any]]] = {
            "matched": [],
            "ledger_only": [],
            "amount_mismatch": [],
            "voucher_only": [],
        }
        for year in range(start_year, end_year + 1):
            parsed = _get_or_create_year_cached_sections(conn, year)
            if not parsed:
                continue
            parsed = apply_saved_recheck_reviews(parsed, reviewed_keys)
            parsed = apply_saved_manual_pair_matches(parsed, manual_pair_matches)
            for status_key in rows_by_status:
                rows_by_status[status_key].extend(parsed[status_key]["rows"])
        _STATUS_ROWS_CACHE.clear()
        _STATUS_ROWS_CACHE[cache_key] = {"ts": now, "rows_by_status": rows_by_status}
        return rows_by_status


def _push_action_history(conn: Any, action_type: str, payload: dict[str, Any]) -> int:
    result = conn.execute(
        text(
            """
            INSERT INTO wehago_action_history (action_type, payload_json, created_at)
            VALUES (:action_type, :payload_json, CURRENT_TIMESTAMP)
            """
        ),
        {"action_type": action_type, "payload_json": json.dumps(payload, ensure_ascii=False)},
    )
    return int(result.lastrowid)


def apply_saved_manual_pair_matches(
    sections: dict[str, dict[str, Any]],
    pair_matches: list[dict[str, Any]],
) -> dict[str, dict[str, Any]]:
    if not pair_matches:
        return sections

    adjusted: dict[str, dict[str, Any]] = {}
    for key, section in sections.items():
        adjusted[key] = {
            "count": section["count"],
            "columns": section["columns"],
            "rows": list(section["rows"]),
        }

    ledger_rows = adjusted["ledger_only"]["rows"]
    voucher_rows = adjusted["voucher_only"]["rows"]
    matched_rows = adjusted["matched"]["rows"]

    ledger_by_key = {clean(row.get("ledger_row_key")): row for row in ledger_rows if clean(row.get("ledger_row_key"))}
    voucher_by_key = {clean(row.get("voucher_row_key")): row for row in voucher_rows if clean(row.get("voucher_row_key"))}

    ledger_remove: set[str] = set()
    voucher_remove: set[str] = set()
    matched_pair_keys = {clean(row.get("pair_match_key")) for row in matched_rows if clean(row.get("pair_match_key"))}

    for pair in pair_matches:
        ledger_key = clean(pair.get("ledger_row_key"))
        voucher_key = clean(pair.get("voucher_row_key"))
        pair_key = clean(pair.get("pair_key"))
        if not ledger_key and not voucher_key:
            continue
        if ledger_key:
            ledger_remove.add(ledger_key)
        if voucher_key:
            voucher_remove.add(voucher_key)

        ledger_row = ledger_by_key.get(ledger_key)
        voucher_row = voucher_by_key.get(voucher_key)
        if not ledger_row or not voucher_row:
            continue
        if pair_key and pair_key in matched_pair_keys:
            continue

        merged = dict(ledger_row)
        merged.update(
            {
                "proof_date": voucher_row.get("proof_date", ""),
                "draft_no": voucher_row.get("draft_no", ""),
                "voucher_account_code": voucher_row.get("voucher_account_code", ""),
                "voucher_account_name": voucher_row.get("voucher_account_name", ""),
                "voucher_vendor": voucher_row.get("voucher_vendor", ""),
                "voucher_debit": voucher_row.get("voucher_debit", 0),
                "voucher_credit": voucher_row.get("voucher_credit", 0),
                "voucher_desc": voucher_row.get("voucher_desc", ""),
                "voucher_row_key": voucher_row.get("voucher_row_key", ""),
                "review_reason": "수동쌍매칭",
                "review_memo": clean(pair.get("pair_note")),
                "pair_match_key": pair_key,
                "match_identity_key": build_manual_pair_key(ledger_key, voucher_key),
            }
        )
        matched_rows.append(merged)
        if pair_key:
            matched_pair_keys.add(pair_key)

    if ledger_remove:
        adjusted["ledger_only"]["rows"] = [row for row in ledger_rows if clean(row.get("ledger_row_key")) not in ledger_remove]
    if voucher_remove:
        adjusted["voucher_only"]["rows"] = [row for row in voucher_rows if clean(row.get("voucher_row_key")) not in voucher_remove]

    adjusted["matched"]["count"] = len(adjusted["matched"]["rows"])
    adjusted["ledger_only"]["count"] = len(adjusted["ledger_only"]["rows"])
    adjusted["voucher_only"]["count"] = len(adjusted["voucher_only"]["rows"])
    return adjusted


def save_recheck_review_rows(engine: Any, rows: list[dict[str, Any]]) -> int:
    init_wehago_compare_db(engine)
    normalized_rows = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        review_key = clean(row.get("review_key"))
        if not review_key:
            continue
        normalized_rows.append(
            {
                "review_key": review_key,
                "fiscal_year": int(row.get("fiscal_year") or 0) or None,
                "voucher_no": clean(row.get("voucher_no")),
                "draft_no": clean(row.get("draft_no")),
                "voucher_account_code": clean(row.get("voucher_account_code")),
                "voucher_account_name": clean(row.get("voucher_account_name")),
                "voucher_vendor": clean(row.get("voucher_vendor")),
                "voucher_desc": clean(row.get("voucher_desc")),
                "review_reason": clean(row.get("review_reason")),
                "review_memo": clean(row.get("review_memo")),
            }
        )
    if not normalized_rows:
        return 0
    saved_keys: list[str] = []
    with engine.begin() as conn:
        for item in normalized_rows:
            conn.execute(
                text(
                    """
                    INSERT INTO wehago_recheck_reviews (
                        review_key, fiscal_year, voucher_no, draft_no,
                        voucher_account_code, voucher_account_name, voucher_vendor,
                        voucher_desc, review_reason, review_memo, reviewed_at
                    ) VALUES (
                        :review_key, :fiscal_year, :voucher_no, :draft_no,
                        :voucher_account_code, :voucher_account_name, :voucher_vendor,
                        :voucher_desc, :review_reason, :review_memo, CURRENT_TIMESTAMP
                    )
                    ON CONFLICT(review_key) DO UPDATE SET
                        fiscal_year = excluded.fiscal_year,
                        voucher_no = excluded.voucher_no,
                        draft_no = excluded.draft_no,
                        voucher_account_code = excluded.voucher_account_code,
                        voucher_account_name = excluded.voucher_account_name,
                        voucher_vendor = excluded.voucher_vendor,
                        voucher_desc = excluded.voucher_desc,
                        review_reason = excluded.review_reason,
                        review_memo = excluded.review_memo,
                        reviewed_at = CURRENT_TIMESTAMP
                    """
                ),
                item,
            )
            saved_keys.append(item["review_key"])
        _push_action_history(
            conn,
            "recheck_save",
            {
                "review_keys": saved_keys,
                "count": len(saved_keys),
            },
        )
    _DASHBOARD_CACHE.clear()
    _SUGGEST_CACHE.clear()
    _STATUS_ROWS_CACHE.clear()
    return len(normalized_rows)


def save_manual_pair_matches(
    engine: Any,
    ledger_rows: list[dict[str, Any]],
    voucher_rows: list[dict[str, Any]],
    pair_meta_by_key: dict[str, dict[str, Any]] | None = None,
    match_source: str = "manual",
) -> int:
    init_wehago_compare_db(engine)
    normalized_ledger = [row for row in ledger_rows if isinstance(row, dict) and clean(row.get("ledger_row_key"))]
    normalized_voucher = [row for row in voucher_rows if isinstance(row, dict) and clean(row.get("voucher_row_key"))]
    if not normalized_ledger or not normalized_voucher:
        return 0
    if len(normalized_ledger) != len(normalized_voucher):
        raise ValueError("WEHAGO/ERP 선택 건수를 동일하게 맞춘 뒤 저장해주세요.")

    rows_to_save: list[dict[str, Any]] = []
    for ledger_row, voucher_row in zip(normalized_ledger, normalized_voucher):
        ledger_key = clean(ledger_row.get("ledger_row_key"))
        voucher_key = clean(voucher_row.get("voucher_row_key"))
        pair_key = build_manual_pair_key(ledger_key, voucher_key)
        meta = (pair_meta_by_key or {}).get(pair_key, {})
        fiscal_year = int(ledger_row.get("fiscal_year") or voucher_row.get("fiscal_year") or 0) or None
        rows_to_save.append(
            {
                "pair_key": pair_key,
                "fiscal_year": fiscal_year,
                "ledger_row_key": ledger_key,
                "voucher_row_key": voucher_key,
                "ledger_voucher_no": clean(ledger_row.get("voucher_no")),
                "ledger_account_code": clean(ledger_row.get("ledger_account_code")),
                "ledger_account_name": clean(ledger_row.get("ledger_account_name")),
                "ledger_vendor": clean(ledger_row.get("ledger_vendor")),
                "ledger_debit": parse_amount(ledger_row.get("ledger_debit")),
                "ledger_credit": parse_amount(ledger_row.get("ledger_credit")),
                "ledger_desc": clean(ledger_row.get("ledger_desc")),
                "voucher_no": clean(voucher_row.get("voucher_no")),
                "draft_no": clean(voucher_row.get("draft_no")),
                "voucher_account_code": clean(voucher_row.get("voucher_account_code")),
                "voucher_account_name": clean(voucher_row.get("voucher_account_name")),
                "voucher_vendor": clean(voucher_row.get("voucher_vendor")),
                "voucher_debit": parse_amount(voucher_row.get("voucher_debit")),
                "voucher_credit": parse_amount(voucher_row.get("voucher_credit")),
                "voucher_desc": clean(voucher_row.get("voucher_desc")),
                "match_source": clean(meta.get("match_source")) or clean(match_source) or "manual",
                "confidence_score": float(meta.get("score") or 0),
                "confidence_level": clean(meta.get("confidence_level")),
                "match_reason": clean(meta.get("reason")),
                "pair_note": "",
            }
        )

    saved_pair_keys: list[str] = []
    with engine.begin() as conn:
        for item in rows_to_save:
            conn.execute(
                text(
                    """
                    INSERT INTO wehago_manual_pair_matches (
                        pair_key, fiscal_year, ledger_row_key, voucher_row_key,
                        ledger_voucher_no, ledger_account_code, ledger_account_name, ledger_vendor,
                        ledger_debit, ledger_credit, ledger_desc,
                        voucher_no, draft_no, voucher_account_code, voucher_account_name, voucher_vendor,
                        voucher_debit, voucher_credit, voucher_desc,
                        match_source, confidence_score, confidence_level, match_reason,
                        pair_note, created_at
                    ) VALUES (
                        :pair_key, :fiscal_year, :ledger_row_key, :voucher_row_key,
                        :ledger_voucher_no, :ledger_account_code, :ledger_account_name, :ledger_vendor,
                        :ledger_debit, :ledger_credit, :ledger_desc,
                        :voucher_no, :draft_no, :voucher_account_code, :voucher_account_name, :voucher_vendor,
                        :voucher_debit, :voucher_credit, :voucher_desc,
                        :match_source, :confidence_score, :confidence_level, :match_reason,
                        :pair_note, CURRENT_TIMESTAMP
                    )
                    ON CONFLICT(pair_key) DO UPDATE SET
                        fiscal_year = excluded.fiscal_year,
                        ledger_row_key = excluded.ledger_row_key,
                        voucher_row_key = excluded.voucher_row_key,
                        ledger_voucher_no = excluded.ledger_voucher_no,
                        ledger_account_code = excluded.ledger_account_code,
                        ledger_account_name = excluded.ledger_account_name,
                        ledger_vendor = excluded.ledger_vendor,
                        ledger_debit = excluded.ledger_debit,
                        ledger_credit = excluded.ledger_credit,
                        ledger_desc = excluded.ledger_desc,
                        voucher_no = excluded.voucher_no,
                        draft_no = excluded.draft_no,
                        voucher_account_code = excluded.voucher_account_code,
                        voucher_account_name = excluded.voucher_account_name,
                        voucher_vendor = excluded.voucher_vendor,
                        voucher_debit = excluded.voucher_debit,
                        voucher_credit = excluded.voucher_credit,
                        voucher_desc = excluded.voucher_desc,
                        match_source = excluded.match_source,
                        confidence_score = excluded.confidence_score,
                        confidence_level = excluded.confidence_level,
                        match_reason = excluded.match_reason,
                        pair_note = excluded.pair_note,
                        created_at = CURRENT_TIMESTAMP
                    """
                ),
                item,
            )
            saved_pair_keys.append(item["pair_key"])
        _push_action_history(
            conn,
            "pair_save",
            {
                "pair_keys": saved_pair_keys,
                "count": len(saved_pair_keys),
                "match_source": clean(match_source) or "manual",
            },
        )
    _DASHBOARD_CACHE.clear()
    _SUGGEST_CACHE.clear()
    _STATUS_ROWS_CACHE.clear()
    return len(rows_to_save)


def get_last_action_summary(engine: Any | None = None, conn: Any | None = None) -> dict[str, Any] | None:
    if conn is None:
        if engine is None:
            return None
        init_wehago_compare_db(engine)
        with engine.begin() as local_conn:
            return get_last_action_summary(conn=local_conn)
    row = conn.execute(
        text(
            """
            SELECT id, action_type, payload_json, created_at
            FROM wehago_action_history
            ORDER BY id DESC
            LIMIT 1
            """
        )
    ).mappings().first()
    if not row:
        return None
    payload = {}
    try:
        payload = json.loads(row["payload_json"] or "{}")
    except json.JSONDecodeError:
        payload = {}
    count = int(payload.get("count") or 0)
    return {
        "id": int(row["id"]),
        "action_type": clean(row["action_type"]),
        "count": count,
        "created_at": clean(row["created_at"]),
    }


def undo_last_action(engine: Any) -> dict[str, Any]:
    init_wehago_compare_db(engine)
    with engine.begin() as conn:
        row = conn.execute(
            text(
                """
                SELECT id, action_type, payload_json
                FROM wehago_action_history
                ORDER BY id DESC
                LIMIT 1
                """
            )
        ).mappings().first()
        if not row:
            return {"undone": False, "message": "되돌릴 작업이 없습니다."}
        action_type = clean(row["action_type"])
        payload = {}
        try:
            payload = json.loads(row["payload_json"] or "{}")
        except json.JSONDecodeError:
            payload = {}
        affected = 0
        if action_type == "recheck_save":
            keys = [clean(key) for key in payload.get("review_keys", []) if clean(key)]
            if keys:
                placeholders = ", ".join(f":k{i}" for i in range(len(keys)))
                params = {f"k{i}": key for i, key in enumerate(keys)}
                result = conn.execute(text(f"DELETE FROM wehago_recheck_reviews WHERE review_key IN ({placeholders})"), params)
                affected = int(result.rowcount or 0)
        elif action_type == "pair_save":
            keys = [clean(key) for key in payload.get("pair_keys", []) if clean(key)]
            if keys:
                placeholders = ", ".join(f":k{i}" for i in range(len(keys)))
                params = {f"k{i}": key for i, key in enumerate(keys)}
                result = conn.execute(text(f"DELETE FROM wehago_manual_pair_matches WHERE pair_key IN ({placeholders})"), params)
                affected = int(result.rowcount or 0)
        conn.execute(text("DELETE FROM wehago_action_history WHERE id = :id"), {"id": row["id"]})
    _DASHBOARD_CACHE.clear()
    _SUGGEST_CACHE.clear()
    _STATUS_ROWS_CACHE.clear()
    return {"undone": True, "action_type": action_type, "affected": affected}


def _parse_iso_date(value: Any) -> date | None:
    parsed = parse_excel_date(value)
    if not parsed:
        return None
    try:
        return datetime.strptime(parsed, "%Y-%m-%d").date()
    except ValueError:
        return None


def _tokenize_for_similarity(value: Any) -> set[str]:
    normalized = normalize_text(value)
    if not normalized:
        return set()
    if len(normalized) <= 2:
        return {normalized}
    return {normalized[i : i + 2] for i in range(len(normalized) - 1)}


def _jaccard_similarity(left: Any, right: Any) -> float:
    left_tokens = _tokenize_for_similarity(left)
    right_tokens = _tokenize_for_similarity(right)
    if not left_tokens or not right_tokens:
        return 0.0
    inter = len(left_tokens & right_tokens)
    union = len(left_tokens | right_tokens)
    return (inter / union) if union else 0.0


def _numeric_amount_for_side(row: dict[str, Any], side: str) -> float:
    if side == "debit":
        return parse_amount(row.get("ledger_debit") if "ledger_debit" in row else row.get("voucher_debit"))
    if side == "credit":
        return parse_amount(row.get("ledger_credit") if "ledger_credit" in row else row.get("voucher_credit"))
    return max(
        parse_amount(row.get("ledger_debit") if "ledger_debit" in row else row.get("voucher_debit")),
        parse_amount(row.get("ledger_credit") if "ledger_credit" in row else row.get("voucher_credit")),
    )


def _determine_primary_side(ledger_row: dict[str, Any]) -> str:
    debit = parse_amount(ledger_row.get("ledger_debit"))
    credit = parse_amount(ledger_row.get("ledger_credit"))
    if debit > 0 and credit <= 0:
        return "debit"
    if credit > 0 and debit <= 0:
        return "credit"
    return "either"


def _score_pair_match(ledger_row: dict[str, Any], voucher_row: dict[str, Any]) -> dict[str, Any]:
    score = 0.0
    reasons: list[str] = []
    hard_pass = True

    side = _determine_primary_side(ledger_row)
    if side == "debit":
        ledger_amount = parse_amount(ledger_row.get("ledger_debit"))
        voucher_amount = parse_amount(voucher_row.get("voucher_debit"))
    elif side == "credit":
        ledger_amount = parse_amount(ledger_row.get("ledger_credit"))
        voucher_amount = parse_amount(voucher_row.get("voucher_credit"))
    else:
        ledger_amount = max(parse_amount(ledger_row.get("ledger_debit")), parse_amount(ledger_row.get("ledger_credit")))
        voucher_amount = max(parse_amount(voucher_row.get("voucher_debit")), parse_amount(voucher_row.get("voucher_credit")))

    amount_gap = abs(ledger_amount - voucher_amount)
    if amount_gap < 0.5 and ledger_amount > 0:
        score += 50
        reasons.append("금액 일치")
    elif amount_gap < 5 and ledger_amount > 0:
        score += 35
        reasons.append("금액 근접")
    elif amount_gap < 100 and ledger_amount > 0:
        score += 10
        reasons.append("금액 유사")
    else:
        hard_pass = False

    ledger_code = clean(ledger_row.get("ledger_account_code"))
    voucher_code = clean(voucher_row.get("voucher_account_code"))
    account_sim = _jaccard_similarity(ledger_row.get("ledger_account_name"), voucher_row.get("voucher_account_name"))
    if ledger_code and voucher_code and ledger_code == voucher_code:
        score += 22
        reasons.append("계정코드 일치")
    elif ledger_code and voucher_code and ledger_code[:4] == voucher_code[:4]:
        score += 10
        reasons.append("계정코드 대분류 일치")
    else:
        if account_sim >= 0.8:
            score += 16
            reasons.append("계정명 유사도 높음")
        elif account_sim >= 0.55:
            score += 8
            reasons.append("계정명 유사")
        else:
            hard_pass = False

    vendor_sim = _jaccard_similarity(ledger_row.get("ledger_vendor"), voucher_row.get("voucher_vendor"))
    if vendor_sim >= 0.9:
        score += 16
        reasons.append("거래처 일치")
    elif vendor_sim >= 0.65:
        score += 10
        reasons.append("거래처 유사")
    elif vendor_sim >= 0.4:
        score += 4
        reasons.append("거래처 일부 유사")
    else:
        score -= 8

    desc_sim = _jaccard_similarity(ledger_row.get("ledger_desc"), voucher_row.get("voucher_desc"))
    if desc_sim >= 0.85:
        score += 10
        reasons.append("적요 매우 유사")
    elif desc_sim >= 0.6:
        score += 6
        reasons.append("적요 유사")
    elif desc_sim >= 0.35:
        score += 2

    ledger_date = _parse_iso_date(ledger_row.get("ledger_date"))
    proof_date = _parse_iso_date(voucher_row.get("proof_date"))
    if ledger_date and proof_date:
        day_gap = abs((ledger_date - proof_date).days)
        if day_gap <= 3:
            score += 8
            reasons.append("일자 근접")
        elif day_gap <= 10:
            score += 4
        elif day_gap <= 45:
            score += 1
        else:
            score -= 6

    confidence = "low"
    if score >= 88:
        confidence = "high"
    elif score >= 72:
        confidence = "medium"

    auto_eligible = bool(
        hard_pass
        and score >= 88
        and amount_gap < 0.5
        and (
            (ledger_code and voucher_code and ledger_code == voucher_code)
            or _jaccard_similarity(ledger_row.get("ledger_account_name"), voucher_row.get("voucher_account_name")) >= 0.8
        )
        and vendor_sim >= 0.65
    )
    return {
        "score": round(score, 2),
        "confidence_level": confidence,
        "reason": ", ".join(reasons[:4]),
        "auto_eligible": auto_eligible,
        "hard_pass": hard_pass,
        "vendor_similarity": round(vendor_sim, 4),
        "account_similarity": round(account_sim, 4),
        "amount_gap": round(amount_gap, 2),
    }


def _collect_status_rows_for_workbench(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    ledger_voucher_no: str = "",
    ledger_review_reason: str = "",
    voucher_voucher_no: str = "",
    voucher_review_reason: str = "",
) -> dict[str, list[dict[str, Any]]]:
    if start_year is None or end_year is None:
        return {"ledger_only": [], "voucher_only": []}
    if engine is not None:
        init_wehago_compare_db(engine)
    reviewed_keys: set[str] = set()
    manual_pair_matches: list[dict[str, Any]] = []
    if engine is not None:
        with engine.begin() as conn:
            reviewed_keys = get_saved_recheck_review_keys(conn, start_year, end_year)
            manual_pair_matches = get_saved_manual_pair_matches(conn, start_year, end_year)

    result: dict[str, list[dict[str, Any]]] = {"ledger_only": [], "voucher_only": []}
    ledger_voucher_filter = normalize_text(ledger_voucher_no)
    ledger_reason_filter = normalize_text(ledger_review_reason)
    voucher_voucher_filter = normalize_text(voucher_voucher_no)
    voucher_reason_filter = normalize_text(voucher_review_reason)

    for year in range(start_year, end_year + 1):
        bundle = discover_compare_result_bundle(year)
        if not bundle:
            continue
        parsed = parse_compare_result_bundle(
            str(bundle["ledger_result"]),
            bundle["ledger_result"].stat().st_mtime,
            str(bundle["voucher_result"]),
            bundle["voucher_result"].stat().st_mtime,
            year,
        )
        parsed = apply_saved_recheck_reviews(parsed, reviewed_keys)
        parsed = apply_saved_manual_pair_matches(parsed, manual_pair_matches)
        for row in parsed["ledger_only"]["rows"]:
            if _filter_status_row(
                row,
                ledger_voucher_filter,
                "",
                "",
                "",
                "",
                "",
                "",
                "",
                ledger_reason_filter,
            ):
                result["ledger_only"].append(row)
        for row in parsed["voucher_only"]["rows"]:
            if _filter_status_row(
                row,
                voucher_voucher_filter,
                "",
                "",
                "",
                "",
                "",
                "",
                "",
                voucher_reason_filter,
            ):
                result["voucher_only"].append(row)
    return result


def recommend_pair_matches(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    ledger_voucher_no: str = "",
    ledger_review_reason: str = "",
    voucher_voucher_no: str = "",
    voucher_review_reason: str = "",
    limit: int = 300,
) -> dict[str, Any]:
    dataset = _collect_status_rows_for_workbench(
        engine,
        start_year,
        end_year,
        ledger_voucher_no=ledger_voucher_no,
        ledger_review_reason=ledger_review_reason,
        voucher_voucher_no=voucher_voucher_no,
        voucher_review_reason=voucher_review_reason,
    )
    ledger_rows = dataset["ledger_only"]
    voucher_rows = dataset["voucher_only"]
    if not ledger_rows or not voucher_rows:
        return {"pairs": [], "stats": {"ledger_rows": len(ledger_rows), "voucher_rows": len(voucher_rows), "recommended": 0, "auto_eligible": 0}}

    amount_index: dict[float, list[dict[str, Any]]] = {}
    for voucher_row in voucher_rows:
        amounts = {
            round(parse_amount(voucher_row.get("voucher_debit")), 2),
            round(parse_amount(voucher_row.get("voucher_credit")), 2),
        }
        for amount in amounts:
            if amount <= 0:
                continue
            amount_index.setdefault(amount, []).append(voucher_row)

    edge_candidates: list[dict[str, Any]] = []
    for ledger_row in ledger_rows:
        candidate_amounts = {
            round(parse_amount(ledger_row.get("ledger_debit")), 2),
            round(parse_amount(ledger_row.get("ledger_credit")), 2),
        }
        voucher_candidates: list[dict[str, Any]] = []
        seen_keys: set[str] = set()
        for amount in candidate_amounts:
            if amount <= 0:
                continue
            for voucher_row in amount_index.get(amount, []):
                voucher_key = clean(voucher_row.get("voucher_row_key"))
                if voucher_key and voucher_key not in seen_keys:
                    seen_keys.add(voucher_key)
                    voucher_candidates.append(voucher_row)
        if not voucher_candidates:
            continue
        scored_candidates: list[tuple[dict[str, Any], dict[str, Any]]] = []
        for voucher_row in voucher_candidates:
            score_result = _score_pair_match(ledger_row, voucher_row)
            if not score_result["hard_pass"] or score_result["score"] < 72:
                continue
            scored_candidates.append((voucher_row, score_result))
        if not scored_candidates:
            continue
        scored_candidates.sort(key=lambda item: item[1]["score"], reverse=True)
        top_candidate, top_score = scored_candidates[0]
        second_score = scored_candidates[1][1]["score"] if len(scored_candidates) > 1 else -999
        if top_score["score"] - second_score < 6:
            continue
        edge_candidates.append(
            {
                "ledger_row": ledger_row,
                "voucher_row": top_candidate,
                "score": top_score["score"],
                "confidence_level": top_score["confidence_level"],
                "reason": top_score["reason"],
                "auto_eligible": top_score["auto_eligible"],
                "vendor_similarity": top_score.get("vendor_similarity", 0),
                "account_similarity": top_score.get("account_similarity", 0),
                "amount_gap": top_score.get("amount_gap", 0),
            }
        )

    edge_candidates.sort(key=lambda item: item["score"], reverse=True)
    matched_ledger: set[str] = set()
    matched_voucher: set[str] = set()
    picked: list[dict[str, Any]] = []
    for edge in edge_candidates:
        ledger_key = clean(edge["ledger_row"].get("ledger_row_key"))
        voucher_key = clean(edge["voucher_row"].get("voucher_row_key"))
        if not ledger_key or not voucher_key:
            continue
        if ledger_key in matched_ledger or voucher_key in matched_voucher:
            continue
        matched_ledger.add(ledger_key)
        matched_voucher.add(voucher_key)
        pair_key = build_manual_pair_key(ledger_key, voucher_key)
        picked.append(
            {
                "pair_key": pair_key,
                "ledger_row_key": ledger_key,
                "voucher_row_key": voucher_key,
                "score": edge["score"],
                "confidence_level": edge["confidence_level"],
                "reason": edge["reason"],
                "auto_eligible": edge["auto_eligible"],
                "vendor_similarity": edge.get("vendor_similarity", 0),
                "account_similarity": edge.get("account_similarity", 0),
                "amount_gap": edge.get("amount_gap", 0),
                "ledger_row": edge["ledger_row"],
                "voucher_row": edge["voucher_row"],
            }
        )
        if len(picked) >= max(min(int(limit or 300), 1000), 1):
            break

    return {
        "pairs": picked,
        "stats": {
            "ledger_rows": len(ledger_rows),
            "voucher_rows": len(voucher_rows),
            "recommended": len(picked),
            "auto_eligible": sum(1 for row in picked if row["auto_eligible"]),
            "high_confidence": sum(1 for row in picked if row["confidence_level"] == "high"),
        },
    }


def save_recommended_pair_matches(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    pair_keys: list[str] | None = None,
    auto_only: bool = False,
    ledger_voucher_no: str = "",
    ledger_review_reason: str = "",
    voucher_voucher_no: str = "",
    voucher_review_reason: str = "",
) -> dict[str, Any]:
    recommended = recommend_pair_matches(
        engine,
        start_year=start_year,
        end_year=end_year,
        ledger_voucher_no=ledger_voucher_no,
        ledger_review_reason=ledger_review_reason,
        voucher_voucher_no=voucher_voucher_no,
        voucher_review_reason=voucher_review_reason,
        limit=1000,
    )
    selected_keys = {clean(key) for key in (pair_keys or []) if clean(key)}
    selected_pairs: list[dict[str, Any]] = []
    for pair in recommended["pairs"]:
        if auto_only and not pair.get("auto_eligible"):
            continue
        if selected_keys and clean(pair.get("pair_key")) not in selected_keys:
            continue
        selected_pairs.append(pair)

    ledger_rows = [pair["ledger_row"] for pair in selected_pairs]
    voucher_rows = [pair["voucher_row"] for pair in selected_pairs]
    pair_meta_by_key = {
        clean(pair["pair_key"]): {
            "score": pair["score"],
            "confidence_level": pair["confidence_level"],
            "reason": pair["reason"],
            "match_source": "auto_recommend" if auto_only else "recommend",
        }
        for pair in selected_pairs
    }
    saved_count = save_manual_pair_matches(
        engine,
        ledger_rows=ledger_rows,
        voucher_rows=voucher_rows,
        pair_meta_by_key=pair_meta_by_key,
        match_source="auto_recommend" if auto_only else "recommend",
    )
    return {
        "saved_count": saved_count,
        "selected_count": len(selected_pairs),
        "recommended_count": recommended["stats"]["recommended"],
    }


def get_individual_pair_recommendations(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    source_status: str,
    source_row_key: str,
    offset: int = 0,
    limit: int = 10,
) -> dict[str, Any]:
    source_status = normalize_text(source_status).lower()
    if source_status not in {"ledgeronly", "voucheronly", "ledger_only", "voucher_only"}:
        raise ValueError("개별 추천의 상태 값이 올바르지 않습니다.")
    normalized_source_status = "ledger_only" if source_status in {"ledgeronly", "ledger_only"} else "voucher_only"
    normalized_row_key = clean(source_row_key)
    if not normalized_row_key:
        raise ValueError("추천할 기준 행 키가 필요합니다.")

    dataset = _collect_status_rows_for_workbench(engine, start_year, end_year)
    ledger_rows = dataset["ledger_only"]
    voucher_rows = dataset["voucher_only"]

    source_row = None
    if normalized_source_status == "ledger_only":
        for row in ledger_rows:
            if clean(row.get("ledger_row_key")) == normalized_row_key:
                source_row = row
                break
    else:
        for row in voucher_rows:
            if clean(row.get("voucher_row_key")) == normalized_row_key:
                source_row = row
                break
    if source_row is None:
        return {
            "source_status": normalized_source_status,
            "source_row": None,
            "rows": [],
            "total_count": 0,
            "shown_count": 0,
            "offset": 0,
            "limit": limit,
            "has_more": False,
            "next_offset": 0,
        }

    candidates: list[dict[str, Any]] = []
    if normalized_source_status == "ledger_only":
        source_amounts = {
            round(parse_amount(source_row.get("ledger_debit")), 2),
            round(parse_amount(source_row.get("ledger_credit")), 2),
        }
        for target in voucher_rows:
            target_amounts = {
                round(parse_amount(target.get("voucher_debit")), 2),
                round(parse_amount(target.get("voucher_credit")), 2),
            }
            if not (source_amounts & target_amounts):
                continue
            score = _score_pair_match(source_row, target)
            if not score["hard_pass"] or score["score"] < 60:
                continue
            candidates.append(
                {
                    "pair_key": build_manual_pair_key(clean(source_row.get("ledger_row_key")), clean(target.get("voucher_row_key"))),
                    "score": score["score"],
                    "confidence_level": score["confidence_level"],
                    "amount_gap": score.get("amount_gap", 0),
                    "account_similarity": score.get("account_similarity", 0),
                    "vendor_similarity": score.get("vendor_similarity", 0),
                    "reason": score.get("reason", ""),
                    "target_row": target,
                }
            )
    else:
        source_amounts = {
            round(parse_amount(source_row.get("voucher_debit")), 2),
            round(parse_amount(source_row.get("voucher_credit")), 2),
        }
        for target in ledger_rows:
            target_amounts = {
                round(parse_amount(target.get("ledger_debit")), 2),
                round(parse_amount(target.get("ledger_credit")), 2),
            }
            if not (source_amounts & target_amounts):
                continue
            score = _score_pair_match(target, source_row)
            if not score["hard_pass"] or score["score"] < 60:
                continue
            candidates.append(
                {
                    "pair_key": build_manual_pair_key(clean(target.get("ledger_row_key")), clean(source_row.get("voucher_row_key"))),
                    "score": score["score"],
                    "confidence_level": score["confidence_level"],
                    "amount_gap": score.get("amount_gap", 0),
                    "account_similarity": score.get("account_similarity", 0),
                    "vendor_similarity": score.get("vendor_similarity", 0),
                    "reason": score.get("reason", ""),
                    "target_row": target,
                }
            )

    candidates.sort(
        key=lambda row: (
            -float(row.get("score") or 0),
            float(row.get("amount_gap") or 0),
            -float(row.get("vendor_similarity") or 0),
        )
    )
    safe_offset = max(int(offset or 0), 0)
    safe_limit = max(min(int(limit or 10), 100), 1)
    rows = candidates[safe_offset : safe_offset + safe_limit]
    next_offset = safe_offset + len(rows)
    return {
        "source_status": normalized_source_status,
        "source_row": source_row,
        "rows": rows,
        "total_count": len(candidates),
        "shown_count": len(rows),
        "offset": safe_offset,
        "limit": safe_limit,
        "has_more": next_offset < len(candidates),
        "next_offset": next_offset,
    }


def _contains_filter(value: Any, keyword: str) -> bool:
    if not keyword:
        return True
    return keyword in normalize_text(value)


def _filter_status_row(
    row: dict[str, Any],
    voucher_no: str,
    draft_no: str,
    wehago_account: str,
    erp_account: str,
    wehago_amount: str,
    erp_amount: str,
    wehago_vendor: str,
    erp_vendor: str,
    desc_keyword: str,
) -> bool:
    if voucher_no and not _contains_filter(row.get("voucher_no"), voucher_no):
        return False
    if draft_no and not _contains_filter(row.get("draft_no"), draft_no):
        return False
    wehago_account_candidates = [
        row.get("ledger_account_code"),
        row.get("ledger_account_name"),
        " ".join(part for part in (clean(row.get("ledger_account_code")), clean(row.get("ledger_account_name"))) if part),
    ]
    if wehago_account and not any(_contains_filter(item, wehago_account) for item in wehago_account_candidates):
        return False
    erp_account_candidates = [
        row.get("voucher_account_code"),
        row.get("voucher_account_name"),
        " ".join(part for part in (clean(row.get("voucher_account_code")), clean(row.get("voucher_account_name"))) if part),
    ]
    if erp_account and not any(_contains_filter(item, erp_account) for item in erp_account_candidates):
        return False
    if wehago_vendor and not _contains_filter(row.get("ledger_vendor"), wehago_vendor):
        return False
    if erp_vendor and not _contains_filter(row.get("voucher_vendor"), erp_vendor):
        return False
    if wehago_amount:
        target = parse_amount(wehago_amount)
        if abs(parse_amount(row.get("ledger_debit")) - target) >= 0.5 and abs(parse_amount(row.get("ledger_credit")) - target) >= 0.5:
            return False
    if erp_amount:
        target = parse_amount(erp_amount)
        if abs(parse_amount(row.get("voucher_debit")) - target) >= 0.5 and abs(parse_amount(row.get("voucher_credit")) - target) >= 0.5:
            return False
    desc_candidates = [
        row.get("ledger_desc"),
        row.get("voucher_desc"),
        row.get("review_reason"),
        row.get("review_memo"),
    ]
    if desc_keyword and not any(_contains_filter(item, desc_keyword) for item in desc_candidates):
        return False
    return True


def _apply_broad_query_guard(
    rows: list[dict[str, Any]],
    voucher_no: str,
    account_code: str,
    vendor_name: str,
    review_reason: str,
    preview_limit: int = 500,
) -> tuple[list[dict[str, Any]], str]:
    if voucher_no or account_code or vendor_name or review_reason:
        return rows, ""
    if len(rows) <= preview_limit:
        return rows, ""
    return rows[:preview_limit], f"조건 없이 조회된 항목이 많아 상위 {preview_limit:,}건만 먼저 표시합니다. 전표번호, 계정코드, 거래처, 검증근거로 좁혀서 전체 항목에 접근할 수 있습니다."


def get_status_detail_rows(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    status: str,
    voucher_no: str = "",
    draft_no: str = "",
    wehago_account: str = "",
    erp_account: str = "",
    wehago_amount: str = "",
    erp_amount: str = "",
    wehago_vendor: str = "",
    erp_vendor: str = "",
    desc_keyword: str = "",
    review_reason: str = "",
    offset: int = 0,
    limit: int = 200,
) -> dict[str, Any]:
    if engine is not None:
        init_wehago_compare_db(engine)
    if start_year is None or end_year is None:
        return {
            "columns": DETAIL_COLUMN_MAP.get(status, []),
            "rows": [],
            "total_count": 0,
            "shown_count": 0,
            "offset": 0,
            "limit": limit,
            "has_more": False,
            "next_offset": 0,
        }

    normalized_status = normalize_text(status).lower()
    allowed = {
        "matched": "matched",
        "ledgeronly": "ledger_only",
        "amountmismatch": "amount_mismatch",
        "voucheronly": "voucher_only",
    }
    if normalized_status not in {"matched", "ledger_only", "amount_mismatch", "voucher_only"}:
        normalized_status = allowed.get(normalized_status, normalized_status)
    if normalized_status not in {"matched", "ledger_only", "amount_mismatch", "voucher_only"}:
        raise ValueError("상태 값이 올바르지 않습니다.")

    voucher_filter = normalize_text(voucher_no)
    draft_filter = normalize_text(draft_no)
    wehago_account_filter = normalize_text(wehago_account)
    erp_account_filter = normalize_text(erp_account)
    wehago_vendor_filter = normalize_text(wehago_vendor)
    erp_vendor_filter = normalize_text(erp_vendor)
    desc_filter = normalize_text(desc_keyword or review_reason)

    safe_offset = max(int(offset or 0), 0)
    safe_limit = max(min(int(limit or 200), 500), 1)
    has_filters = any(
        [
            voucher_filter,
            draft_filter,
            wehago_account_filter,
            erp_account_filter,
            clean(wehago_amount),
            clean(erp_amount),
            wehago_vendor_filter,
            erp_vendor_filter,
            desc_filter,
        ]
    )

    if not has_filters:
        columns = DETAIL_COLUMN_MAP[normalized_status]
        rows_by_status = _get_cached_status_rows_by_range(engine, start_year, end_year)
        total_count = 0
        rows_out: list[dict[str, Any]] = []
        skipped = 0
        needed = safe_limit
        section_rows = rows_by_status.get(normalized_status, [])
        total_count = len(section_rows)
        if section_rows:
            rows_out = section_rows[safe_offset : safe_offset + safe_limit]
        next_offset = safe_offset + len(rows_out)
        return {
            "columns": columns,
            "rows": rows_out,
            "total_count": total_count,
            "shown_count": len(rows_out),
            "offset": safe_offset,
            "limit": safe_limit,
            "has_more": next_offset < total_count,
            "next_offset": next_offset,
            "notice": "",
        }

    rows_by_status = _get_cached_status_rows_by_range(engine, start_year, end_year)
    collected: list[dict[str, Any]] = []
    columns = DETAIL_COLUMN_MAP[normalized_status]
    for row in rows_by_status.get(normalized_status, []):
        if _filter_status_row(
            row,
            voucher_filter,
            draft_filter,
            wehago_account_filter,
            erp_account_filter,
            clean(wehago_amount),
            clean(erp_amount),
            wehago_vendor_filter,
            erp_vendor_filter,
            desc_filter,
        ):
            collected.append(row)
    shown_rows = collected[safe_offset : safe_offset + safe_limit]
    next_offset = safe_offset + len(shown_rows)
    return {
        "columns": columns,
        "rows": shown_rows,
        "total_count": len(collected),
        "shown_count": len(shown_rows),
        "offset": safe_offset,
        "limit": safe_limit,
        "has_more": next_offset < len(collected),
        "next_offset": next_offset,
        "notice": "",
    }


def get_status_field_suggestions(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    status: str,
    field: str = "voucher_no",
    keyword: str = "",
    offset: int = 0,
    limit: int = 10,
) -> dict[str, Any]:
    if start_year is None or end_year is None:
        return {"rows": [], "total_count": 0, "shown_count": 0, "offset": 0, "limit": limit, "has_more": False, "next_offset": 0}
    normalized_status = normalize_text(status).lower()
    status_map = {
        "matched": "matched",
        "amountmismatch": "amount_mismatch",
        "ledgeronly": "ledger_only",
        "voucheronly": "voucher_only",
        "ledger_only": "ledger_only",
        "voucher_only": "voucher_only",
        "amount_mismatch": "amount_mismatch",
    }
    normalized_status = status_map.get(normalized_status, normalized_status)
    if normalized_status not in {"matched", "amount_mismatch", "ledger_only", "voucher_only"}:
        raise ValueError("자동완성 상태 값이 올바르지 않습니다.")

    normalized_field = normalize_text(field).lower()
    if normalized_field not in {
        "voucherno",
        "draftno",
        "account",
        "vendor",
        "wehagoaccount",
        "erpaccount",
        "wehagovendor",
        "erpvendor",
    }:
        raise ValueError("자동완성 필드 값이 올바르지 않습니다.")
    cache_key = "|".join(
        [
            normalized_status,
            normalized_field,
            str(start_year),
            str(end_year),
            _build_bundle_signature(start_year, end_year),
            normalize_text(keyword),
        ]
    )
    now = time.time()
    cached = _SUGGEST_CACHE.get(cache_key)
    if cached and (now - float(cached.get("ts", 0))) <= _SUGGEST_CACHE_TTL_SEC:
        all_rows = cached["rows"]
    else:
        rows_by_status = _get_cached_status_rows_by_range(engine, start_year, end_year)
        keyword_norm = normalize_text(keyword)
        seen: set[str] = set()
        all_rows: list[dict[str, Any]] = []
        for row in rows_by_status.get(normalized_status, []):
            code = clean(row.get("voucher_account_code"))
            name = clean(row.get("voucher_account_name"))
            voucher_no = clean(row.get("voucher_no"))
            if normalized_field == "account":
                if normalized_status == "voucher_only":
                    code = clean(row.get("voucher_account_code"))
                    name = clean(row.get("voucher_account_name"))
                else:
                    code = clean(row.get("ledger_account_code"))
                    name = clean(row.get("ledger_account_name"))
                key = "|".join([code, name])
                if not code and not name:
                    continue
                label = " ".join(part for part in [code, name] if part)
                search_text = normalize_text(label)
            elif normalized_field == "wehagoaccount":
                code = clean(row.get("ledger_account_code"))
                name = clean(row.get("ledger_account_name"))
                key = "|".join([code, name])
                if not code and not name:
                    continue
                label = " ".join(part for part in [code, name] if part)
                search_text = normalize_text(label)
            elif normalized_field == "erpaccount":
                code = clean(row.get("voucher_account_code"))
                name = clean(row.get("voucher_account_name"))
                key = "|".join([code, name])
                if not code and not name:
                    continue
                label = " ".join(part for part in [code, name] if part)
                search_text = normalize_text(label)
            elif normalized_field == "vendor":
                vendor_name = clean(row.get("voucher_vendor")) if normalized_status == "voucher_only" else clean(row.get("ledger_vendor"))
                key = vendor_name
                if not vendor_name:
                    continue
                label = vendor_name
                search_text = normalize_text(vendor_name)
            elif normalized_field == "wehagovendor":
                vendor_name = clean(row.get("ledger_vendor"))
                key = vendor_name
                if not vendor_name:
                    continue
                label = vendor_name
                search_text = normalize_text(vendor_name)
            elif normalized_field == "erpvendor":
                vendor_name = clean(row.get("voucher_vendor"))
                key = vendor_name
                if not vendor_name:
                    continue
                label = vendor_name
                search_text = normalize_text(vendor_name)
            elif normalized_field == "draftno":
                draft_no = clean(row.get("draft_no"))
                key = draft_no
                if not draft_no:
                    continue
                label = draft_no
                search_text = normalize_text(draft_no)
            else:
                key = voucher_no
                if not voucher_no:
                    continue
                label = voucher_no
                search_text = normalize_text(voucher_no)
            if key in seen:
                continue
            if keyword_norm and keyword_norm not in search_text:
                continue
            seen.add(key)
            all_rows.append(
                {
                    "voucher_no": voucher_no,
                    "code": code,
                    "name": name,
                    "label": label,
                }
            )
        _SUGGEST_CACHE.clear()
        _SUGGEST_CACHE[cache_key] = {"ts": now, "rows": all_rows}

    safe_offset = max(int(offset or 0), 0)
    safe_limit = max(min(int(limit or 10), 50), 1)
    rows = all_rows[safe_offset : safe_offset + safe_limit]
    next_offset = safe_offset + len(rows)
    return {
        "rows": rows,
        "total_count": len(all_rows),
        "shown_count": len(rows),
        "offset": safe_offset,
        "limit": safe_limit,
        "has_more": next_offset < len(all_rows),
        "next_offset": next_offset,
    }


def _fetch_filtered_table_rows(
    conn: Any,
    table_name: str,
    columns: list[tuple[str, str]],
    start_year: int | None,
    end_year: int | None,
    voucher_no: str = "",
    account_code: str = "",
    vendor_name: str = "",
    review_reason: str = "",
) -> dict[str, Any]:
    del review_reason
    voucher_filter = f"%{clean(voucher_no)}%"
    account_filter = f"%{clean(account_code)}%"
    vendor_filter = f"%{clean(vendor_name)}%"

    if table_name == "wehago_ledger_rows":
        filters = [
            build_year_filter_sql(),
            "(:voucher_no = '' OR COALESCE(voucher_no, '') LIKE :voucher_like)",
            "(:account_code = '' OR COALESCE(account_code, '') LIKE :account_like OR COALESCE(account_name, '') LIKE :account_like)",
            "(:vendor_name = '' OR COALESCE(vendor_name, '') LIKE :vendor_like OR COALESCE(description, '') LIKE :vendor_like)",
        ]
        order_sql = "ORDER BY COALESCE(ledger_date, '') DESC, voucher_no DESC, row_number DESC"
    else:
        filters = [
            build_year_filter_sql(),
            "(:voucher_no = '' OR COALESCE(confirmed_no, '') LIKE :voucher_like OR COALESCE(draft_no, '') LIKE :voucher_like)",
            "(:account_code = '' OR COALESCE(account_code, '') LIKE :account_like OR COALESCE(account_name, '') LIKE :account_like)",
            "(:vendor_name = '' OR COALESCE(vendor_name, '') LIKE :vendor_like OR COALESCE(desc1, '') LIKE :vendor_like OR COALESCE(desc2, '') LIKE :vendor_like)",
        ]
        order_sql = "ORDER BY COALESCE(proof_date, '') DESC, confirmed_no DESC, draft_no DESC, row_number DESC"

    params = {
        "start_year": start_year,
        "end_year": end_year,
        "voucher_no": clean(voucher_no),
        "voucher_like": voucher_filter,
        "account_code": clean(account_code),
        "account_like": account_filter,
        "vendor_name": clean(vendor_name),
        "vendor_like": vendor_filter,
    }
    selected_columns = ", ".join(name for name, _ in columns)
    query = text(
        f"""
        SELECT {selected_columns}
        FROM {table_name}
        WHERE {' AND '.join(filters)}
        {order_sql}
        """
    )
    rows = [dict(row._mapping) for row in conn.execute(query, params).fetchall()]
    shown_rows, notice = _apply_broad_query_guard(
        rows,
        clean(voucher_no),
        clean(account_code),
        clean(vendor_name),
        "",
    )
    return {
        "columns": columns,
        "rows": shown_rows,
        "total_count": len(rows),
        "shown_count": len(shown_rows),
        "notice": notice,
    }


def get_wehago_filtered_rows(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    voucher_no: str = "",
    account_code: str = "",
    vendor_name: str = "",
) -> dict[str, Any]:
    init_wehago_compare_db(engine)
    with engine.begin() as conn:
        return _fetch_filtered_table_rows(
            conn,
            "wehago_ledger_rows",
            WEHAGO_COLUMNS,
            start_year,
            end_year,
            voucher_no=voucher_no,
            account_code=account_code,
            vendor_name=vendor_name,
        )


def get_erp_filtered_rows(
    engine: Any,
    start_year: int | None,
    end_year: int | None,
    voucher_no: str = "",
    account_code: str = "",
    vendor_name: str = "",
) -> dict[str, Any]:
    init_wehago_compare_db(engine)
    with engine.begin() as conn:
        return _fetch_filtered_table_rows(
            conn,
            "wehago_voucher_rows",
            ERP_COLUMNS,
            start_year,
            end_year,
            voucher_no=voucher_no,
            account_code=account_code,
            vendor_name=vendor_name,
        )


def _warm_status_cache_worker(engine: Any, start_year: int, end_year: int, warm_key: str) -> None:
    try:
        _get_cached_status_rows_by_range(engine, start_year, end_year)
    except Exception:
        pass
    finally:
        with _STATUS_CACHE_WARMING_LOCK:
            _STATUS_CACHE_WARMING.discard(warm_key)


def warm_status_cache_async(engine: Any, start_year: int | None, end_year: int | None) -> None:
    if start_year is None or end_year is None:
        return
    if start_year > end_year:
        start_year, end_year = end_year, start_year
    warm_key = f"{start_year}:{end_year}:{_build_bundle_signature(start_year, end_year)}"
    with _STATUS_CACHE_WARMING_LOCK:
        if warm_key in _STATUS_CACHE_WARMING:
            return
        _STATUS_CACHE_WARMING.add(warm_key)
    worker = threading.Thread(
        target=_warm_status_cache_worker,
        args=(engine, start_year, end_year, warm_key),
        daemon=True,
    )
    worker.start()


def get_wehago_compare_dashboard(
    engine: Any,
    start_year: int | None = None,
    end_year: int | None = None,
) -> dict[str, Any]:
    init_wehago_compare_db(engine)
    years = discover_available_years()
    default_year = get_default_year(years)
    if start_year is None and end_year is None:
        start_year = default_year
        end_year = default_year
    elif start_year is None:
        start_year = end_year
    elif end_year is None:
        end_year = start_year
    valid_years = set(years)
    if start_year not in valid_years:
        start_year = default_year
    if end_year not in valid_years:
        end_year = start_year or default_year
    if start_year and end_year and start_year > end_year:
        start_year, end_year = end_year, start_year

    rows_by_status = _get_cached_status_rows_by_range(engine, start_year, end_year)
    metric_sections = [
        {
            "key": status_key,
            "label": label,
            "description": description,
            "count": len(rows_by_status.get(status_key, [])),
            "columns": DETAIL_COLUMN_MAP[status_key],
            "rows": [],
        }
        for status_key, label, description in STATUS_META
    ]

    with engine.begin() as conn:
        account_options = build_account_options_from_db(conn, start_year, end_year)
        latest_upload = fetch_latest_upload_meta(conn)
        last_action = get_last_action_summary(conn=conn)
    warm_status_cache_async(engine, start_year, end_year)

    return {
        "page_title": "전표비교",
        "source_root": str(WEHAGO_SOURCE_ROOT),
        "db_file": "data.db",
        "selected_start_year": start_year,
        "selected_end_year": end_year,
        "available_years": years,
        "metric_sections": metric_sections,
        "wehago_account_options": account_options["wehago"],
        "erp_account_options": account_options["erp"],
        "latest_upload": latest_upload,
        "last_action": last_action,
    }