"""Load a security-content export (quiz translations / questions / awareness materials / answer-library entries) into a Helix database. Idempotent. The content itself is NOT in the repository — by decision (2026-09-07) it is data, not code, and a 10k-line data PR is not reviewable. It lives in a JSON export (``~/.helix-content/security-content-.json`` on the authoring machine) and is loaded once per environment with this script. The feature that reads it (``secquiz_questions.localized``, migration 627) IS in the repo. Every step is "insert / fill only what is missing", so re-running is a no-op and nothing an admin has since edited is touched: * quiz translations → ``localized`` written only where it is NULL, matched on ``prompt`` (unique across the bank) * new quiz questions → skipped when the prompt already exists * awareness materials → skipped when the Chinese title already exists * answer library → skipped when ``question_zh`` already exists Uses the SYNC driver (``DATABASE_URL_SYNC``, pymysql): from a laptop the async driver hangs silently against TiDB Cloud, and this is a one-off writer with a few hundred rows, not a request path. Usage:: DATABASE_URL_SYNC='mysql+pymysql://...' \\ .venv/bin/python scripts/load_security_content.py export.json --dry-run ...same without --dry-run to write. """ from __future__ import annotations import argparse import json import os import sys import uuid from datetime import UTC, datetime from pathlib import Path from typing import Any from sqlalchemy import create_engine, text sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.domains.security_quiz import localization # noqa: E402 QUESTIONS = "secquiz_questions" MATERIALS = "secaw_materials" ANSWERS = "auditor_answer_library_entries" def _j(value: Any) -> Any: return json.loads(value) if isinstance(value, (str, bytes)) else value def _validated(prompt: str, options: list, localized: dict) -> dict: """Re-validate the export against the same rule the API enforces — option keys must match the baseline — so a hand-edited export cannot smuggle a mis-keyed block past the grader.""" try: return localization.validate_localized(localized, baseline_options=options) except localization.LocalizationError as exc: raise SystemExit(f"invalid translation for {prompt[:40]!r}: {exc}") from exc def load_quiz_translations(conn, rows: list[dict], write: bool) -> tuple[int, int]: have = { r[0]: (r[1], _j(r[2]), r[3]) for r in conn.execute(text(f"SELECT prompt, id, options, localized FROM {QUESTIONS}")) # noqa: S608 } filled = skipped = 0 for item in rows: row = have.get(item["prompt"]) if row is None or row[2] is not None: skipped += 1 continue blocks = _validated(item["prompt"], row[1], item["localized"]) if write: conn.execute( text(f"UPDATE {QUESTIONS} SET localized = :loc WHERE id = :id"), # noqa: S608 {"loc": json.dumps(blocks, ensure_ascii=False), "id": row[0]}, ) filled += 1 return filled, skipped def load_new_questions(conn, rows: list[dict], write: bool) -> tuple[int, int]: existing = {r[0] for r in conn.execute(text(f"SELECT prompt FROM {QUESTIONS}"))} # noqa: S608 inserted = skipped = 0 for q in rows: if q["prompt"] in existing: skipped += 1 continue existing.add(q["prompt"]) blocks = _validated(q["prompt"], q["options"], q.get("localized") or {}) if write: conn.execute( text( f"INSERT INTO {QUESTIONS} " # noqa: S608 "(id, prompt, options, qtype, correct_key, explanation, localized, " " category, difficulty, status, source, created_at, updated_at) " "VALUES (:id, :prompt, :options, :qtype, :ck, :expl, :loc, " " :cat, :diff, 'active', 'import', NOW(), NOW())", ), { "id": str(uuid.uuid4()), "prompt": q["prompt"], "options": json.dumps(q["options"], ensure_ascii=False), "qtype": q.get("qtype", "single"), "ck": q["correct_key"], "expl": q.get("explanation"), "loc": json.dumps(blocks, ensure_ascii=False) if blocks else None, "cat": q.get("category", "general"), "diff": q.get("difficulty", "normal"), }, ) inserted += 1 return inserted, skipped def load_materials(conn, rows: list[dict], write: bool) -> tuple[int, int]: titles: set[str] = set() for (raw,) in conn.execute(text(f"SELECT localized FROM {MATERIALS}")): # noqa: S608 for block in (_j(raw) or {}).values(): if isinstance(block, dict) and block.get("title"): titles.add(str(block["title"]).strip()) now = datetime.now(UTC).replace(tzinfo=None) inserted = skipped = 0 for m in rows: zh_title = m["localized"]["zh"]["title"].strip() if zh_title in titles: skipped += 1 continue titles.add(zh_title) if write: conn.execute( text( f"INSERT INTO {MATERIALS} " # noqa: S608 "(id, kind, category, tags, localized, status, is_pinned, " " display_order, version, published_at, created_at, updated_at) " "VALUES (:id, 'article', :cat, :tags, :loc, 'published', :pinned, " " :order_, 1, :now, :now, :now)", ), { "id": str(uuid.uuid4()), "cat": m["category"], "tags": json.dumps(m.get("tags") or [], ensure_ascii=False), "loc": json.dumps(m["localized"], ensure_ascii=False), "pinned": 1 if m.get("is_pinned") else 0, "order_": int(m.get("display_order") or 0), "now": now, }, ) inserted += 1 return inserted, skipped def load_answers(conn, batches: dict[str, list[dict]], write: bool) -> tuple[int, int]: existing = {r[0] for r in conn.execute(text(f"SELECT question_zh FROM {ANSWERS}"))} # noqa: S608 inserted = skipped = 0 for key, visible in (("employee_visible", 1), ("compliance_evidence", 0)): for e in batches.get(key) or []: if e["question_zh"] in existing: skipped += 1 continue existing.add(e["question_zh"]) if write: conn.execute( text( f"INSERT INTO {ANSWERS} " # noqa: S608 "(id, category, question_zh, question_en, answer_zh, answer_en, " " tags, employee_phrasings, status, usage_count, employee_visible, " " is_featured, employee_view_count, employee_helpful_count, " " employee_unhelpful_count, review_interval_months, " " backing_control_keys, created_at, updated_at) " "VALUES (:id, :cat, :qzh, :qen, :azh, :aen, :tags, :phr, " " 'active', 0, :vis, :feat, 0, 0, 0, 12, :ctrl, NOW(), NOW())", ), { "id": str(uuid.uuid4()), "cat": e["category"], "qzh": e["question_zh"], "qen": e["question_en"], "azh": e["answer_zh"], "aen": e["answer_en"], "tags": json.dumps(e.get("tags") or [], ensure_ascii=False), "phr": e.get("employee_phrasings"), "vis": visible, "feat": 1 if e.get("is_featured") else 0, # NULL, not '[]', when unlinked — the freshness job reads # "has backing controls" as a predicate. "ctrl": json.dumps(e["backing_control_keys"]) if e.get("backing_control_keys") else None, }, ) inserted += 1 return inserted, skipped def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("export", type=Path) ap.add_argument("--dry-run", action="store_true", help="count only, write nothing") args = ap.parse_args() url = os.environ.get("DATABASE_URL_SYNC") if not url or not url.startswith("mysql+pymysql://"): raise SystemExit("DATABASE_URL_SYNC (mysql+pymysql://...) is required") data = json.loads(args.export.read_text(encoding="utf-8")) write = not args.dry_run engine = create_engine(url, connect_args={"connect_timeout": 10, "read_timeout": 120}) with engine.begin() as conn: banner = conn.execute(text("SELECT VERSION()")).scalar() has_col = conn.execute(text( "SELECT 1 FROM information_schema.columns WHERE table_schema = DATABASE() " "AND table_name = :t AND column_name = 'localized'"), {"t": QUESTIONS}).scalar() if not has_col: raise SystemExit("secquiz_questions.localized is missing — deploy migration 627 first") print(f"target: {banner} | mode: {'DRY RUN' if args.dry_run else 'WRITE'}") print("quiz translations filled/skipped:", load_quiz_translations(conn, data.get("quiz_translations", []), write)) print("quiz new questions inserted/skipped:", load_new_questions(conn, data.get("quiz_new_questions", []), write)) print("materials inserted/skipped:", load_materials(conn, data.get("materials", []), write)) print("answers inserted/skipped:", load_answers(conn, data.get("answers", {}), write)) if __name__ == "__main__": main()