#!/usr/bin/env python3 """New-side emitter for the diff harness (spec.md, spec_version 1). Reads the R1 ``public`` schema (the migrated party/engagement model) and emits the SAME canonical JSON document as extract_legacy.py — shared builders in harness/common.py guarantee identical serialization. Spec mappings (how R1 answers the legacy-shaped questions): - operators = parties on the operator side of direct_client engagements (the hierarchy lives on the edges — 02 §2) - clients = the client side of those engagements, nested by operator party name - team_members = one rates row per subcontract engagement, from its default (project-NULL) cost/billout compensation terms; keyed by lower(person party email); active = ended_on IS NULL - months = project_id groups for assigned rows; the raw_* triple VERBATIM for unassigned rows (byte-identical keys to the old side by construction) - invoices = attached entries via the invoice_lines kind='time' lock chain (invoice_line_id) - cc = batches/lines by carried ids Every query is scoped to LEGACY-MIRRORED tenants (MIRRORED_TENANTS): books tenants — sheet-imported or app-owned (ADR #016/#021) — are not rebuilt from ``legacy.*`` and are invisible to the parity comparison by construction. A legacy-only database (every fixture/parity test DB) matches all tenants, so the document is unchanged there. Usage: extract_new.py [--dsn DSN | --env-file PATH] [--out FILE] Without --dsn, WITH_DB_URL (in-container) or WITH_DB_URL_HOST (host-side) is read from the process env or --env-file. """ from __future__ import annotations import argparse import datetime import os import sys from typing import Any try: # imported as harness.extract_new (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] NEW_TABLES = ( "parties", "tenants", "users", "engagements", "compensation_terms", "rate_overrides", "projects", "import_batches", "time_entries", "invoices", "invoice_lines", "cc_expense_batches", "cc_expense_lines", "cc_highlight_keywords", "cc_auto_rules", "comments", "audit_log", ) #: The parity comparison is legacy↔mirror ONLY: tenants whose books come #: from the sheet pipeline or are app-owned (settings.books_owner of #: 'sheet'/'app', or the legacy parallel_run boolean — ADR #016/#021) #: are NOT rebuilt from ``legacy.*`` and must be invisible to the #: harness by construction. On a legacy-only database this predicate #: matches every tenant, so the emitted document is byte-identical to #: the unscoped extraction. MIRRORED_TENANTS = ( "(SELECT id FROM tenants" " WHERE coalesce(settings->>'books_owner', '') NOT IN ('sheet', 'app')" " AND lower(coalesce(settings->>'parallel_run', '')) <> 'true')" ) def extract(conn) -> dict: with conn.cursor() as cur: cur.execute("SET default_transaction_read_only = on") def rows(sql: str, params: tuple = ()) -> list[tuple]: cur.execute(sql, params) return cur.fetchall() def one(sql: str) -> Any: return rows(sql)[0][0] m = MIRRORED_TENANTS # Operator/client inventory from the direct_client engagement edges # (05 step 2.2 seeds one per legacy clients row, so zero-entry # entities still appear — inventory, not usage). operator_rows = rows( "SELECT DISTINCT pa.name FROM engagements e" " JOIN parties pa ON pa.id = e.party_a_id" f" WHERE e.type = 'direct_client' AND e.tenant_id IN {m}") client_rows = rows( "SELECT pa.name, pb.name FROM engagements e" " JOIN parties pa ON pa.id = e.party_a_id" " JOIN parties pb ON pb.id = e.party_b_id" f" WHERE e.type = 'direct_client' AND e.tenant_id IN {m}") project_rows = rows( "SELECT p.id::text, p.name, cp.name, op.name, p.income," " p.billout_adjustment_pct, p.billout_adjustment_amount" " FROM projects p" " JOIN parties cp ON cp.id = p.client_party_id" " JOIN parties op ON op.id = p.operator_party_id" f" WHERE p.tenant_id IN {m}") project_entry_counts = dict(rows( "SELECT project_id::text, count(*)::int FROM time_entries" f" WHERE project_id IS NOT NULL AND tenant_id IN {m} GROUP BY 1")) # One rates row per worker (subcontract) engagement; its default # terms are the project-NULL, currently-effective rows (the # migration seeds exactly one per kind). member_rows = rows( "SELECT pb.email, pb.name, cost.rate_amount, bill.rate_amount," " (e.ended_on IS NULL)" " FROM engagements e" " JOIN parties pb ON pb.id = e.party_b_id" " LEFT JOIN compensation_terms cost ON cost.engagement_id = e.id" " AND cost.kind = 'cost' AND cost.project_id IS NULL" " AND cost.effective_to IS NULL" " LEFT JOIN compensation_terms bill ON bill.engagement_id = e.id" " AND bill.kind = 'billout' AND bill.project_id IS NULL" " AND bill.effective_to IS NULL" f" WHERE e.type = 'subcontract' AND e.tenant_id IN {m}") member_entry_counts = dict(rows( "SELECT pb.email, count(te.id)::int" " FROM (SELECT DISTINCT party_b_id FROM engagements" f" WHERE type = 'subcontract' AND tenant_id IN {m}) w" " JOIN parties pb ON pb.id = w.party_b_id" " LEFT JOIN time_entries te ON te.worker_party_id = pb.id" f" AND te.tenant_id IN {m}" " GROUP BY pb.email")) entities = common.build_entities( operator_rows, client_rows, project_rows, project_entry_counts, member_rows, member_entry_counts, unassigned_entry_count=one( "SELECT count(*)::int FROM time_entries WHERE project_id IS NULL" f" AND tenant_id IN {m}"), ) agg = ("count(*)::int, coalesce(sum(duration_seconds), 0)::bigint," " sum(cost_amount), sum(billout_amount)," " sum(source_rate), sum(source_amount)") months = common.build_months( rows(f"SELECT to_char(start_at, 'YYYY-MM'), project_id::text, {agg}" " FROM time_entries WHERE project_id IS NOT NULL" f" AND tenant_id IN {m} GROUP BY 1, 2"), rows(f"SELECT to_char(start_at, 'YYYY-MM'), raw_operator, raw_client," f" raw_project, {agg}" " FROM time_entries WHERE project_id IS NULL" f" AND tenant_id IN {m} GROUP BY 1, 2, 3, 4"), entities["projects"], ) invoices = common.build_invoices( rows("SELECT id::text, name, status, invoice_date, manual_total" f" FROM invoices WHERE tenant_id IN {m}"), rows("SELECT il.invoice_id::text, te.id::text, te.billout_amount" " FROM time_entries te" " JOIN invoice_lines il ON il.id = te.invoice_line_id" f" WHERE il.kind = 'time' AND te.tenant_id IN {m}"), ) cc = common.build_cc( rows(f"SELECT id::text, name FROM cc_expense_batches" f" WHERE tenant_id IN {m}"), rows("SELECT l.batch_id::text, l.project_id::text, count(*)::int," " sum(l.amount) FROM cc_expense_lines l" " JOIN cc_expense_batches b ON b.id = l.batch_id" f" WHERE b.tenant_id IN {m} GROUP BY 1, 2"), highlight_keyword_count=one( "SELECT count(*)::int FROM cc_highlight_keywords" f" WHERE tenant_id IN {m}"), auto_rule_count=one( f"SELECT count(*)::int FROM cc_auto_rules WHERE tenant_id IN {m}"), ) meta = { "extracted_at": datetime.datetime.now(datetime.UTC) .strftime("%Y-%m-%dT%H:%M:%SZ"), "source_mode": "new-public", "schema": "public", "row_counts": { table: one(f"SELECT count(*)::int FROM {table}") # noqa: S608 for table in NEW_TABLES }, } return common.assemble(months, invoices, cc, entities, meta) def _dsn_from_env(env: dict[str, str]) -> str: url = env.get("WITH_DB_URL") or env.get("WITH_DB_URL_HOST") if not url: raise SystemExit( "missing connection settings: pass --dsn, or provide WITH_DB_URL /" " WITH_DB_URL_HOST (process env or --env-file)") # SQLAlchemy-style URLs in .env carry a driver tag psycopg doesn't want. return url.replace("postgresql+psycopg://", "postgresql://", 1) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Emit the canonical diff-harness JSON from the NEW schema.") parser.add_argument("--dsn", help="Postgres DSN (else WITH_DB_URL[_HOST])") parser.add_argument("--env-file", help="KEY=VALUE file providing WITH_DB_URL[_HOST]") 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(common.load_env_file(args.env_file)) conninfo = _dsn_from_env(env) with psycopg.connect(conninfo, options="-c default_transaction_read_only=on", application_name="with-diff-harness-new") as conn: conn.read_only = True doc = extract(conn) output = common.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())