spb/airiskindex Public

1#  File:    load.py2#  Path:    apps/etl/src/airiskindex_etl/load.py3#  Project: AI Risk Index — airiskindex.io4#  Author:  Simon-Pierre Boucher5#  Contact: contact@spboucher.ai6#  Copyright © 2026 Simon-Pierre Boucher. All rights reserved.7#8#  Description: Load derived CSVs into Postgres (idempotent upserts).910"""Load derived O*NET + OEWS CSVs into Postgres (idempotent upserts).1112Occupations are keyed by O*NET-SOC code; wages join by SOC prefix13("15-1252.00" -> "15-1252"). Demo-seeded rows are overwritten by real data14where codes collide, which is the desired direction of truth.15"""1617from __future__ import annotations1819import csv20import os21from pathlib import Path2223import psycopg2425REPO_ROOT = Path(__file__).resolve().parents[4]26DERIVED = REPO_ROOT / "data" / "derived" / "onet"272829def read(name: str) -> list[dict[str, str]]:30    with (DERIVED / name).open(encoding="utf-8", newline="") as fh:31        return list(csv.DictReader(fh))323334def main() -> None:35    dsn = os.environ["DATABASE_URL"]36    occupations = read("occupations.csv")37    tasks = read("tasks.csv")38    wages = {row["soc"]: row for row in read("wages.csv")}3940    with psycopg.connect(dsn) as conn, conn.cursor() as cur:41        for occ in occupations:42            wage = wages.get(occ["code"].split(".")[0], {})43            median = wage.get("median_wage_cents") or None44            employment = wage.get("employment") or None45            cur.execute(46                """47                INSERT INTO "Occupation" (code, title, description, "medianWageCents",48                                          "wageCurrency", employment)49                VALUES (%s, %s, %s, %s, 'USD', %s)50                ON CONFLICT (code) DO UPDATE SET51                  title = EXCLUDED.title,52                  description = EXCLUDED.description,53                  "medianWageCents" = COALESCE(EXCLUDED."medianWageCents", "Occupation"."medianWageCents"),54                  employment = COALESCE(EXCLUDED.employment, "Occupation".employment)55                """,56                (occ["code"], occ["title"], occ["description"], median, employment),57            )5859        for task in tasks:60            cur.execute(61                """62                INSERT INTO "Task" (id, "occupationCode", statement, importance)63                VALUES (%s, %s, %s, %s)64                ON CONFLICT (id) DO UPDATE SET65                  statement = EXCLUDED.statement,66                  importance = EXCLUDED.importance67                """,68                (69                    task["task_id"],70                    task["code"],71                    task["statement"],72                    float(task["importance"]) if task["importance"] else None,73                ),74            )7576        cur.execute('SELECT count(*) FROM "Occupation"')77        occ_count = cur.fetchone()[0]78        cur.execute('SELECT count(*) FROM "Task"')79        task_count = cur.fetchone()[0]8081    print(f"loaded: {occ_count} occupations, {task_count} tasks (incl. pre-existing)")828384if __name__ == "__main__":85    main()86