HM_project_Viewer_Board/scripts/promote_wehago_recheck_projection.py

from __future__ import annotations

import sqlite3
import sys
import re
from datetime import datetime
from pathlib import Path
from typing import Any

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

from runtime_config import DB_PATH
from wehago_compare import (
    QUERY_PROJECTION_VERSION,
    _account_base_names_compatible,
    _contained_core_desc_match,
    _erp_section_identity,
    _is_obvious_recheck_group,
    _recheck_group_rank_key,
    _section_vat_exception_capacity,
    _short_core_desc_fuzzy_match,
    _same_or_similar_desc,
    _wehago_section_identity,
    clean,
)


TARGET_START_YEAR = 2025
TARGET_END_YEAR = 2025


GROUP_COLUMNS = (
    "start_year",
    "end_year",
    "status_key",
    "signature",
    "group_index",
    "fiscal_year",
    "ledger_date",
    "proof_date",
    "voucher_no",
    "draft_no",
    "ledger_row_count",
    "voucher_row_count",
    "ledger_debit",
    "ledger_credit",
    "voucher_debit",
    "voucher_credit",
    "ledger_accounts",
    "voucher_accounts",
    "ledger_vendors",
    "voucher_vendors",
    "review_reason",
    "search_text",
)

ROW_COLUMNS = (
    "start_year",
    "end_year",
    "status_key",
    "signature",
    "group_index",
    "row_index",
    "fiscal_year",
    "status_label",
    "ledger_date",
    "proof_date",
    "voucher_no",
    "draft_no",
    "ledger_account_name",
    "voucher_account_name",
    "ledger_vendor",
    "voucher_vendor",
    "ledger_debit",
    "ledger_credit",
    "voucher_debit",
    "voucher_credit",
    "ledger_desc",
    "voucher_desc",
    "review_reason",
    "matched_case",
    "ledger_row_key",
    "voucher_row_key",
    "match_identity_key",
)


def _dict(row: sqlite3.Row) -> dict[str, Any]:
    return {key: row[key] for key in row.keys()}


def _amount_key(value: Any) -> str:
    try:
        amount = float(str(value or "0").replace(",", ""))
    except Exception:
        amount = 0.0
    if abs(amount - round(amount)) < 0.0001:
        return str(int(round(amount)))
    return f"{amount:.2f}".rstrip("0").rstrip(".")


def _parse_amount(value: Any) -> float:
    try:
        return float(str(value or "0").replace(",", ""))
    except Exception:
        return 0.0


def _has_ledger_value(row: dict[str, Any]) -> bool:
    return bool(clean(row.get("ledger_account_name"))) and (
        abs(_parse_amount(row.get("ledger_debit"))) > 0.0001
        or abs(_parse_amount(row.get("ledger_credit"))) > 0.0001
    )


def _has_voucher_value(row: dict[str, Any]) -> bool:
    return bool(clean(row.get("voucher_account_name"))) and (
        abs(_parse_amount(row.get("voucher_debit"))) > 0.0001
        or abs(_parse_amount(row.get("voucher_credit"))) > 0.0001
    )


def _same_side_amount_match(ledger_row: dict[str, Any], voucher_row: dict[str, Any]) -> bool:
    return (
        abs(_parse_amount(ledger_row.get("ledger_debit")) - _parse_amount(voucher_row.get("voucher_debit"))) < 0.5
        and abs(_parse_amount(ledger_row.get("ledger_credit")) - _parse_amount(voucher_row.get("voucher_credit"))) < 0.5
        and (
            abs(_parse_amount(ledger_row.get("ledger_debit"))) > 0.0001
            or abs(_parse_amount(ledger_row.get("ledger_credit"))) > 0.0001
        )
    )


def _desc_core_match(ledger_row: dict[str, Any], voucher_row: dict[str, Any]) -> bool:
    probe = {
        "ledger_desc": ledger_row.get("ledger_desc"),
        "voucher_desc": voucher_row.get("voucher_desc"),
    }
    matched = (
        _same_or_similar_desc(probe)
        or _contained_core_desc_match(ledger_row.get("ledger_desc"), voucher_row.get("voucher_desc"))
        or _short_core_desc_fuzzy_match(ledger_row.get("ledger_desc"), voucher_row.get("voucher_desc"))
    )
    if matched:
        return True

    stop_words = {
        "관련",
        "전표",
        "처리",
        "정산",
        "금액",
        "비용",
        "지급",
        "입금",
        "출금",
        "매입",
        "매출",
        "급여",
        "제경비",
    }

    def tokens(value: Any) -> set[str]:
        found: set[str] = set()
        for token in re.split(r"[^0-9A-Za-z가-힣]+", clean(value)):
            token = token.strip()
            if len(token) < 2 or token in stop_words or re.fullmatch(r"\d+월?", token):
                continue
            found.add(token)
        return found

    shared = tokens(ledger_row.get("ledger_desc")) & tokens(voucher_row.get("voucher_desc"))
    return len(shared) >= 2 or any(len(token) >= 3 for token in shared)


def _merge_internal_recheck_pairs(group: dict[str, Any]) -> dict[str, Any]:
    rows = [dict(row) for row in group.get("rows") or []]
    ledger_only = [row for row in rows if _has_ledger_value(row) and not _has_voucher_value(row)]
    voucher_only = [row for row in rows if _has_voucher_value(row) and not _has_ledger_value(row)]
    if not ledger_only or not voucher_only:
        return group

    used_ledger: set[int] = set()
    used_voucher: set[int] = set()
    merged_rows: list[dict[str, Any]] = []
    candidates: list[tuple[float, int, int]] = []
    for ledger_index, ledger_row in enumerate(ledger_only):
        for voucher_index, voucher_row in enumerate(voucher_only):
            if not _same_side_amount_match(ledger_row, voucher_row):
                continue
            if not _account_base_names_compatible(ledger_row.get("ledger_account_name"), voucher_row.get("voucher_account_name")):
                continue
            if not _desc_core_match(ledger_row, voucher_row):
                continue
            amount = max(abs(_parse_amount(ledger_row.get("ledger_debit"))), abs(_parse_amount(ledger_row.get("ledger_credit"))))
            candidates.append((amount, ledger_index, voucher_index))
    for _amount, ledger_index, voucher_index in sorted(candidates, reverse=True):
        if ledger_index in used_ledger or voucher_index in used_voucher:
            continue
        ledger_row = ledger_only[ledger_index]
        voucher_row = voucher_only[voucher_index]
        merged = dict(ledger_row)
        for field in (
            "proof_date",
            "draft_no",
            "voucher_account_name",
            "voucher_vendor",
            "voucher_debit",
            "voucher_credit",
            "voucher_desc",
            "voucher_row_key",
            "match_identity_key",
        ):
            merged[field] = voucher_row.get(field, "")
        merged["review_reason"] = "RECHECK_INTERNAL_CORE_MATCH"
        merged_rows.append(merged)
        used_ledger.add(ledger_index)
        used_voucher.add(voucher_index)
    if not merged_rows:
        return group

    remaining_rows: list[dict[str, Any]] = []
    ledger_ids = {id(row): index for index, row in enumerate(ledger_only)}
    voucher_ids = {id(row): index for index, row in enumerate(voucher_only)}
    for row in rows:
        if _has_ledger_value(row) and not _has_voucher_value(row):
            index = ledger_ids.get(id(row))
            if index is not None and index in used_ledger:
                continue
        if _has_voucher_value(row) and not _has_ledger_value(row):
            index = voucher_ids.get(id(row))
            if index is not None and index in used_voucher:
                continue
        remaining_rows.append(row)
    return {**group, "rows": merged_rows + remaining_rows}


def _group_manual_review_key(group: dict[str, Any]) -> str:
    summary = group.get("summary") or {}
    return "|".join(
        [
            "manual-recheck",
            clean(summary.get("fiscal_year")),
            clean(summary.get("ledger_date")),
            clean(summary.get("voucher_no")),
            clean(summary.get("draft_no")),
            "",
            "",
            _amount_key(summary.get("ledger_debit")),
            _amount_key(summary.get("ledger_credit")),
            _amount_key(summary.get("voucher_debit")),
            _amount_key(summary.get("voucher_credit")),
        ]
    )


def _row_change_key(row: dict[str, Any], change_type: str) -> str:
    explicit_key = clean(row.get("review_key")) or clean(row.get("match_identity_key"))
    if explicit_key:
        return f"{change_type}:{explicit_key}"
    return "|".join(
        [
            change_type,
            clean(row.get("fiscal_year")),
            clean(row.get("ledger_date")),
            clean(row.get("voucher_no")),
            clean(row.get("draft_no")),
            clean(row.get("ledger_account_name")),
            clean(row.get("voucher_account_name")),
            _amount_key(row.get("ledger_debit")),
            _amount_key(row.get("ledger_credit")),
            _amount_key(row.get("voucher_debit")),
            _amount_key(row.get("voucher_credit")),
            clean(row.get("ledger_desc")),
            clean(row.get("voucher_desc")),
        ]
    )


def _ensure_change_table(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS wehago_recheck_row_changes (
            change_key TEXT PRIMARY KEY,
            change_type TEXT NOT NULL DEFAULT 'match',
            fiscal_year INTEGER,
            voucher_no TEXT NOT NULL DEFAULT '',
            draft_no TEXT NOT NULL DEFAULT '',
            ledger_date TEXT NOT NULL DEFAULT '',
            proof_date TEXT NOT NULL DEFAULT '',
            ledger_account_name TEXT NOT NULL DEFAULT '',
            voucher_account_name TEXT NOT NULL DEFAULT '',
            ledger_debit REAL NOT NULL DEFAULT 0,
            ledger_credit REAL NOT NULL DEFAULT 0,
            voucher_debit REAL NOT NULL DEFAULT 0,
            voucher_credit REAL NOT NULL DEFAULT 0,
            ledger_desc TEXT NOT NULL DEFAULT '',
            voucher_desc TEXT NOT NULL DEFAULT '',
            changed_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
        """
    )


def _load_manual_change_keys(conn: sqlite3.Connection) -> tuple[set[str], set[str], set[str]]:
    _ensure_change_table(conn)
    review_keys = {
        str(row[0])
        for row in conn.execute(
            """
            SELECT review_key
            FROM wehago_recheck_reviews
            WHERE fiscal_year BETWEEN ? AND ?
            """,
            (TARGET_START_YEAR, TARGET_END_YEAR),
        ).fetchall()
    }
    match_change_keys = {
        str(row[0])
        for row in conn.execute(
            """
            SELECT change_key
            FROM wehago_recheck_row_changes
            WHERE fiscal_year BETWEEN ? AND ?
              AND change_type = 'match'
            """,
            (TARGET_START_YEAR, TARGET_END_YEAR),
        ).fetchall()
    }
    split_change_keys = {
        str(row[0])
        for row in conn.execute(
            """
            SELECT change_key
            FROM wehago_recheck_row_changes
            WHERE fiscal_year BETWEEN ? AND ?
              AND change_type = 'split'
            """,
            (TARGET_START_YEAR, TARGET_END_YEAR),
        ).fetchall()
    }
    return review_keys, match_change_keys, split_change_keys


def _snapshot_signature(conn: sqlite3.Connection) -> str:
    row = conn.execute(
        """
        SELECT snapshot_signature
        FROM wehago_snapshot_status
        WHERE fiscal_year = ?
          AND state = 'ready'
        LIMIT 1
        """,
        (TARGET_START_YEAR,),
    ).fetchone()
    if row is not None and str(row["snapshot_signature"] or ""):
        signature = str(row["snapshot_signature"])
        exists = conn.execute(
            """
            SELECT 1
            FROM wehago_compare_export_row_cache
            WHERE fiscal_year BETWEEN ? AND ?
              AND snapshot_signature = ?
            LIMIT 1
            """,
            (TARGET_START_YEAR, TARGET_END_YEAR, signature),
        ).fetchone()
        if exists:
            return signature
    row = conn.execute(
        """
        SELECT snapshot_signature, COUNT(*) AS row_count, MAX(rowid) AS max_rowid
        FROM wehago_compare_export_row_cache
        WHERE fiscal_year BETWEEN ? AND ?
        GROUP BY snapshot_signature
        ORDER BY
            CASE WHEN snapshot_signature LIKE 'voucher-summary-v7|recheck-v20%' THEN 0 ELSE 1 END ASC,
            row_count DESC,
            max_rowid DESC
        LIMIT 1
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR),
    ).fetchone()
    if row is None or not str(row["snapshot_signature"] or ""):
        raise RuntimeError("No snapshot/export signature was found.")
    return str(row["snapshot_signature"])


def _latest_query_projection_signature(conn: sqlite3.Connection) -> str | None:
    row = conn.execute(
        """
        SELECT signature, COUNT(DISTINCT status_key) AS status_count, MAX(updated_at) AS updated_at
        FROM wehago_compare_query_groups
        WHERE start_year = ?
          AND end_year = ?
          AND signature LIKE ?
          AND signature NOT LIKE '%|snapshot-recheck-promote|%'
        GROUP BY signature
        HAVING status_count >= 5
        ORDER BY updated_at DESC
        LIMIT 1
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR, f"{QUERY_PROJECTION_VERSION}|%"),
    ).fetchone()
    return str(row["signature"] or "") if row is not None else None


def _load_groups_from_query_projection(conn: sqlite3.Connection, signature: str) -> dict[str, list[dict[str, Any]]]:
    group_rows = conn.execute(
        """
        SELECT *
        FROM wehago_compare_query_groups
        WHERE start_year = ?
          AND end_year = ?
          AND signature = ?
        ORDER BY status_key, group_index
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR, signature),
    ).fetchall()
    detail_rows = conn.execute(
        """
        SELECT *
        FROM wehago_compare_query_rows
        WHERE start_year = ?
          AND end_year = ?
          AND signature = ?
        ORDER BY status_key, group_index, row_index
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR, signature),
    ).fetchall()
    rows_by_key: dict[tuple[str, int], list[dict[str, Any]]] = {}
    for row in detail_rows:
        payload = _dict(row)
        rows_by_key.setdefault((str(payload["status_key"]), int(payload["group_index"])), []).append(payload)
    groups: dict[str, list[dict[str, Any]]] = {}
    for row in group_rows:
        summary = _dict(row)
        status_key = str(summary["status_key"])
        group_index = int(summary["group_index"])
        groups.setdefault(status_key, []).append(
            {
                "summary": summary,
                "rows": rows_by_key.get((status_key, group_index), []),
                "source_group_index": group_index,
            }
        )
    return groups


def _source_signature(conn: sqlite3.Connection) -> str:
    row = conn.execute(
        """
        SELECT signature, COUNT(DISTINCT status_key) AS status_count, COUNT(*) AS group_count
        FROM wehago_compare_query_groups
        WHERE fiscal_year BETWEEN ? AND ?
          AND signature NOT LIKE ?
        GROUP BY signature
        HAVING status_count >= 5
        ORDER BY group_count DESC
        LIMIT 1
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR, f"{QUERY_PROJECTION_VERSION}|%"),
    ).fetchone()
    if row is None:
        raise RuntimeError("No source query projection with voucher status groups was found.")
    return str(row["signature"])


def _matched_source_signature(conn: sqlite3.Connection) -> str:
    row = conn.execute(
        """
        SELECT signature, COUNT(*) AS group_count
        FROM wehago_compare_query_groups
        WHERE fiscal_year BETWEEN ? AND ?
          AND status_key = 'voucher_matched'
          AND signature NOT LIKE ?
        GROUP BY signature
        ORDER BY group_count DESC
        LIMIT 1
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR, f"{QUERY_PROJECTION_VERSION}|%"),
    ).fetchone()
    if row is None:
        raise RuntimeError("No matched source query projection was found.")
    return str(row["signature"])


def _load_groups(
    conn: sqlite3.Connection,
    signature: str,
    statuses: tuple[str, ...] | None = None,
) -> dict[str, list[dict[str, Any]]]:
    status_filter = ""
    params: list[Any] = [signature, TARGET_START_YEAR, TARGET_END_YEAR]
    if statuses:
        status_filter = f" AND status_key IN ({', '.join('?' for _ in statuses)})"
        params.extend(statuses)
    groups: dict[str, list[dict[str, Any]]] = {}
    group_rows = conn.execute(
        f"""
        SELECT *
        FROM wehago_compare_query_groups
        WHERE signature = ?
          AND fiscal_year BETWEEN ? AND ?
          {status_filter}
        ORDER BY status_key, group_index
        """,
        params,
    ).fetchall()
    detail_rows = conn.execute(
        f"""
        SELECT *
        FROM wehago_compare_query_rows
        WHERE signature = ?
          AND fiscal_year BETWEEN ? AND ?
          {status_filter}
        ORDER BY status_key, group_index, row_index
        """,
        params,
    ).fetchall()
    rows_by_key: dict[tuple[str, int], list[dict[str, Any]]] = {}
    for row in detail_rows:
        item = _dict(row)
        rows_by_key.setdefault((str(item["status_key"]), int(item["group_index"])), []).append(item)
    for row in group_rows:
        summary = _dict(row)
        status_key = str(summary["status_key"])
        group_index = int(summary["group_index"])
        groups.setdefault(status_key, []).append(
            {
                "summary": summary,
                "rows": rows_by_key.get((status_key, group_index), []),
                "source_group_index": group_index,
            }
        )
    return groups


def _load_groups_from_export_cache(conn: sqlite3.Connection, snapshot_signature: str) -> dict[str, list[dict[str, Any]]]:
    export_rows = conn.execute(
        """
        SELECT *
        FROM wehago_compare_export_row_cache
        WHERE fiscal_year BETWEEN ? AND ?
          AND snapshot_signature = ?
        ORDER BY status_key, group_sort, row_sort
        """,
        (TARGET_START_YEAR, TARGET_END_YEAR, snapshot_signature),
    ).fetchall()
    grouped_rows: dict[tuple[str, int], list[dict[str, Any]]] = {}
    for row in export_rows:
        item = _dict(row)
        grouped_rows.setdefault((str(item["status_key"]), int(item["group_sort"])), []).append(item)

    groups: dict[str, list[dict[str, Any]]] = {}
    for (status_key, group_index), rows in grouped_rows.items():
        first = rows[0]
        detail_rows: list[dict[str, Any]] = []
        ledger_accounts: list[str] = []
        voucher_accounts: list[str] = []
        ledger_vendors: list[str] = []
        voucher_vendors: list[str] = []

        def append_unique(target: list[str], value: Any) -> None:
            text_value = clean(value)
            if text_value and text_value not in target:
                target.append(text_value)

        ledger_row_count = 0
        voucher_row_count = 0
        for row_index, row in enumerate(rows):
            detail = {
                "fiscal_year": int(row.get("fiscal_year") or 0),
                "status_label": "Matched" if status_key in {"voucher_matched", "erp_voucher_matched"} else "Recheck" if status_key == "voucher_recheck" else "Unmatched",
                "ledger_date": clean(row.get("ledger_date")),
                "proof_date": "",
                "voucher_no": clean(row.get("voucher_no")),
                "draft_no": clean(row.get("draft_no")),
                "ledger_account_name": clean(row.get("ledger_account_name")),
                "voucher_account_name": clean(row.get("voucher_account_name")),
                "ledger_vendor": clean(row.get("ledger_vendor")),
                "voucher_vendor": clean(row.get("voucher_vendor")),
                "ledger_debit": float(row.get("ledger_debit") or 0),
                "ledger_credit": float(row.get("ledger_credit") or 0),
                "voucher_debit": float(row.get("voucher_debit") or 0),
                "voucher_credit": float(row.get("voucher_credit") or 0),
                "ledger_desc": clean(row.get("ledger_desc")),
                "voucher_desc": clean(row.get("voucher_desc")),
                "review_reason": "SNAPSHOT_EXPORT_CACHE",
                "matched_case": "",
                "ledger_row_key": "",
                "voucher_row_key": "",
                "match_identity_key": "",
                "row_index": row_index,
            }
            if detail["ledger_account_name"] or detail["ledger_desc"]:
                ledger_row_count += 1
            if detail["voucher_account_name"] or detail["voucher_desc"]:
                voucher_row_count += 1
            append_unique(ledger_accounts, detail["ledger_account_name"])
            append_unique(voucher_accounts, detail["voucher_account_name"])
            append_unique(ledger_vendors, detail["ledger_vendor"])
            append_unique(voucher_vendors, detail["voucher_vendor"])
            detail_rows.append(detail)

        summary = {
            "fiscal_year": int(first.get("fiscal_year") or 0),
            "status_label": detail_rows[0]["status_label"] if detail_rows else "",
            "ledger_date": clean(first.get("ledger_date")),
            "proof_date": "",
            "voucher_no": clean(first.get("group_voucher_no")) or clean(first.get("voucher_no")),
            "draft_no": clean(first.get("group_draft_no")) or clean(first.get("draft_no")),
            "ledger_row_count": ledger_row_count,
            "voucher_row_count": voucher_row_count,
            "ledger_debit": float(first.get("group_ledger_debit") or 0),
            "ledger_credit": float(first.get("group_ledger_credit") or 0),
            "voucher_debit": float(first.get("group_voucher_debit") or 0),
            "voucher_credit": float(first.get("group_voucher_credit") or 0),
            "ledger_accounts": clean(first.get("group_ledger_accounts")) or ", ".join(ledger_accounts),
            "voucher_accounts": clean(first.get("group_voucher_accounts")) or ", ".join(voucher_accounts),
            "ledger_vendors": clean(first.get("group_ledger_vendors")) or ", ".join(ledger_vendors),
            "voucher_vendors": clean(first.get("group_voucher_vendors")) or ", ".join(voucher_vendors),
            "review_reason": "SNAPSHOT_EXPORT_CACHE",
            "search_text": "",
        }
        summary["search_text"] = _search_text(summary, detail_rows)
        groups.setdefault(status_key, []).append(
            {
                "summary": summary,
                "rows": detail_rows,
                "source_group_index": group_index,
            }
        )
    return groups


def _seed_used_identities(groups: dict[str, list[dict[str, Any]]]) -> tuple[dict[str, int], dict[str, int]]:
    used_wehago: dict[str, int] = {}
    used_erp: dict[str, int] = {}
    for group in groups.get("voucher_matched", []):
        identity = _wehago_section_identity(group)
        if identity:
            used_wehago[identity] = used_wehago.get(identity, 0) + 1
    for group in groups.get("erp_voucher_matched", []):
        identity = _erp_section_identity(group)
        if identity:
            used_erp[identity] = used_erp.get(identity, 0) + 1
    return used_wehago, used_erp


def _group_has_manual_match(
    group: dict[str, Any],
    review_keys: set[str],
    match_change_keys: set[str],
) -> bool:
    if _group_manual_review_key(group) in review_keys:
        return True
    for row in group.get("rows") or []:
        if _row_change_key(row, "match") in match_change_keys:
            return True
    return False


def _select_promotions(
    groups: dict[str, list[dict[str, Any]]],
    review_keys: set[str],
    match_change_keys: set[str],
) -> set[int]:
    used_wehago, _used_erp = _seed_used_identities(groups)
    candidates: list[dict[str, Any]] = []
    for group in groups.get("voucher_recheck", []):
        manual_match = _group_has_manual_match(group, review_keys, match_change_keys)
        if not manual_match and not _is_obvious_recheck_group(group):
            continue
        wehago_identity = _wehago_section_identity(group)
        erp_identity = _erp_section_identity(group)
        if not wehago_identity or not erp_identity:
            continue
        candidates.append(
            {
                "group": group,
                "wehago_identity": wehago_identity,
                "erp_identity": erp_identity,
                "wehago_capacity": _section_vat_exception_capacity(group, side="wehago"),
                "rank": (1 if manual_match else 0, *_recheck_group_rank_key(group)),
                "manual_match": manual_match,
            }
        )
    candidates.sort(key=lambda item: item["rank"], reverse=True)
    promoted: set[int] = set()
    for candidate in candidates:
        wehago_identity = candidate["wehago_identity"]
        erp_identity = candidate["erp_identity"]
        wehago_capacity = int(candidate["wehago_capacity"] or 1)
        if used_wehago.get(wehago_identity, 0) >= wehago_capacity:
            continue
        promoted.add(id(candidate["group"]))
        used_wehago[wehago_identity] = used_wehago.get(wehago_identity, 0) + 1
    return promoted


def _search_text(summary: dict[str, Any], rows: list[dict[str, Any]]) -> str:
    parts = [
        summary.get("voucher_no"),
        summary.get("draft_no"),
        summary.get("ledger_accounts"),
        summary.get("voucher_accounts"),
        summary.get("ledger_vendors"),
        summary.get("voucher_vendors"),
        summary.get("review_reason"),
    ]
    for row in rows[:20]:
        parts.extend([row.get("ledger_desc"), row.get("voucher_desc")])
    return " ".join(clean(part) for part in parts if clean(part))


def _insert_group(
    conn: sqlite3.Connection,
    *,
    signature: str,
    status_key: str,
    group_index: int,
    group: dict[str, Any],
    promoted: bool = False,
    split_change_keys: set[str] | None = None,
    omit_split_rows: bool = False,
) -> None:
    summary = dict(group["summary"])
    rows = [dict(row) for row in group.get("rows") or []]
    split_change_keys = split_change_keys or set()
    if promoted and omit_split_rows:
        rows = [row for row in rows if _row_change_key(row, "split") not in split_change_keys]
    summary.update(
        {
            "start_year": TARGET_START_YEAR,
            "end_year": TARGET_END_YEAR,
            "status_key": status_key,
            "signature": signature,
            "group_index": group_index,
        }
    )
    if promoted:
        summary["review_reason"] = clean(summary.get("review_reason")) or "RECHECK_PROMOTED_BY_USER_RULE"
        summary["search_text"] = _search_text(summary, rows)
    values = [summary.get(column, "") for column in GROUP_COLUMNS]
    conn.execute(
        f"""
        INSERT INTO wehago_compare_query_groups ({', '.join(GROUP_COLUMNS)}, created_at, updated_at)
        VALUES ({', '.join('?' for _ in GROUP_COLUMNS)}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
        """,
        values,
    )
    for row_index, row in enumerate(rows):
        split_row = promoted and _row_change_key(row, "split") in split_change_keys
        row.update(
            {
                "start_year": TARGET_START_YEAR,
                "end_year": TARGET_END_YEAR,
                "status_key": status_key,
                "signature": signature,
                "group_index": group_index,
                "row_index": row_index,
            }
        )
        if promoted:
            if split_row:
                row["status_label"] = "Unmatched"
                row["review_reason"] = "MANUAL_SPLIT_FROM_RECHECK"
                row["proof_date"] = ""
                row["draft_no"] = ""
                row["voucher_account_name"] = ""
                row["voucher_vendor"] = ""
                row["voucher_debit"] = 0
                row["voucher_credit"] = 0
                row["voucher_desc"] = ""
                row["voucher_row_key"] = ""
                row["match_identity_key"] = ""
            else:
                row["status_label"] = "Matched"
                row["review_reason"] = clean(row.get("review_reason")) or "RECHECK_PROMOTED_BY_USER_RULE"
        values = [row.get(column, "") for column in ROW_COLUMNS]
        conn.execute(
            f"""
            INSERT INTO wehago_compare_query_rows ({', '.join(ROW_COLUMNS)}, created_at, updated_at)
            VALUES ({', '.join('?' for _ in ROW_COLUMNS)}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """,
            values,
        )


def main() -> None:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    source_signature = _latest_query_projection_signature(conn)
    source_kind = "query_projection"
    if not source_signature:
        source_signature = _snapshot_signature(conn)
        source_kind = "export_cache"
    target_signature = f"{QUERY_PROJECTION_VERSION}|snapshot-recheck-promote|{source_signature}"
    groups = (
        _load_groups_from_query_projection(conn, source_signature)
        if source_kind == "query_projection"
        else _load_groups_from_export_cache(conn, source_signature)
    )
    groups["voucher_recheck"] = [
        _merge_internal_recheck_pairs(group)
        for group in groups.get("voucher_recheck", [])
    ]
    review_keys, match_change_keys, split_change_keys = _load_manual_change_keys(conn)
    promoted_ids = _select_promotions(groups, review_keys, match_change_keys)
    conn.execute("BEGIN")
    try:
        conn.execute(
            "DELETE FROM wehago_compare_query_rows WHERE start_year = ? AND end_year = ? AND signature = ?",
            (TARGET_START_YEAR, TARGET_END_YEAR, target_signature),
        )
        conn.execute(
            "DELETE FROM wehago_compare_query_groups WHERE start_year = ? AND end_year = ? AND signature = ?",
            (TARGET_START_YEAR, TARGET_END_YEAR, target_signature),
        )
        max_group_index = {
            status_key: max([int(group["source_group_index"]) for group in status_groups] or [0])
            for status_key, status_groups in groups.items()
        }
        for status_key, status_groups in groups.items():
            for group in status_groups:
                if status_key == "voucher_recheck" and id(group) in promoted_ids:
                    continue
                _insert_group(
                    conn,
                    signature=target_signature,
                    status_key=status_key,
                    group_index=int(group["source_group_index"]),
                    group=group,
                )
        for group in groups.get("voucher_recheck", []):
            if id(group) not in promoted_ids:
                continue
            max_group_index["voucher_matched"] = max_group_index.get("voucher_matched", 0) + 1
            _insert_group(
                conn,
                signature=target_signature,
                status_key="voucher_matched",
                group_index=max_group_index["voucher_matched"],
                group=group,
                promoted=True,
                split_change_keys=split_change_keys,
            )
            max_group_index["erp_voucher_matched"] = max_group_index.get("erp_voucher_matched", 0) + 1
            _insert_group(
                conn,
                signature=target_signature,
                status_key="erp_voucher_matched",
                group_index=max_group_index["erp_voucher_matched"],
                group=group,
                promoted=True,
                split_change_keys=split_change_keys,
                omit_split_rows=True,
            )
        conn.execute(
            """
            INSERT INTO wehago_action_history (action_type, payload_json, created_at)
            VALUES ('auto_recheck_promote', ?, CURRENT_TIMESTAMP)
            """,
            (
                f'{{"count": {len(promoted_ids)}, "start_year": {TARGET_START_YEAR}, '
                f'"manual_review_keys": {len(review_keys)}, "match_change_keys": {len(match_change_keys)}, '
                f'"split_change_keys": {len(split_change_keys)}, '
                f'"source_kind": "{source_kind}", '
                f'"end_year": {TARGET_END_YEAR}, "signature": "{target_signature}", '
                f'"created_at": "{datetime.now().isoformat(timespec="seconds")}"}}',
            ),
        )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    counts = conn.execute(
        """
        SELECT status_key, COUNT(*)
        FROM wehago_compare_query_groups
        WHERE signature = ?
          AND fiscal_year BETWEEN ? AND ?
        GROUP BY status_key
        ORDER BY status_key
        """,
        (target_signature, TARGET_START_YEAR, TARGET_END_YEAR),
    ).fetchall()
    print(
        {
            "source_signature": source_signature,
            "source_kind": source_kind,
            "target_signature": target_signature,
            "promoted": len(promoted_ids),
            "counts": {str(row[0]): int(row[1]) for row in counts},
        }
    )


if __name__ == "__main__":
    main()