#!/usr/bin/env python3 """Old-side extractor for the diff harness (spec.md, spec_version 1). Reads the legacy database — live Supabase (tables in ``public``) or the nightly-landed snapshot (same tables in ``legacy``) — and emits the canonical JSON document. Read-only by construction: SELECTs only, and the session forces ``default_transaction_read_only``. The section builders and serialization rules are shared with the new-side emitter — see harness/common.py. Usage: extract_legacy.py [--dsn DSN | --env-file PATH] [--schema public|legacy] [--out FILE] Without --dsn, the connection is built from SUPABASE_DB_HOST / _PORT / _USER / _NAME / _PASSWORD (process env, or --env-file, which is read by this script so the secret never touches the shell). """ from __future__ import annotations import argparse import datetime import os import sys from typing import Any try: # imported as harness.extract_legacy (app/nightly) or run as a script from harness import common except ImportError: # pragma: no cover - script-mode fallback import common # type: ignore[no-redef] # Re-exported for backwards compatibility (harness tests import from here). SPEC_VERSION = common.SPEC_VERSION CURRENCY = common.CURRENCY qstr = common.qstr money = common.money money_sum = common.money_sum unassigned_key = common.unassigned_key canonical_dumps = common.canonical_dumps adjusted_billout = common.adjusted_billout build_months = common.build_months build_invoices = common.build_invoices build_cc = common.build_cc build_entities = common.build_entities assemble = common.assemble load_env_file = common.load_env_file SCHEMAS = ("public", "legacy") ENV_VARS = { "host": "SUPABASE_DB_HOST", "port": "SUPABASE_DB_PORT", "user": "SUPABASE_DB_USER", "dbname": "SUPABASE_DB_NAME", "password": "SUPABASE_DB_PASSWORD", } TABLES = ( "organizations", "profiles", "organization_members", "teams", "team_members", "clockify_imports", "time_entries", "comments", "operators", "clients", "projects", "pending_imports", "project_rate_overrides", "invoices", "cc_expense_batches", "cc_expense_lines", "cc_highlight_keywords", "cc_auto_rules", ) # --- database access ---------------------------------------------------- def conninfo_from_env(env: dict[str, str]) -> str: import psycopg.conninfo missing = [v for v in ENV_VARS.values() if not env.get(v)] if missing: raise SystemExit(f"missing connection settings: {', '.join(missing)} " "(pass --dsn or --env-file, or export them)") return psycopg.conninfo.make_conninfo( **{arg: env[var] for arg, var in ENV_VARS.items()}) def detect_schema(cur) -> str: cur.execute( "SELECT EXISTS (SELECT 1 FROM information_schema.tables" " WHERE table_schema = 'legacy' AND table_name = 'time_entries')") return "legacy" if cur.fetchone()[0] else "public" def table_exists(cur, schema: str, table: str) -> bool: cur.execute( "SELECT EXISTS (SELECT 1 FROM information_schema.tables" " WHERE table_schema = %s AND table_name = %s)", (schema, table)) return cur.fetchone()[0] def extract(conn, schema_override: str | None = None) -> dict: with conn.cursor() as cur: cur.execute("SET default_transaction_read_only = on") schema = schema_override or detect_schema(cur) if schema not in SCHEMAS: # interpolated below — allowlist, not quoting raise SystemExit(f"unsupported schema {schema!r}") s = schema def rows(sql: str, params: tuple = ()) -> list[tuple]: cur.execute(sql.format(s=s), params) return cur.fetchall() def one(sql: str) -> Any: return rows(sql)[0][0] operator_rows = rows("SELECT name FROM {s}.operators") client_rows = rows( "SELECT o.name, c.name FROM {s}.clients c" " JOIN {s}.operators o ON o.id = c.operator_id") project_rows = rows( "SELECT p.id::text, p.name, c.name, o.name, p.income_usd," " p.billout_adjustment_pct, p.billout_adjustment_amount" " FROM {s}.projects p" " JOIN {s}.clients c ON c.id = p.client_id" " JOIN {s}.operators o ON o.id = c.operator_id") project_entry_counts = dict(rows( "SELECT project_id::text, count(*)::int FROM {s}.time_entries" " WHERE project_id IS NOT NULL GROUP BY 1")) member_rows = rows( "SELECT lower(email), display_name, cost_rate_usd," " billout_rate_usd, is_active FROM {s}.team_members") member_entry_counts = dict(rows( "SELECT lower(tm.email), count(te.id)::int FROM {s}.team_members tm" " LEFT JOIN {s}.time_entries te ON te.team_member_id = tm.id" " GROUP BY 1")) entities = build_entities( operator_rows, client_rows, project_rows, project_entry_counts, member_rows, member_entry_counts, unassigned_entry_count=one( "SELECT count(*)::int FROM {s}.time_entries" " WHERE project_id IS NULL"), ) agg = ("count(*)::int, coalesce(sum(duration_seconds), 0)::bigint," " sum(billout_cost_usd), sum(billout_amount_usd)," " sum(source_rate_usd), sum(source_amount_usd)") months = build_months( rows(f"SELECT to_char(start_at, 'YYYY-MM'), project_id::text, {agg}" " FROM {s}.time_entries WHERE project_id IS NOT NULL" " GROUP BY 1, 2"), rows(f"SELECT to_char(start_at, 'YYYY-MM'), operator, client," f" project, {agg}" " FROM {s}.time_entries WHERE project_id IS NULL" " GROUP BY 1, 2, 3, 4"), entities["projects"], ) invoices = build_invoices( rows("SELECT id::text, name, status, invoice_date," " manual_total_usd FROM {s}.invoices"), rows("SELECT invoice_id::text, id::text, billout_amount_usd" " FROM {s}.time_entries WHERE invoice_id IS NOT NULL"), ) cc = build_cc( rows("SELECT id::text, name FROM {s}.cc_expense_batches"), rows("SELECT batch_id::text, project_id::text, count(*)::int," " sum(amount_usd) FROM {s}.cc_expense_lines GROUP BY 1, 2"), highlight_keyword_count=one( "SELECT count(*)::int FROM {s}.cc_highlight_keywords"), auto_rule_count=one( "SELECT count(*)::int FROM {s}.cc_auto_rules"), ) row_counts = { table: (rows(f"SELECT count(*)::int FROM {{s}}.{table}")[0][0] if table_exists(cur, s, table) else None) for table in TABLES } meta = { "extracted_at": datetime.datetime.now(datetime.UTC) .strftime("%Y-%m-%dT%H:%M:%SZ"), "source_mode": "supabase-public" if s == "public" else "legacy-schema", "schema": s, "row_counts": row_counts, } return assemble(months, invoices, cc, entities, meta) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Emit the canonical diff-harness JSON from the legacy DB.") parser.add_argument("--dsn", help="Postgres DSN (else SUPABASE_DB_* vars)") parser.add_argument("--env-file", help="KEY=VALUE file providing SUPABASE_DB_* vars") parser.add_argument("--schema", choices=SCHEMAS, help="override source-mode auto-detection") parser.add_argument("--out", help="write here instead of stdout") args = parser.parse_args(argv) import psycopg if args.dsn: conninfo = args.dsn else: env = dict(os.environ) if args.env_file: env.update(load_env_file(args.env_file)) conninfo = conninfo_from_env(env) with psycopg.connect(conninfo, options="-c default_transaction_read_only=on", application_name="with-diff-harness") as conn: conn.read_only = True doc = extract(conn, args.schema) output = canonical_dumps(doc) if args.out: with open(args.out, "w", encoding="utf-8") as fh: fh.write(output) else: sys.stdout.write(output) return 0 if __name__ == "__main__": sys.exit(main())