Closer/seed/build_db.py

#!/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/<version>.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 datetime import datetime
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"
# Everything under assets/ is packaged into the APK, so backups must NOT live there —
# a stray .bak beside the db silently ships to every user. /build is gitignored.
BACKUP_DIR = ROOT / "build" / "db-backups"

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_DIR.mkdir(parents=True, exist_ok=True)
                stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
                backup = BACKUP_DIR / f"app.db.{stamp}.bak"
                shutil.copy2(args.out, backup)
                print(f"Backed up existing asset db -> {backup.relative_to(ROOT)}")
            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())