375 lines
15 KiB
Python
375 lines
15 KiB
Python
#!/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 pack_id_for(category: Dict[str, Any]) -> str:
|
|
"""The pack a question belongs to.
|
|
|
|
`question.pack_id` is plumbed all the way to the domain model and is what lets content be
|
|
handled per-pack (query it, gate it, remove it) rather than only per-category. Bundled packs
|
|
get one too, so a later purchased/imported pack is not a special case — it is just another
|
|
pack_id in the same table.
|
|
|
|
Prefer the logical pack id a pack declares in its metadata (the daily pack ships as category
|
|
`daily_fun_mc` but is logically `daily_single_choice_weekly_v1`); otherwise the category id.
|
|
"""
|
|
meta = category.get("metadata") or {}
|
|
return meta.get("pack_id") or category["id"]
|
|
|
|
|
|
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"]
|
|
pack_id = pack_id_for(c)
|
|
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=(",", ":")),
|
|
pack_id,
|
|
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())
|