scripts/48_prepare_version_data.py

"""Prepare reviewed/inherited menu-search data from versioned SQLite files.

This step does not call an LLM. Existing reviewed 1.5.5 descriptions are
inherited only when SCREEN_NUM and menu name remain stable. New or renamed
menus receive a conservative fallback document and are added to review_queue.
"""

from __future__ import annotations

import argparse
import copy
import csv
import json
import os
import re
import sqlite3
import sys
from collections import Counter
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from config import DATA_DIR, DEFAULT_APP_VERSION, GENERATED_DIR, RAW_DIR
from core.auto_candidates import build_embedding_text
from core.query_data import normalize_query_text, sanitize_query_rows
from core.version_paths import VERSIONS_DIR, version_dir


PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_DB_DIR = Path(
    os.getenv("MENU_DB_DIR", str(PROJECT_ROOT / "menu_db_ver"))
).resolve()
ROOT_SCREEN_NUMS = {"10", "295", "310"}
POPUP_SCREEN_NUMS = {"10"}  # 계좌개설은 앱에서 팝업 화면으로 호출
REVIEW_OVERRIDES_PATH = VERSIONS_DIR / "review_overrides.json"
AUTO_CANDIDATES_PATH = VERSIONS_DIR / "auto_candidates.json"
CURATED_QUERY_ALIASES_PATH = DATA_DIR / "curated_query_aliases.json"


def _version_key(value: str) -> tuple[int, int, int]:
    parts = tuple(int(part) for part in value.split("."))
    if len(parts) != 3:
        raise ValueError(f"invalid app version: {value}")
    return parts


def discover_production_versions(db_dir: Path) -> list[str]:
    minimum = _version_key(DEFAULT_APP_VERSION)
    versions = []
    for path in db_dir.glob("*_menu.db"):
        value = path.name.removesuffix("_menu.db")
        if not re.fullmatch(r"\d+\.\d+\.\d+", value):
            continue
        key = _version_key(value)
        if key < minimum or key[0] in {7, 9}:
            continue
        versions.append(value)
    return sorted(set(versions), key=_version_key)


def _load_json(path: Path) -> list:
    return json.loads(path.read_text(encoding="utf-8"))


def _load_jsonl(path: Path) -> list[dict]:
    rows = []
    with path.open("r", encoding="utf-8") as f:
        for line in f:
            if line.strip():
                rows.append(json.loads(line))
    return rows


def _write_json(path: Path, value) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def _write_jsonl(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    content = "".join(
        json.dumps(row, ensure_ascii=False) + "\n" for row in rows
    )
    path.write_text(content, encoding="utf-8")


def _write_review_csv(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = [
        "first_seen_app_version",
        "affected_app_versions",
        "menu_id",
        "screen_num",
        "map_id",
        "menu_name",
        "menu_path",
        "keywords",
        "reason",
        "generation_source",
        "candidate_oneline_desc",
        "candidate_function_desc",
        "candidate_embedding_text",
        "candidate_sample_queries",
        "review_status",
        "reviewed_keywords",
        "reviewed_oneline_desc",
        "reviewed_function_desc",
        "reviewed_sample_queries",
        "review_comment",
    ]
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for item in rows:
            writer.writerow(
                {
                    **item,
                    "affected_app_versions": ", ".join(
                        item["affected_app_versions"]
                    ),
                    "keywords": ", ".join(item["keywords"]),
                    "candidate_sample_queries": " | ".join(
                        item.get("candidate_sample_queries", [])
                    ),
                    "review_status": "",
                    "reviewed_keywords": "",
                    "reviewed_oneline_desc": "",
                    "reviewed_function_desc": "",
                    "reviewed_sample_queries": "",
                    "review_comment": "",
                }
            )


def _text(value) -> str:
    return str(value or "").strip()


def _screen_num(value) -> str:
    text = _text(value)
    return text.removesuffix(".0")


def _bool_text(value) -> bool:
    return _text(value).lower() == "true"


def _int(value) -> int:
    try:
        return int(value or 0)
    except (TypeError, ValueError):
        return 0


def _keywords(value) -> list[str]:
    if not value:
        return []
    parts = re.split(r"[,;\n]+", str(value))
    output = []
    seen = set()
    for part in parts:
        keyword = " ".join(part.split())
        key = keyword.casefold()
        if keyword and key not in seen:
            seen.add(key)
            output.append(keyword)
    return output


def _merge_keywords(*groups: list[str]) -> list[str]:
    merged = []
    seen = set()
    for group in groups:
        for keyword in group:
            key = keyword.casefold()
            if keyword and key not in seen:
                seen.add(key)
                merged.append(keyword)
    return merged


def _menu_path(row: sqlite3.Row) -> str:
    # 기획 확정(2026-09-02, Xstudio 김보미): 노출 경로는 2depth부터 표기한다.
    # 1depth(대분류)는 하위 depth가 전혀 없을 때만 표기한다.
    # 빈 depth 컬럼을 건너뛴 동일 이름 중복(원장 기재 관행, 예: 계좌개설)은
    # 접고, 실제 인접 계층의 동일 이름(예: 로보마켓 > 로보마켓)은 유지한다.
    levels = [
        (2, _text(row["DEPTH2_NAME"])),
        (3, _text(row["DEPTH3_NAME"])),
        (4, _text(row["DEPTH4_NAME"])),
    ]
    filled = [(column, name) for column, name in levels if name]
    if not filled:
        return _text(row["DEPTH1_NAME"])
    parts: list[str] = []
    previous_column = None
    for column, name in filled:
        if (
            parts
            and name == parts[-1]
            and previous_column is not None
            and column - previous_column > 1
        ):
            previous_column = column
            continue
        parts.append(name)
        previous_column = column
    return " > ".join(parts)


def _menu_name(row: sqlite3.Row) -> str:
    for column in ("DEPTH4_NAME", "DEPTH3_NAME", "DEPTH2_NAME", "DEPTH1_NAME"):
        value = _text(row[column])
        if value:
            return value
    return f"화면 {_screen_num(row['SCREEN_NUM'])}"


def _eligible(row: sqlite3.Row) -> bool:
    if not _bool_text(row["IS_VISIBLE_MENU"]):
        return False
    screen_num = _screen_num(row["SCREEN_NUM"])
    if not screen_num:
        return False
    if screen_num in ROOT_SCREEN_NUMS:
        return True
    if not _text(row["DEPTH1_NAME"]):
        return False
    return "팝업" not in _text(row["DEPTH4_NAME"])


def load_sqlite_menus(db_path: Path, app_version: str) -> tuple[list[dict], str]:
    connection = sqlite3.connect(f"{db_path.resolve().as_uri()}?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    try:
        rows = connection.execute("SELECT * FROM Newmenu").fetchall()
        version_row = connection.execute("SELECT UPDATE_DATE FROM version").fetchone()
    finally:
        connection.close()

    menus = []
    seen = set()
    for row in rows:
        if not _eligible(row):
            continue
        screen_num = _screen_num(row["SCREEN_NUM"])
        if screen_num in seen:
            raise ValueError(
                f"{app_version}: duplicate eligible SCREEN_NUM {screen_num}"
            )
        seen.add(screen_num)
        name = _menu_name(row)
        path = _menu_path(row)
        menus.append(
            {
                "menu_id": f"SCR_{screen_num}",
                "menu_name": name,
                "menu_path": path,
                "category": _text(row["DEPTH1_NAME"]) or "공통",
                "screen_num": screen_num,
                "map_id": _int(row["MAP_ID"]),
                "search_count": _int(row["SEARCH_COUNT"]),
                "keywords": _keywords(row["MENU_KEYWORD"]),
                "app_version": app_version,
                "login_auth": _int(row["LOGIN_AUTH"]),
                "is_cert_login": _bool_text(row["IS_CERT_LOGIN"]),
                "is_account": _bool_text(row["IS_ACCOUNT"]),
                "account_group": _int(row["ACCOUNT_GROUP"]),
                "account_type": _int(row["ACCOUNT_TYPE"]),
                "check_service_auth": _int(row["CHECK_SERVICE_AUTH"]),
                "is_popup": screen_num in POPUP_SCREEN_NUMS,
            }
        )
    menus.sort(key=lambda item: int(item["screen_num"]))
    update_date = _text(version_row[0]) if version_row else ""
    return menus, update_date


def _fallback_description(menu: dict, reason: str) -> dict:
    keywords = menu["keywords"]
    keyword_text = ", ".join(keywords[:12])
    function_desc = (
        f"{menu['menu_name']} 관련 기능을 이용할 때 찾는 메뉴입니다. "
        f"앱 경로는 {menu['menu_path']}입니다."
    )
    embedding_text = function_desc
    if keyword_text:
        embedding_text += f" 주요 검색어는 {keyword_text}입니다."
    return {
        "menu_path": menu["menu_path"],
        "menu_name": menu["menu_name"],
        "function_desc": function_desc,
        "action_types": [],
        "user_verbs": [],
        "keywords": keywords,
        "related_concepts": [],
        "sample_queries": [],
        "embedding_text": embedding_text,
        "menu_id": menu["menu_id"],
        "oneline_desc": f"{menu['menu_name']} 관련 기능",
        "description_source": "sqlite_fallback",
        "review_status": reason,
        "review_required": True,
    }


def _auto_candidate_description(menu: dict, candidate: dict, reason: str) -> dict:
    keywords = _merge_keywords(
        menu.get("keywords", []),
        candidate.get("keywords", []),
        candidate.get("aliases", []),
    )
    sample_queries = [
        str(query).strip()
        for query in candidate.get("sample_queries", [])
        if str(query).strip()
    ]
    composed = {
        **candidate,
        "keywords": keywords,
        "aliases": candidate.get("aliases", []),
    }
    return {
        "menu_path": menu["menu_path"],
        "menu_name": menu["menu_name"],
        "function_desc": candidate.get("function_desc", ""),
        "action_types": [],
        "user_verbs": [],
        "keywords": keywords,
        "related_concepts": [],
        "sample_queries": sample_queries,
        "embedding_text": build_embedding_text(menu, composed),
        "menu_id": menu["menu_id"],
        "oneline_desc": candidate.get(
            "oneline_desc", f"{menu['menu_name']} 관련 기능"
        ),
        "description_source": "auto_generated_grounded",
        "review_status": "auto_generated_unreviewed",
        "review_reason": reason,
        "review_required": True,
        "source_fingerprint": candidate.get("source_fingerprint", ""),
    }


def prepare_description(
    menu: dict,
    base_menu: dict | None,
    base_description: dict | None,
    review_override: dict | None = None,
    auto_candidate: dict | None = None,
) -> dict:
    same_name = (
        base_menu is not None
        and base_description is not None
        and base_menu["menu_name"] == menu["menu_name"]
    )
    if not same_name:
        if review_override:
            keywords = review_override.get("keywords") or menu["keywords"]
            oneline = (
                review_override.get("oneline_desc")
                or f"{menu['menu_name']} 관련 기능"
            )
            function_desc = review_override.get("function_desc") or oneline
            sample_queries = review_override.get("sample_queries") or []
            composed = {
                "function_desc": function_desc,
                "keywords": keywords,
                "aliases": review_override.get("aliases", []),
            }
            embedding_text = build_embedding_text(menu, composed)
            return {
                "menu_path": menu["menu_path"],
                "menu_name": menu["menu_name"],
                "function_desc": function_desc,
                "action_types": [],
                "user_verbs": [],
                "keywords": keywords,
                "related_concepts": [],
                "sample_queries": sample_queries,
                "embedding_text": embedding_text,
                "menu_id": menu["menu_id"],
                "oneline_desc": oneline,
                "description_source": "business_review_override",
                "review_status": "reviewed_override",
                "review_required": False,
            }
        reason = "new_menu" if base_description is None else "menu_name_changed"
        if auto_candidate:
            return _auto_candidate_description(menu, auto_candidate, reason)
        return _fallback_description(menu, reason)

    description = copy.deepcopy(base_description)
    description["menu_id"] = menu["menu_id"]
    description["menu_name"] = menu["menu_name"]
    description["menu_path"] = menu["menu_path"]
    description["keywords"] = _merge_keywords(
        base_description.get("keywords", []),
        base_menu.get("keywords", []),
        menu.get("keywords", []),
    )
    description["description_source"] = "reviewed_1.5.5_inherited"
    description["review_status"] = "reviewed_inherited"
    description["review_required"] = False
    return description


def build_query_pool(
    source_rows: list[dict],
    menus: list[dict],
    base_menus: dict[str, dict],
    excluded_queries: set[str],
    descriptions: list[dict],
    curated_rows: list[dict] | None = None,
) -> tuple[list[dict], dict]:
    menu_map = {menu["menu_id"]: menu for menu in menus}
    rows = []
    for item in source_rows:
        menu = menu_map.get(str(item.get("menu_id", "")))
        base = base_menus.get(str(item.get("menu_id", "")))
        if not menu or not base or menu["menu_name"] != base["menu_name"]:
            continue
        row = dict(item)
        row["menu_name"] = menu["menu_name"]
        row["menu_path"] = menu["menu_path"]
        row["app_version"] = menu["app_version"]
        rows.append(row)
    for description in descriptions:
        if description.get("description_source") not in {
            "auto_generated_grounded",
            "business_review_override",
        }:
            continue
        menu = menu_map.get(description["menu_id"])
        if not menu:
            continue
        for query in description.get("sample_queries", []):
            rows.append(
                {
                    "query": query,
                    "menu_id": menu["menu_id"],
                    "menu_name": menu["menu_name"],
                    "menu_path": menu["menu_path"],
                    "source": (
                        "business_reviewed"
                        if description.get("description_source")
                        == "business_review_override"
                        else "auto_generated_unreviewed"
                    ),
                    "app_version": menu["app_version"],
                }
            )
    for item in curated_rows or []:
        menu = menu_map.get(str(item.get("menu_id", "")))
        if not menu:
            continue
        reviewed_menu_name = str(item.get("menu_name", "")).strip()
        if reviewed_menu_name and menu["menu_name"] != reviewed_menu_name:
            continue
        shared_queries = {
            normalize_query_text(query)
            for query in item.get("shared_queries", [])
        }
        for query in [
            *item.get("queries", []),
            *item.get("shared_queries", []),
        ]:
            is_shared_query = normalize_query_text(query) in shared_queries
            rows.append(
                {
                    "query": query,
                    "menu_id": menu["menu_id"],
                    "menu_name": menu["menu_name"],
                    "menu_path": menu["menu_path"],
                    "source": "business_qc",
                    "allow_shared_query": is_shared_query,
                    "exact_match_only": bool(
                        is_shared_query
                        and item.get("shared_exact_only", False)
                    ),
                    "required_terms_any": item.get(
                        "required_terms_any", []
                    ),
                    "app_version": menu["app_version"],
                }
            )
    return sanitize_query_rows(rows, excluded_queries=excluded_queries)


def load_held_out_queries(data_dir: Path) -> set[str]:
    queries = set()
    for name in ("eval_queries_dev.json", "eval_queries_gold.json"):
        path = data_dir / name
        if not path.exists():
            continue
        for item in _load_json(path):
            query = normalize_query_text(item.get("query", ""))
            if query:
                queries.add(query)
    return queries


def diff_versions(previous: list[dict] | None, current: list[dict]) -> dict:
    if previous is None:
        return {
            "added": [item["menu_id"] for item in current],
            "removed": [],
            "menu_name_changed": [],
            "menu_path_changed": [],
            "keywords_changed": [],
        }
    before = {item["menu_id"]: item for item in previous}
    after = {item["menu_id"]: item for item in current}
    common = before.keys() & after.keys()
    return {
        "added": sorted(after.keys() - before.keys()),
        "removed": sorted(before.keys() - after.keys()),
        "menu_name_changed": sorted(
            menu_id
            for menu_id in common
            if before[menu_id]["menu_name"] != after[menu_id]["menu_name"]
        ),
        "menu_path_changed": sorted(
            menu_id
            for menu_id in common
            if before[menu_id]["menu_path"] != after[menu_id]["menu_path"]
        ),
        "keywords_changed": sorted(
            menu_id
            for menu_id in common
            if {
                keyword.casefold()
                for keyword in before[menu_id].get("keywords", [])
            }
            != {
                keyword.casefold()
                for keyword in after[menu_id].get("keywords", [])
            }
        ),
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--db-dir", type=Path, default=DEFAULT_DB_DIR)
    parser.add_argument(
        "--versions",
        default="",
        help=(
            "comma-separated production app versions; default discovers "
            f"{DEFAULT_APP_VERSION}+ DB files and excludes 7.x/9.x"
        ),
    )
    args = parser.parse_args()

    versions = (
        [value.strip() for value in args.versions.split(",") if value.strip()]
        if args.versions
        else discover_production_versions(args.db_dir)
    )
    if not versions:
        raise SystemExit(f"no production SQLite files found in {args.db_dir}")
    base_menus = {
        item["menu_id"]: item for item in _load_json(RAW_DIR / "real_menus.json")
    }
    base_descriptions = {
        item["menu_id"]: item
        for item in _load_jsonl(GENERATED_DIR / "menu_descriptions.jsonl")
    }
    source_query_pool = _load_json(
        Path(__file__).parent.parent / "data" / "query_index_pool.json"
    )
    curated_query_aliases = (
        _load_json(CURATED_QUERY_ALIASES_PATH)
        if CURATED_QUERY_ALIASES_PATH.exists()
        else []
    )
    held_out_queries = load_held_out_queries(Path(__file__).parent.parent / "data")
    review_overrides = {}
    if REVIEW_OVERRIDES_PATH.exists():
        for item in _load_json(REVIEW_OVERRIDES_PATH):
            key = (str(item.get("menu_id", "")), str(item.get("menu_name", "")))
            if all(key):
                review_overrides[key] = item
    auto_candidates = {}
    if AUTO_CANDIDATES_PATH.exists():
        for item in _load_json(AUTO_CANDIDATES_PATH):
            key = (
                str(item.get("menu_id", "")),
                str(item.get("menu_name", "")),
                str(item.get("menu_path", "")),
            )
            if all(key) and item.get("review_status") == "auto_generated_unreviewed":
                auto_candidates[key] = item

    review_items: dict[tuple[str, str], dict] = {}
    summaries = []
    previous_menus = None
    for version in versions:
        db_path = args.db_dir / f"{version}_menu.db"
        if not db_path.exists():
            raise SystemExit(f"SQLite file missing: {db_path}")

        menus, update_date = load_sqlite_menus(db_path, version)
        descriptions = [
            prepare_description(
                menu,
                base_menus.get(menu["menu_id"]),
                base_descriptions.get(menu["menu_id"]),
                review_overrides.get((menu["menu_id"], menu["menu_name"])),
                auto_candidates.get(
                    (menu["menu_id"], menu["menu_name"], menu["menu_path"])
                ),
            )
            for menu in menus
        ]
        query_pool, query_stats = build_query_pool(
            source_query_pool,
            menus,
            base_menus,
            held_out_queries,
            descriptions,
            curated_query_aliases,
        )
        diff = diff_versions(previous_menus, menus)
        target_dir = version_dir(version)
        _write_json(target_dir / "real_menus.json", menus)
        _write_jsonl(target_dir / "menu_descriptions.jsonl", descriptions)
        _write_json(target_dir / "query_index_pool.json", query_pool)
        _write_json(target_dir / "menu_diff.json", diff)

        for menu, description in zip(menus, descriptions):
            if description["review_required"]:
                key = (menu["menu_id"], menu["menu_name"])
                item = review_items.setdefault(
                    key,
                    {
                        "first_seen_app_version": version,
                        "affected_app_versions": [],
                        "menu_id": menu["menu_id"],
                        "screen_num": menu["screen_num"],
                        "map_id": menu["map_id"],
                        "menu_name": menu["menu_name"],
                        "menu_path": menu["menu_path"],
                        "keywords": menu["keywords"],
                        "reason": description.get(
                            "review_reason", description["review_status"]
                        ),
                        "generation_source": description.get(
                            "description_source", "unknown"
                        ),
                        "candidate_oneline_desc": description["oneline_desc"],
                        "candidate_function_desc": description["function_desc"],
                        "candidate_embedding_text": description["embedding_text"],
                        "candidate_sample_queries": description.get(
                            "sample_queries", []
                        ),
                    },
                )
                item["affected_app_versions"].append(version)
                item["menu_path"] = menu["menu_path"]
                item["keywords"] = menu["keywords"]
                item["candidate_oneline_desc"] = description["oneline_desc"]
                item["candidate_function_desc"] = description["function_desc"]
                item["candidate_embedding_text"] = description["embedding_text"]
                item["candidate_sample_queries"] = description.get(
                    "sample_queries", []
                )
                item["generation_source"] = description.get(
                    "description_source", "unknown"
                )

        status_counts = Counter(
            description["review_status"] for description in descriptions
        )
        metadata = {
            "app_version": version,
            "sqlite_file": db_path.name,
            "sqlite_update_date": update_date,
            "menu_count": len(menus),
            "query_count": len(query_pool),
            "review_status": dict(status_counts),
            "query_sanitization": query_stats,
            "diff_from_previous": {
                key: len(value) for key, value in diff.items()
            },
        }
        _write_json(target_dir / "version_metadata.json", metadata)
        summaries.append(metadata)
        previous_menus = menus
        print(
            f"{version}: menus={len(menus):,}, queries={len(query_pool):,}, "
            f"review_required={sum(bool(item['review_required']) for item in descriptions):,}"
        )

    review_queue = sorted(
        review_items.values(),
        key=lambda item: (
            tuple(map(int, item["first_seen_app_version"].split("."))),
            int(item["screen_num"]),
        ),
    )
    _write_json(VERSIONS_DIR / "review_queue.json", review_queue)
    _write_review_csv(VERSIONS_DIR / "review_queue.csv", review_queue)
    _write_json(VERSIONS_DIR / "versions.json", summaries)
    print(f"review queue: {len(review_queue):,} rows")
    print(f"output: {VERSIONS_DIR}")


if __name__ == "__main__":
    main()