#!/usr/bin/env python3 """ export-review-seed.py — READ-ONLY export of the review app's Brentwood project content, written as neutral seed JSON for Volleyboard. STRICT CONSTRAINTS (do not weaken these): - SELECT-only, ever. This script must never INSERT/UPDATE/DELETE/DDL against the review app's database. * The DB session is put into Postgres read-only mode via psycopg2's `readonly=True` (issues `SET TRANSACTION READ ONLY` under the hood), so any accidental write would be rejected by the server itself. * Every data query is routed through run_select(), which asserts the SQL text starts with SELECT/WITH before executing it. - Never print the DB connection string / credentials anywhere (stdout, stderr, exceptions). Only the *name* of the env var and the *path* of the file it came from are ever printed — never the value. - Never touches any file under /srv/apps/review. The connection URI is only *read* from its .env files at runtime by this process. - Writes only within /srv/apps/volleyboard/tools/ and /srv/apps/volleyboard/seed/. Usage: python3 export-review-seed.py [--project-match brentwood] [--out PATH] Connection resolution: Reads /srv/apps/review/.env.migrate first, then /srv/apps/review/.env, looking for the first present of: SUPABASE_DB_URL, DATABASE_URL, POOLER_URL, SUPABASE_POOLER_URL, SUPABASE_DB_URI, DB_URL. Any standard `postgresql://user:pass@host:port/db` URI works (Session Pooler or otherwise). """ import argparse import json import os import re import sys import urllib.parse from datetime import date, datetime, timezone from decimal import Decimal from uuid import UUID REVIEW_ENV_FILES = [ "/srv/apps/review/.env.migrate", "/srv/apps/review/.env", ] CANDIDATE_VARS = [ "SUPABASE_DB_URL", "DATABASE_URL", "POOLER_URL", "SUPABASE_POOLER_URL", "SUPABASE_DB_URI", "DB_URL", ] DEFAULT_OUT = "/srv/apps/volleyboard/seed/brentwood-raw.json" _URI_RE = re.compile(r"(postgres(?:ql)?://)([^\s'\"]*)") def redact(text): """Scrub any postgres(ql):// URI (which may embed a password) out of a string before it is ever printed or raised. Defense in depth in case a driver error message ever echoes the DSN.""" return _URI_RE.sub(lambda m: m.group(1) + "***REDACTED***", str(text)) def parse_env_file(path): """Return {key: value} from a simple KEY=VALUE .env file. Never logs values — caller is responsible for not printing them.""" pairs = {} if not os.path.exists(path): return pairs with open(path, "r") as f: for raw in f: line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, val = line.split("=", 1) key = key.strip() val = val.strip() if len(val) >= 2 and val[0] == val[-1] and val[0] in ("'", '"'): val = val[1:-1] pairs[key] = val return pairs def find_connection_uri(): """Search REVIEW_ENV_FILES in order for the first CANDIDATE_VARS match. Returns (uri, source_path, var_name). Never prints uri.""" for path in REVIEW_ENV_FILES: pairs = parse_env_file(path) for var in CANDIDATE_VARS: if var in pairs and pairs[var]: return pairs[var], path, var return None, None, None def connect_readonly(uri): import psycopg2 parsed = urllib.parse.urlparse(uri) query = urllib.parse.parse_qs(parsed.query) extra_kwargs = {} if "sslmode" not in query: # Supabase poolers support TLS; require it rather than silently # allowing a plaintext fallback. extra_kwargs["sslmode"] = "require" conn = psycopg2.connect(uri, **extra_kwargs) # Defense in depth: hard-enforce a read-only session at the Postgres # level (in addition to this script only ever issuing SELECTs). conn.set_session(readonly=True, autocommit=True) with conn.cursor() as cur: # Client-session safety net only — not a data statement, so it does # not go through run_select()'s SELECT-only guard. cur.execute("SET statement_timeout = '30000'") # 30s return conn def run_select(cur, sql, params=None): """Execute sql, but only if it is (unambiguously) a read query. This is the single choke point every data query in this script goes through — it exists so a future edit can't accidentally turn this into a write without a loud, immediate failure.""" lowered = sql.strip().lower() if not (lowered.startswith("select") or lowered.startswith("with")): raise RuntimeError( "Refusing to execute a non-SELECT statement — this script is " "SELECT-only by design. Offending SQL: " + sql[:120] ) cur.execute(sql, params or ()) return cur def table_columns(cur, table, schema="public"): run_select( cur, "select column_name from information_schema.columns " "where table_schema = %s and table_name = %s", (schema, table), ) return {r[0] for r in cur.fetchall()} def table_exists(cur, table, schema="public"): run_select( cur, "select 1 from information_schema.tables " "where table_schema = %s and table_name = %s", (schema, table), ) return cur.fetchone() is not None def select_existing(cur, table, desired_cols, where_sql="", where_params=None, order_by=None, schema="public"): """ Build and run `SELECT FROM table WHERE ...`. This makes every query resilient to schema drift and is exactly how "author_email if present" etc. gets implemented — we probe information_schema rather than assume. Returns (rows_as_dicts, present_cols, missing_cols). """ existing = table_columns(cur, table, schema) present = [c for c in desired_cols if c in existing] missing = [c for c in desired_cols if c not in existing] if not present: return [], present, missing cols_sql = ", ".join('"{}"'.format(c) for c in present) sql = 'select {} from "{}"."{}"'.format(cols_sql, schema, table) if where_sql: sql += " where " + where_sql if order_by: sql += " order by " + order_by run_select(cur, sql, where_params) rows = cur.fetchall() return [dict(zip(present, row)) for row in rows], present, missing def find_project(cur, match): desired = ["id", "slug", "name", "created_at", "staging_domain", "live_domain"] return select_existing( cur, "projects", desired, where_sql="slug ilike %s or name ilike %s", where_params=("%{}%".format(match), "%{}%".format(match)), order_by="created_at", ) def pick_project(rows, match): if not rows: return None for row in rows: if str(row.get("slug") or "").lower() == match.lower(): return row for row in rows: if str(row.get("name") or "").lower() == match.lower(): return row return rows[0] def json_default(o): if isinstance(o, UUID): return str(o) if isinstance(o, (datetime, date)): return o.isoformat() if isinstance(o, Decimal): return float(o) return str(o) def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--project-match", default="brentwood", help="ilike substring to find the project by slug/name (default: brentwood)") parser.add_argument("--out", default=DEFAULT_OUT, help="output JSON path") args = parser.parse_args() uri, source_path, var_name = find_connection_uri() if not uri: print( "ERROR: could not find a DB connection URI in {} " "(looked for env vars: {})".format(REVIEW_ENV_FILES, CANDIDATE_VARS), file=sys.stderr, ) sys.exit(1) print("Using connection var {!r} from {} (value not shown).".format(var_name, source_path)) try: import psycopg2 # noqa: F401 except ImportError: print( "ERROR: psycopg2 is not installed in this Python. Run this " "script using /srv/apps/volleyboard/tools/.seedvenv/bin/python3", file=sys.stderr, ) sys.exit(1) try: conn = connect_readonly(uri) except Exception as e: # noqa: BLE001 print("ERROR connecting to review DB: {}".format(redact(e)), file=sys.stderr) sys.exit(1) schema_surprises = [] try: cur = conn.cursor() # 1. Locate the Brentwood project ----------------------------------- project_rows, proj_present, proj_missing = find_project(cur, args.project_match) if proj_missing: schema_surprises.append( "projects: missing expected columns {}".format(proj_missing) ) if not project_rows: print( "ERROR: no project found with slug/name ilike '%{}%'".format(args.project_match), file=sys.stderr, ) sys.exit(2) print("Found {} project candidate(s) matching '{}':".format( len(project_rows), args.project_match)) for r in project_rows: print(" - id={} slug={!r} name={!r} created_at={}".format( r.get("id"), r.get("slug"), r.get("name"), r.get("created_at"))) project = pick_project(project_rows, args.project_match) pid = project["id"] print("Selected project: id={} slug={!r} name={!r}".format( pid, project.get("slug"), project.get("name"))) # 2. Pages ------------------------------------------------------------ pages, pages_present, pages_missing = select_existing( cur, "pages", ["id", "project_id", "page_name", "staging_url", "live_url", "status", "signoff", "tags", "position", "slug", "author_id", "author_email", "created_at"], where_sql="project_id = %s", where_params=(pid,), order_by="position, created_at", ) if pages_missing: schema_surprises.append("pages: missing expected columns {}".format(pages_missing)) page_ids = [p["id"] for p in pages] # 3. Features ---------------------------------------------------------- features, features_present, features_missing = select_existing( cur, "features", ["id", "project_id", "description", "status", "signoff", "position", "author_id", "author_email", "created_at"], where_sql="project_id = %s", where_params=(pid,), order_by="position, created_at", ) if features_missing: schema_surprises.append("features: missing expected columns {}".format(features_missing)) feature_ids = [f["id"] for f in features] # 4. General issues ------------------------------------------------------ issues, issues_present, issues_missing = select_existing( cur, "issues", ["id", "project_id", "title", "content", "body", "category", "type", "status", "action_to", "prev_action_to", "author_id", "author_email", "image_path", "created_at"], where_sql="project_id = %s", where_params=(pid,), order_by="created_at", ) if issues_missing: schema_surprises.append( "issues: missing expected columns {} (actual columns used: {})".format( issues_missing, issues_present) ) issue_ids = [i["id"] for i in issues] # 5. Notes (polymorphic parent: exactly one of page_id / feature_id) ---- notes, notes_present, notes_missing = select_existing( cur, "notes", ["id", "page_id", "feature_id", "content", "image_path", "category", "resolved", "resolved_at", "author_id", "author_email", "created_at"], where_sql="page_id = ANY(%s::uuid[]) or feature_id = ANY(%s::uuid[])", where_params=(page_ids, feature_ids), order_by="created_at", ) if notes_missing: schema_surprises.append("notes: missing expected columns {}".format(notes_missing)) note_ids = [n["id"] for n in notes] # 6. Comments (threaded on notes) ---------------------------------------- comments, comments_present, comments_missing = select_existing( cur, "comments", ["id", "note_id", "content", "image_path", "resolved", "author_id", "author_email", "created_at"], where_sql="note_id = ANY(%s::uuid[])", where_params=(note_ids,), order_by="created_at", ) if comments_missing: schema_surprises.append("comments: missing expected columns {}".format(comments_missing)) # 7. Issue comments (threaded on issues) -------------------------------- issue_comments, ic_present, ic_missing = select_existing( cur, "issue_comments", ["id", "issue_id", "content", "image_path", "resolved", "author_id", "author_email", "created_at"], where_sql="issue_id = ANY(%s::uuid[])", where_params=(issue_ids,), order_by="created_at", ) if ic_missing: schema_surprises.append("issue_comments: missing expected columns {}".format(ic_missing)) # 8. Distinct user emails referenced anywhere in this project's data ---- email_set = set() def add_email(e): if e and isinstance(e, str) and e.strip(): email_set.add(e.strip()) for collection in (pages, features, issues, notes, comments, issue_comments): for row in collection: add_email(row.get("author_email")) for i in issues: for e in (i.get("action_to") or []): add_email(e) for e in (i.get("prev_action_to") or []): add_email(e) profiles_rows = [] if email_set and table_exists(cur, "profiles"): lowered = [e.lower() for e in email_set] profiles_rows, _, _ = select_existing( cur, "profiles", ["id", "email"], where_sql="lower(email) = ANY(%s)", where_params=(lowered,), ) for pr in profiles_rows: add_email(pr.get("email")) users = sorted(email_set, key=lambda s: s.lower()) # 9. Assemble + write ------------------------------------------------------ out = { "exported_at": datetime.now(timezone.utc).isoformat(), "source": "review app (Supabase, read-only export)", "project_match": args.project_match, "project_candidates": project_rows, "project": project, "users": users, "profiles": profiles_rows, "pages": pages, "features": features, "issues": issues, "notes": notes, "comments": comments, "issue_comments": issue_comments, "columns_used": { "projects": proj_present, "pages": pages_present, "features": features_present, "issues": issues_present, "notes": notes_present, "comments": comments_present, "issue_comments": ic_present, }, "schema_surprises": schema_surprises, } out_path = args.out os.makedirs(os.path.dirname(out_path), exist_ok=True) with open(out_path, "w") as f: json.dump(out, f, indent=2, default=json_default) # 10. Report ----------------------------------------------------------------- print("\n=== Export summary ===") print("Project: name={!r} slug={!r} id={}".format( project.get("name"), project.get("slug"), pid)) print("Pages: {}".format(len(pages))) print("Features: {}".format(len(features))) print("Issues: {}".format(len(issues))) print("Notes: {}".format(len(notes))) print("Comments: {}".format(len(comments))) print("Issue comments: {}".format(len(issue_comments))) print("Distinct users: {}".format(len(users))) for e in users: print(" - {}".format(e)) if schema_surprises: print("\nSchema surprises:") for s in schema_surprises: print(" - {}".format(s)) print("\nWrote {}".format(out_path)) finally: conn.close() if __name__ == "__main__": main()