#!/usr/bin/env python3 """Apply time_standard_matches.json dossier/workspace link updates to prod.""" from __future__ import annotations import argparse import json import re import subprocess import sys from pathlib import Path from typing import Any PROD_CONTEXT = "arn:aws:eks:us-west-2:557537366020:cluster/devops-prod" DEV_CONTEXT = "arn:aws:eks:us-west-2:552185537300:cluster/devops-staging" NAMESPACE = "investibook" DB_PATH = "/data/db/investibook.db" MATCHES_FILENAME = "time_standard_matches.json" ACTOR = "investibook_bot" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=f"Apply {MATCHES_FILENAME} dossier/workspace link updates to prod.", ) parser.add_argument( "date", help="Export package date, for example 2026-06-01. Reads exports//.", ) parser.add_argument( "--exports-root", default="exports", help="Root directory containing date-named export packages. Default: exports", ) parser.add_argument( "--apply", action="store_true", help="Actually write to the prod SQLite DB. Without this flag, only dry-run.", ) parser.add_argument( "--pod", help="Optional prod backend pod name. If omitted, discover the running ready pod.", ) return parser.parse_args() def normalize_date(value: str) -> str: match = re.fullmatch(r"\s*(\d{4})-(\d{1,2})-(\d{1,2})\s*", value) if not match: raise ValueError("date must be YYYY-MM-DD, for example 2026-06-01") year, month, day = (int(part) for part in match.groups()) return f"{year:04d}-{month:02d}-{day:02d}" def run_command(args: list[str]) -> str: proc = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) if proc.returncode != 0: detail = proc.stderr.strip() or proc.stdout.strip() raise RuntimeError("command failed: " + " ".join(args[:8]) + ("\n" + detail if detail else "")) return proc.stdout def kubectl_base() -> list[str]: if PROD_CONTEXT == DEV_CONTEXT or "devops-prod" not in PROD_CONTEXT: raise RuntimeError("refusing to run: prod context guard failed") return ["kubectl", "--context", PROD_CONTEXT, "-n", NAMESPACE] def find_backend_pod() -> str: output = run_command(kubectl_base() + ["get", "pods", "-l", "app=backend", "-o", "json"]) payload = json.loads(output) candidates: list[str] = [] for item in payload.get("items", []): if item.get("status", {}).get("phase") != "Running": continue conditions = item.get("status", {}).get("conditions", []) ready = any(cond.get("type") == "Ready" and cond.get("status") == "True" for cond in conditions) name = item.get("metadata", {}).get("name") if ready and name: candidates.append(name) if not candidates: raise RuntimeError("no running and ready prod backend pod found") return sorted(candidates)[0] def kubectl_exec_python(pod: str, code: str) -> str: return run_command(kubectl_base() + ["exec", pod, "--", "python3", "-c", code]) def read_json_array(path: Path) -> list[dict[str, Any]]: if not path.exists(): raise FileNotFoundError(str(path)) data = json.loads(path.read_text(encoding="utf-8")) if not isinstance(data, list): raise ValueError(f"{path} must contain a JSON array") for index, item in enumerate(data): if not isinstance(item, dict): raise ValueError(f"{path} item {index} must be an object") return data def load_matches(input_dir: Path) -> list[dict[str, str]]: rows = read_json_array(input_dir / MATCHES_FILENAME) matches: list[dict[str, str]] = [] seen: set[tuple[str, str]] = set() for index, row in enumerate(rows): issue_key = str(row.get("issue_key") or "").strip() workspace_id = str(row.get("workspace_id") or "").strip() file_dossier_id = str(row.get("dossier_id") or "").strip() if not issue_key or not workspace_id: raise ValueError(f"{input_dir / MATCHES_FILENAME} item {index} missing issue_key or workspace_id") key = (issue_key, workspace_id) if key in seen: continue seen.add(key) matches.append( { "issue_key": issue_key, "workspace_id": workspace_id, "file_dossier_id": file_dossier_id, } ) if not matches: raise ValueError(f"{input_dir / MATCHES_FILENAME} contains no linkable issue/workspace pairs") return matches def remote_update(pod: str, matches: list[dict[str, str]], apply: bool) -> dict[str, Any]: payload = { "db_path": DB_PATH, "actor": ACTOR, "apply": apply, "matches": matches, } code = f""" import json import sqlite3 payload = {json.dumps(payload, ensure_ascii=False)!r} req = json.loads(payload) db_path = req["db_path"] actor = req["actor"] apply = bool(req["apply"]) matches = req["matches"] conn = sqlite3.connect(db_path if apply else "file:" + db_path + "?mode=ro", uri=not apply) conn.row_factory = sqlite3.Row link_rows = [] for item in matches: issue_key = item["issue_key"] workspace_id = item["workspace_id"] issue_rows = conn.execute( "SELECT issue_key, dossier_id FROM incident_dossier_issues WHERE issue_key = ? ORDER BY dossier_id", (issue_key,), ).fetchall() current_dossier_id = issue_rows[0]["dossier_id"] if issue_rows else None existing = [] if current_dossier_id: existing = conn.execute( ''' SELECT dossier_id, workspace_id, created_at, created_by FROM incident_dossier_workspaces WHERE dossier_id = ? AND workspace_id = ? ''', (current_dossier_id, workspace_id), ).fetchall() workspace_exists = conn.execute("SELECT 1 FROM workspaces WHERE id = ? LIMIT 1", (workspace_id,)).fetchone() is not None link_rows.append({{ "issue_key": issue_key, "workspace_id": workspace_id, "file_dossier_id": item.get("file_dossier_id") or "", "current_dossier_id": current_dossier_id, "workspace_exists": workspace_exists, "already_linked": bool(existing), "stale_file_dossier": bool(current_dossier_id and item.get("file_dossier_id") and item.get("file_dossier_id") != current_dossier_id), "status": "matched" if current_dossier_id else "missing_issue", }}) inserted = [] if apply: conn.execute("BEGIN") insert_sql = ''' INSERT OR IGNORE INTO incident_dossier_workspaces ( dossier_id, workspace_id, created_at, created_by ) SELECT dossier_id, ?, strftime('%s', 'now'), ? FROM incident_dossier_issues WHERE issue_key = ? LIMIT 1 ''' for item in matches: cur = conn.execute(insert_sql, (item["workspace_id"], actor, item["issue_key"])) inserted.append({{ "issue_key": item["issue_key"], "workspace_id": item["workspace_id"], "rowcount": cur.rowcount, }}) conn.commit() result = {{ "apply": apply, "input_count": len(matches), "matched_count": sum(1 for row in link_rows if row["status"] == "matched"), "missing_issue_count": sum(1 for row in link_rows if row["status"] == "missing_issue"), "already_linked_count": sum(1 for row in link_rows if row.get("already_linked")), "stale_file_dossier_count": sum(1 for row in link_rows if row.get("stale_file_dossier")), "missing_workspace_count": sum(1 for row in link_rows if not row.get("workspace_exists")), "rows": link_rows, "inserts": inserted, "inserted_rowcount": sum(row["rowcount"] for row in inserted), }} print(json.dumps(result, ensure_ascii=False, indent=2)) """ output = kubectl_exec_python(pod, code) return json.loads(output) def main() -> int: args = parse_args() try: selected_date = normalize_date(args.date) input_dir = Path(args.exports_root) / selected_date if not input_dir.exists(): raise FileNotFoundError(str(input_dir)) matches = load_matches(input_dir) pod = args.pod or find_backend_pod() result = remote_update(pod, matches, args.apply) report = { "date": selected_date, "input_file": str((input_dir / MATCHES_FILENAME).resolve()), "mode": "apply" if args.apply else "dry-run", "prod_context": PROD_CONTEXT, "namespace": NAMESPACE, "backend_pod": pod, "result": result, } print(json.dumps(report, ensure_ascii=False, indent=2)) if not args.apply: print("\nDRY-RUN only. Re-run with --apply to write these prod updates.", file=sys.stderr) return 0 except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())