skillfactor-pipeline/pipeline/p7_backfill_onet.py

"""Phase 7 — backfill O*NET mappings for packages the official crosswalk missed.

The ESCO<->O*NET crosswalk covers only ~43% of the catalog. For every package
with ids.onet_soc == null this script assigns the nearest O*NET occupation:

  1. Candidate SOCs = O*NET occupations that the OFFICIAL crosswalk uses for
     other ESCO occupations in the same ISCO group (4-digit, fallback 3-, 2-digit).
  2. Best candidate = highest title-token overlap between the ESCO title/alt
     labels and the O*NET title; ties -> most frequent SOC in the group.
  3. No candidates at any prefix level -> package stays unmapped (reported).

Per remapped package it regenerates references/tasks.md + tools.md (p2 format,
with an explicit "manual nearest-occupation mapping" note), updates manifest
ids, adds the SKILL.md Hot-technologies section + fixes the sources footer,
and re-inserts the market-evidence section for packages whose evidence is
already aggregated. PROVENANCE refresh happens via p3d afterwards.

Run:  python pipeline/p7_backfill_onet.py [--dry-run]
Idempotent: only touches packages with missing onet_soc.
"""
import json
import os
import re
import sys
from collections import Counter, defaultdict

sys.path.insert(0, os.path.dirname(__file__))
from db import connect
import p3c_aggregate as p3c
import progress

BASE = os.path.join(os.path.dirname(__file__), "..")
SKILLS = os.path.join(BASE, "skills")

STOP = {"and", "of", "the", "for", "in", "or", "a", "an", "to", "on", "&",
        "specialised", "specialized", "general", "worker", "workers",
        "officer", "operator", "technician", "manager", "assistant"}


def tokens(s):
    return {w for w in re.split(r"[^a-z0-9]+", (s or "").lower()) if w and w not in STOP}


def main():
    dry = "--dry-run" in sys.argv
    cn = connect()
    cur = cn.cursor()

    # Official crosswalk pairs + ESCO ISCO groups -> learn ISCO->SOC usage
    cur.execute("""SELECT e.isco_group, c.onet_id, c.onet_title
                   FROM crosswalk_esco_onet c
                   JOIN esco_occupation e ON e.concept_uri = c.esco_uri""")
    isco_soc = defaultdict(Counter)          # isco4 -> Counter(soc)
    soc_title = {}
    for isco, soc, title in cur.fetchall():
        isco = str(isco or "").strip()
        if not isco or not soc:
            continue
        isco_soc[isco][soc] += 1
        soc_title[soc] = title
    # also index by 3- and 2-digit prefixes
    isco_soc3, isco_soc2 = defaultdict(Counter), defaultdict(Counter)
    for isco, ctr in isco_soc.items():
        isco_soc3[isco[:3]].update(ctr)
        isco_soc2[isco[:2]].update(ctr)

    # ESCO alt labels for token matching
    cur.execute("SELECT concept_uri, preferred_label, alt_labels FROM esco_occupation")
    esco_alt = {u: (pl, al) for u, pl, al in cur.fetchall()}

    # evidence-done slugs (need market section re-inserted after tools.md regen)
    state = progress.load()
    ev_done = {s for s, o in state["occupations"].items() if o.get("evidence") == "done"}

    mapped = skipped = 0
    unmatched = []
    for slug in sorted(os.listdir(SKILLS)):
        mp = os.path.join(SKILLS, slug, "manifest.json")
        if not os.path.isfile(mp):
            continue
        m = json.load(open(mp, encoding="utf-8"))
        soc0 = m.get("ids", {}).get("onet_soc")
        if soc0 and soc0 != "None":
            continue

        isco = str(m.get("ids", {}).get("isco_group") or "").strip()
        cands = isco_soc.get(isco) or isco_soc3.get(isco[:3]) or isco_soc2.get(isco[:2])
        if not cands:
            unmatched.append(slug)
            continue

        pl, al = esco_alt.get(m["ids"].get("esco_uri"), (m.get("title", slug), ""))
        occ_tok = tokens(pl) | tokens(al)
        best = max(cands, key=lambda s: (len(occ_tok & tokens(soc_title.get(s, ""))),
                                         cands[s]))
        mapped += 1
        if dry:
            continue

        soc, st = best, soc_title[best]
        label = m.get("title", slug.replace("-", " "))
        note = (f"Source: O*NET 30.3, occupation {soc} ({st}) — manual nearest-"
                f"occupation mapping via ISCO group {isco}; the official ESCO "
                f"crosswalk has no entry for this ESCO occupation.")
        pkg = os.path.join(SKILLS, slug)

        cur.execute("SELECT task, task_type FROM onet_task WHERE soc_code=? ORDER BY task_id", soc)
        tasks = cur.fetchall()
        cur.execute("""SELECT DISTINCT d.dwa_name FROM onet_task_dwa td
                       JOIN onet_dwa d ON d.dwa_id=td.dwa_id
                       WHERE td.soc_code=? ORDER BY d.dwa_name""", soc)
        dwas = [r[0] for r in cur.fetchall()]
        cur.execute("""SELECT example, element_name, hot_technology FROM onet_software
                       WHERE soc_code=? ORDER BY hot_technology DESC, example""", soc)
        software = cur.fetchall()

        tl = [f"# Tasks & work activities — {label}", "", note, "",
              "## Task statements", ""]
        tl += [f"- **[{t.task_type or 'n/a'}]** {t.task}" for t in tasks]
        if dwas:
            tl += ["", "## Detailed work activities", ""] + [f"- {d}" for d in dwas]
        open(os.path.join(pkg, "references", "tasks.md"), "w", encoding="utf-8",
             newline="\n").write("\n".join(tl) + "\n")

        ol = [f"# Tools & technology — {label}", "", note, "",
              "| Software | Category | Hot technology |", "|---|---|---|"]
        ol += [f"| {s.example} | {s.element_name} | {'yes' if s.hot_technology == 'Y' else ''} |"
               for s in software]
        open(os.path.join(pkg, "references", "tools.md"), "w", encoding="utf-8",
             newline="\n").write("\n".join(ol) + "\n")

        m["ids"]["onet_soc"] = soc
        m["ids"]["crosswalk_match"] = f"manual nearest via ISCO {isco} ({st})"
        json.dump(m, open(mp, "w", encoding="utf-8"), indent=2)

        # SKILL.md: footer + Hot technologies
        sp = os.path.join(pkg, "SKILL.md")
        if os.path.isfile(sp):
            text = open(sp, encoding="utf-8").read()
            text = text.replace("O*NET 30.3 (None)",
                                f"O*NET 30.3 ({soc}, manual nearest match)")
            if "## Hot technologies" not in text:
                hot = [s.example for s in software if s.hot_technology == "Y"][:10]
                if hot:
                    block = ("## Hot technologies\n\n"
                             + "\n".join(f"- {h}" for h in hot) + "\n\n")
                    text = re.sub(r"(?m)^---\n\*Sources:", block + "---\n*Sources:",
                                  text, count=1)
            open(sp, "w", encoding="utf-8", newline="\n").write(text)

        if slug in ev_done:
            try:
                p3c.aggregate_for_occupation(cn, slug, SKILLS)
            except Exception as exc:
                print(f"WARN market re-insert {slug}: {exc}")

        if mapped % 200 == 0:
            print(f"... {mapped} mapped")

    cn.close()
    print(f"backfill: {mapped} mapped, {len(unmatched)} without any ISCO candidate"
          f"{' (dry run)' if dry else ''}")
    if unmatched:
        print("unmatched:", unmatched[:10], "..." if len(unmatched) > 10 else "")


if __name__ == "__main__":
    main()