"""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()