#!/usr/bin/env python3 """ Build the Room-loadable SQLite asset database from the question JSON packs. The app loads this file via Room's `createFromAsset("database/app.db")` with **no migrations and no destructive fallback**, so the output must satisfy Room exactly: * every entity table declared by `AppDatabase` must exist, with Room's exact SQL * `room_master_table` must carry Room's identity hash for the schema version Both of those come from Room's own exported schema (`app/schemas/app.closer.data.local.AppDatabase/.json`, produced by `room.schemaLocation` in app/build.gradle.kts). That file is the single source of truth here — when the Room schema changes, re-run the app build to re-export it and this script follows automatically. Nothing about the schema is hardcoded. `answer_config` is written in the shape the app's parser actually reads (`data/local/mapper/QuestionMapper.kt` -> `parseAnswerConfig`), which uses camelCase keys. The authoring JSON uses snake_case (see seed/questions/QUESTION_SCHEMA.md); translating between the two is this script's job. Usage: python3 seed/build_db.py # build + replace the asset db (backs up first) python3 seed/build_db.py --out /tmp/x.db # build somewhere else (no backup, no replace) python3 seed/build_db.py --check # validate the JSON only, build nothing Any content problem is a hard failure. This script never writes a partial database. """ import argparse import json import shutil import sqlite3 import sys import tempfile from pathlib import Path from typing import Any, Dict, List from validate_question_variety import load_json_records, validate_records ROOT = Path(__file__).resolve().parents[1] JSON_DIR = ROOT / "seed" / "questions" SCHEMA_DIR = ROOT / "app" / "schemas" / "app.closer.data.local.AppDatabase" ASSET_DB = ROOT / "app" / "src" / "main" / "assets" / "database" / "app.db" VALID_TYPES = {"written", "single_choice", "multi_choice", "scale", "this_or_that"} VALID_ACCESS = {"free", "premium"} CHOICE_TYPES = {"single_choice", "multi_choice"} # Room writes its identity row with this fixed id; see RoomOpenHelper. ROOM_MASTER_TABLE_ID = 42 ROOM_MASTER_DDL = ( "CREATE TABLE IF NOT EXISTS room_master_table " "(id INTEGER PRIMARY KEY, identity_hash TEXT)" ) class ContentError(Exception): """A problem in the seed JSON. Always fatal — never import partial content.""" # -------------------------------------------------------------------------- # Room schema # -------------------------------------------------------------------------- def load_room_schema() -> Dict[str, Any]: """Load Room's exported schema (highest version present).""" if not SCHEMA_DIR.is_dir(): raise ContentError( f"Room schema export not found at {SCHEMA_DIR}.\n" "Build the app once (room.schemaLocation is configured) so Room exports it." ) versions = [] for path in SCHEMA_DIR.glob("*.json"): try: versions.append((int(path.stem), path)) except ValueError: continue if not versions: raise ContentError(f"No exported schema json in {SCHEMA_DIR}") _, path = max(versions) return json.loads(path.read_text(encoding="utf-8"))["database"] def create_schema(cursor: sqlite3.Cursor, db_schema: Dict[str, Any]) -> None: """Create every Room entity table + the identity row, exactly as Room expects.""" for entity in db_schema["entities"]: table = entity["tableName"] cursor.execute(entity["createSql"].replace("${TABLE_NAME}", table)) for index in entity.get("indices", []): cursor.execute(index["createSql"].replace("${TABLE_NAME}", table)) for view in db_schema.get("views", []): cursor.execute(view["createSql"].replace("${VIEW_NAME}", view["viewName"])) cursor.execute(ROOM_MASTER_DDL) cursor.execute( "INSERT OR REPLACE INTO room_master_table (id, identity_hash) VALUES (?, ?)", (ROOM_MASTER_TABLE_ID, db_schema["identityHash"]), ) # -------------------------------------------------------------------------- # answer_config: authoring shape (snake_case) -> app parser shape (camelCase) # -------------------------------------------------------------------------- def _options(question: Dict[str, Any]) -> List[Dict[str, str]]: """Options live top-level per the schema guide, mirrored into answer_config.""" opts = question.get("options") or (question.get("answer_config") or {}).get("options") return opts or [] def build_answer_config(question: Dict[str, Any], where: str) -> Dict[str, Any]: """Produce the answer_config the app's QuestionMapper can actually read.""" qtype = question["type"] src = question.get("answer_config") or {} if qtype == "written": return { "type": "written", "config": { "minLength": src.get("min_length", 1), "maxLength": src.get("max_length", 1000), "placeholder": src.get("placeholder", "Write your answer..."), }, } if qtype in CHOICE_TYPES: opts = _options(question) if not opts: raise ContentError(f"{where}: {qtype} has no options") config: Dict[str, Any] = {"options": [{"id": o["id"], "text": o["text"]} for o in opts]} if qtype == "multi_choice": config["maxSelections"] = src.get("max_selections", 0) return {"type": qtype, "config": config} if qtype == "scale": return { "type": "scale", "config": { "minScale": src.get("min", 1), "maxScale": src.get("max", 5), "minLabel": src.get("min_label", ""), "maxLabel": src.get("max_label", ""), "scaleStep": src.get("scale_step", 1), }, } if qtype == "this_or_that": opts = _options(question) if len(opts) != 2: raise ContentError(f"{where}: this_or_that needs exactly 2 options, found {len(opts)}") return { "type": "this_or_that", "config": { "optionA": {"id": opts[0]["id"], "text": opts[0]["text"]}, "optionB": {"id": opts[1]["id"], "text": opts[1]["text"]}, }, } raise ContentError(f"{where}: unsupported type {qtype!r}") # -------------------------------------------------------------------------- # Validation — every problem is fatal # -------------------------------------------------------------------------- def load_packs() -> List[Dict[str, Any]]: """Load and hard-validate every pack. Raises on the first structural problem.""" files = sorted(JSON_DIR.glob("*.json")) if not files: raise ContentError(f"No question JSON found in {JSON_DIR}") packs, seen_ids = [], {} for path in files: name = path.name try: data = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as e: raise ContentError(f"{name}: invalid JSON: {e}") from e category = data.get("category") if not isinstance(category, dict): raise ContentError( f"{name}: missing top-level 'category' object. " "A patch manifest or partial batch must never sit under a production filename." ) for field in ("id", "display_name", "description", "access", "icon_name"): if not category.get(field): raise ContentError(f"{name}: category.{field} is missing or empty") questions = data.get("questions") if not isinstance(questions, list) or not questions: raise ContentError(f"{name}: 'questions' is missing or empty") cid = category["id"] for i, q in enumerate(questions): qid = q.get("id") where = f"{name}:{qid or f'#{i}'}" if not qid: raise ContentError(f"{where}: question has no id") if qid in seen_ids: raise ContentError(f"{where}: duplicate id (also in {seen_ids[qid]})") seen_ids[qid] = name if not q.get("text"): raise ContentError(f"{where}: empty text") if q.get("type") not in VALID_TYPES: raise ContentError(f"{where}: invalid type {q.get('type')!r}") if q.get("access") not in VALID_ACCESS: raise ContentError(f"{where}: invalid access {q.get('access')!r}") if q.get("category_id") != cid: raise ContentError( f"{where}: category_id {q.get('category_id')!r} != category.id {cid!r}" ) depth = q.get("depth", q.get("depth_level")) if not isinstance(depth, int) or isinstance(depth, bool): raise ContentError( f"{where}: depth {depth!r} is not an integer. " "String depth is a future migration and is not production-ready " "(see QUESTION_SCHEMA.md); Room reads depth_level as an integer." ) if not isinstance(q.get("tags"), list): raise ContentError(f"{where}: tags must be an array") # Surfaces option problems now rather than as an unusable question later. build_answer_config(q, where) packs.append({"name": name, "category": category, "questions": questions, "mtime": int(path.stat().st_mtime)}) errors = validate_records(load_json_records(JSON_DIR), "JSON") if errors: raise ContentError("Catalog variety gate failed:\n" + "\n".join(errors)) return packs # -------------------------------------------------------------------------- # Build # -------------------------------------------------------------------------- def build(packs: List[Dict[str, Any]], db_schema: Dict[str, Any], out: Path) -> Dict[str, int]: out.parent.mkdir(parents=True, exist_ok=True) if out.exists(): out.unlink() conn = sqlite3.connect(out) try: cur = conn.cursor() create_schema(cur, db_schema) total = 0 for pack in packs: c = pack["category"] cur.execute( "INSERT OR REPLACE INTO question_category " "(id, display_name, description, access, icon_name) VALUES (?, ?, ?, ?, ?)", (c["id"], c["display_name"], c["description"], c["access"], c["icon_name"]), ) for q in pack["questions"]: where = f"{pack['name']}:{q['id']}" cur.execute( "INSERT OR REPLACE INTO question " "(id, text, category_id, depth_level, is_premium, type, tags, answer_config," " pack_id, created_at, status, sex) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( q["id"], q["text"], q["category_id"], q.get("depth", q.get("depth_level")), 1 if q["access"] == "premium" else 0, q["type"], json.dumps(q.get("tags", []), separators=(",", ":")), json.dumps(build_answer_config(q, where), separators=(",", ":")), None, pack["mtime"], "active", q.get("sex"), ), ) total += 1 conn.commit() finally: conn.close() return {"categories": len(packs), "questions": total} def verify(out: Path, db_schema: Dict[str, Any]) -> None: """Re-open the built file and prove it satisfies Room before it is shipped.""" conn = sqlite3.connect(out) try: tables = {r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='table'")} expected = {e["tableName"] for e in db_schema["entities"]} | {"room_master_table"} missing = expected - tables if missing: raise ContentError(f"built db is missing Room tables: {sorted(missing)}") row = conn.execute( "SELECT id, identity_hash FROM room_master_table").fetchone() if not row or row[0] != ROOM_MASTER_TABLE_ID or row[1] != db_schema["identityHash"]: raise ContentError(f"identity hash mismatch: {row} != {db_schema['identityHash']}") empty = [t for t in ("question", "question_category") if conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] == 0] if empty: raise ContentError(f"built db has empty tables: {empty}") finally: conn.close() def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--out", type=Path, default=ASSET_DB, help=f"output path (default: {ASSET_DB})") ap.add_argument("--check", action="store_true", help="validate the JSON and exit without building") args = ap.parse_args() try: db_schema = load_room_schema() packs = load_packs() print(f"Validated {len(packs)} packs, " f"{sum(len(p['questions']) for p in packs)} questions — catalog gate passed.") if args.check: return 0 replacing_asset = args.out.resolve() == ASSET_DB.resolve() with tempfile.TemporaryDirectory() as tmp: staged = Path(tmp) / "app.db" stats = build(packs, db_schema, staged) verify(staged, db_schema) if replacing_asset and args.out.exists(): backup = args.out.with_suffix(".db.bak") shutil.copy2(args.out, backup) print(f"Backed up existing asset db -> {backup.name}") args.out.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(staged, args.out) except ContentError as e: print(f"\nBUILD FAILED\n{e}", file=sys.stderr) return 1 print(f"Built {args.out}") print(f" schema version {db_schema['version']} · identity {db_schema['identityHash']}") print(f" {stats['categories']} categories · {stats['questions']} questions") return 0 if __name__ == "__main__": sys.exit(main())