spb/airiskindex Public

1#  File:    transform.py2#  Path:    apps/etl/src/airiskindex_etl/transform.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: Transform raw O*NET + OEWS dumps into derived CSVs with a manifest.910"""Transform raw O*NET + OEWS dumps into derived CSVs with a manifest.1112Raw is never edited in place; derived payloads are gitignored, only the13manifest (hashes + row counts) is committed (CLAUDE.md §5).14"""1516from __future__ import annotations1718import csv19import hashlib20import json21from pathlib import Path2223from openpyxl import load_workbook2425REPO_ROOT = Path(__file__).resolve().parents[4]26RAW_ONET = REPO_ROOT / "data" / "raw" / "onet"27RAW_OEWS = REPO_ROOT / "data" / "raw" / "oews"28DERIVED = REPO_ROOT / "data" / "derived" / "onet"293031def read_onet_table(name: str) -> list[dict[str, str]]:32    # The text dump extracts into a versioned subdirectory (db_30_3_text/).33    path = next(RAW_ONET.glob(f"**/{name}"))34    with path.open(encoding="utf-8", newline="") as fh:35        return list(csv.DictReader(fh, delimiter="\t"))363738def transform_occupations() -> list[dict[str, str]]:39    rows = read_onet_table("Occupation Data.txt")40    return [41        {42            "code": row["O*NET-SOC Code"],43            "title": row["Title"],44            "description": row["Description"],45        }46        for row in rows47    ]484950def transform_tasks() -> list[dict[str, str]]:51    statements = read_onet_table("Task Statements.txt")52    ratings = read_onet_table("Task Ratings.txt")5354    # Importance = Task Ratings rows with Scale ID "IM" (1–5), one per task.55    importance: dict[str, str] = {}56    for row in ratings:57        if row["Scale ID"] == "IM" and row.get("Recommend Suppress", "N") != "Y":58            importance[row["Task ID"]] = row["Data Value"]5960    return [61        {62            "task_id": row["Task ID"],63            "code": row["O*NET-SOC Code"],64            "statement": row["Task"],65            "task_type": row.get("Task Type", ""),66            "importance": importance.get(row["Task ID"], ""),67        }68        for row in statements69    ]707172def transform_wages() -> list[dict[str, str]]:73    """OEWS national medians per detailed SOC: annual median (cents) + employment."""74    xlsx = next(RAW_OEWS.glob("**/national_M2025_dl.xlsx"), None) or next(75        RAW_OEWS.glob("**/*_dl.xlsx")76    )77    sheet = load_workbook(xlsx, read_only=True).active78    header = [str(cell.value).strip().upper() for cell in next(sheet.iter_rows(max_row=1))]79    idx = {name: header.index(name) for name in ("OCC_CODE", "O_GROUP", "TOT_EMP", "A_MEDIAN")}8081    out: list[dict[str, str]] = []82    for row in sheet.iter_rows(min_row=2, values_only=True):83        if str(row[idx["O_GROUP"]]).strip() != "detailed":84            continue85        soc = str(row[idx["OCC_CODE"]]).strip()86        median = row[idx["A_MEDIAN"]]87        employment = row[idx["TOT_EMP"]]88        # "#" = wage above the top-code (~$239,200); "*" / "**" = unavailable.89        if median == "#":90            median_cents = 23_920_00091        elif isinstance(median, (int, float)):92            median_cents = int(round(float(median) * 100))93        else:94            median_cents = ""95        out.append(96            {97                "soc": soc,98                "median_wage_cents": str(median_cents),99                "employment": str(int(employment)) if isinstance(employment, (int, float)) else "",100            }101        )102    return out103104105def write_csv(path: Path, rows: list[dict[str, str]]) -> dict[str, object]:106    path.parent.mkdir(parents=True, exist_ok=True)107    with path.open("w", encoding="utf-8", newline="") as fh:108        writer = csv.DictWriter(fh, fieldnames=list(rows[0].keys()))109        writer.writeheader()110        writer.writerows(rows)111    return {112        "file": path.name,113        "rows": len(rows),114        "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),115    }116117118def main() -> None:119    onet_manifest = json.loads((RAW_ONET / "manifest.json").read_text())120    entries = [121        write_csv(DERIVED / "occupations.csv", transform_occupations()),122        write_csv(DERIVED / "tasks.csv", transform_tasks()),123        write_csv(DERIVED / "wages.csv", transform_wages()),124    ]125    (DERIVED / "manifest.json").write_text(126        json.dumps(127            {"source_onet_version": onet_manifest["version"], "outputs": entries}, indent=2128        )129    )130    for entry in entries:131        print(f"{entry['file']}: {entry['rows']} rows")132133134if __name__ == "__main__":135    main()136