#!/usr/bin/env python3 """ Ingest AWS Activate daily CSV into program_application (TiDB / MySQL). Usage: python ingest_aws_activate_batch.py --csv sample_aws_activate_batch.csv --actor ingest-cron python ingest_aws_activate_batch.py --csv batch.csv --dry-run Environment: PROGRAM_OPS_DATABASE_URL mysql://user:pass@host:4000/program_ops Or: TIDB_HOST, TIDB_PORT, TIDB_USER, TIDB_PASSWORD, TIDB_DATABASE """ from __future__ import annotations import argparse import csv import json import os import sys import uuid from datetime import datetime, timedelta from pathlib import Path from typing import Any from urllib.parse import urlparse try: import pymysql except ImportError: pymysql = None # type: ignore REQUIRED_COLUMNS = {"external_ref", "applicant_email"} OPTIONAL_COLUMNS = {"company_name", "aws_submission_date", "website", "country"} def get_connection(): if pymysql is None: raise RuntimeError("Install pymysql: pip install pymysql") url = os.environ.get("PROGRAM_OPS_DATABASE_URL") if url: parsed = urlparse(url.replace("mysql://", "mysql://", 1)) return pymysql.connect( host=parsed.hostname or "127.0.0.1", port=parsed.port or 4000, user=parsed.username or "root", password=parsed.password or "", database=(parsed.path or "/program_ops").lstrip("/"), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) return pymysql.connect( host=os.environ.get("TIDB_HOST", "127.0.0.1"), port=int(os.environ.get("TIDB_PORT", "4000")), user=os.environ.get("TIDB_USER", "root"), password=os.environ.get("TIDB_PASSWORD", ""), database=os.environ.get("TIDB_DATABASE", "program_ops"), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) def parse_date(value: str | None) -> str | None: if not value or not str(value).strip(): return None raw = str(value).strip()[:10] for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%d/%m/%Y"): try: return datetime.strptime(raw, fmt).strftime("%Y-%m-%d") except ValueError: continue return None def normalize_email(email: str) -> str: return email.strip().lower() def read_rows(csv_path: Path) -> list[dict[str, Any]]: with csv_path.open(newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) if not reader.fieldnames: raise ValueError("CSV has no header row") headers = {h.strip() for h in reader.fieldnames if h} missing = REQUIRED_COLUMNS - headers if missing: raise ValueError(f"CSV missing required columns: {missing}") return [dict(row) for row in reader] def insert_application( conn, row: dict[str, Any], *, actor: str, batch_file: str, row_number: int, dry_run: bool, ) -> str | None: external_ref = (row.get("external_ref") or "").strip() email = normalize_email(row.get("applicant_email") or "") if not external_ref or not email: raise ValueError("external_ref and applicant_email required") application_id = str(uuid.uuid4()) company_name = (row.get("company_name") or "").strip() or None aws_date = parse_date(row.get("aws_submission_date")) sla_due = datetime.utcnow() + timedelta(days=1) if dry_run: print(f"[dry-run] would insert {external_ref} -> {email}") return application_id with conn.cursor() as cur: cur.execute( "SELECT application_id FROM program_application WHERE external_ref = %s", (external_ref,), ) if cur.fetchone(): print(f"skip duplicate external_ref={external_ref}") return None cur.execute( """ INSERT INTO program_application ( application_id, source, region, applicant_email, company_name, aws_submission_date, external_ref, status, assigned_to, sla_due_at ) VALUES (%s, 'aws_activate', 'EMEA', %s, %s, %s, %s, 'received', 'program-ops', %s) """, ( application_id, email, company_name, aws_date, external_ref, sla_due.strftime("%Y-%m-%d %H:%M:%S"), ), ) cur.execute( """ INSERT INTO program_application_event ( application_id, from_status, to_status, event_type, actor, payload_json ) VALUES (%s, NULL, 'received', 'ingest', %s, %s) """, ( application_id, actor, json.dumps( { "batch_file": batch_file, "row_number": row_number, "website": row.get("website"), "country": row.get("country"), } ), ), ) conn.commit() print(f"inserted {external_ref} application_id={application_id}") return application_id def insert_dlq(conn, batch_file: str, row_number: int, row: dict, error: str, dry_run: bool): if dry_run: print(f"[dry-run] DLQ row {row_number}: {error}") return with conn.cursor() as cur: cur.execute( """ INSERT INTO program_ingest_dlq (batch_file, row_number, raw_payload, error_message) VALUES (%s, %s, %s, %s) """, (batch_file, row_number, json.dumps(row), error), ) conn.commit() def main() -> int: parser = argparse.ArgumentParser(description="Ingest AWS Activate CSV batch") parser.add_argument("--csv", required=True, type=Path, help="Path to AWS batch CSV") parser.add_argument("--actor", default="ingest-cron", help="Actor label for audit") parser.add_argument("--dry-run", action="store_true", help="Parse only, no DB writes") args = parser.parse_args() if not args.csv.exists(): print(f"File not found: {args.csv}", file=sys.stderr) return 1 rows = read_rows(args.csv) batch_file = str(args.csv.resolve()) conn = None if not args.dry_run: try: conn = get_connection() except Exception as e: print(f"Database connection failed: {e}", file=sys.stderr) print("Set PROGRAM_OPS_DATABASE_URL or TIDB_* env vars.", file=sys.stderr) return 1 inserted = skipped = failed = 0 for i, row in enumerate(rows, start=2): try: aid = insert_application( conn, row, actor=args.actor, batch_file=batch_file, row_number=i, dry_run=args.dry_run, ) if aid and args.dry_run: inserted += 1 elif aid: inserted += 1 else: skipped += 1 except Exception as e: failed += 1 print(f"row {i} failed: {e}", file=sys.stderr) if conn: insert_dlq(conn, batch_file, i, row, str(e), args.dry_run) if conn: conn.close() print(f"done: inserted={inserted} skipped={skipped} failed={failed}") return 0 if failed == 0 else 2 if __name__ == "__main__": sys.exit(main())