"""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()