skillfactor-pipeline/pipeline/regen_flagship.py

"""Phase 4 — regenerate one package at reference quality, end to end.

Steps (all from stored raw data + the new EU slice; no US re-fetching):
  1. gate     all raw ads (US re-classified + EU) via ad_gate
  2. extract  gate-relevant ads that lack extraction output (Ollama)
  3. store    gated-relevant only (p3b delete+insert)
  4. compile  market.md with regional sections + SKILL.md hot-tech (p3c)
  5. SKILL.md regeneration (p2: clean intro, full competences, overlays)
  6. ai-skills tiered enrichment (p5)
  7. provenance + stats (p3d slug-scoped rebuild via full run, gen_stats)
  8. lint     must be CLEAN (exit 1 otherwise)

Usage:  python pipeline/regen_flagship.py [slug]   (default: AI engineer)
"""
import json
import os
import subprocess
import sys

sys.path.insert(0, os.path.dirname(__file__))
BASE = os.path.join(os.path.dirname(__file__), "..")

import ad_gate  # noqa: E402
import p2_generate  # noqa: E402
import p3b_store_evidence as p3b  # noqa: E402
import p3c_aggregate as p3c  # noqa: E402
from db import connect  # noqa: E402

SLUG = sys.argv[1] if len(sys.argv) > 1 else "artificial-intelligence-engineer"
PY = sys.executable


def sh(args):
    print(f"--> {' '.join(args[1:])}")
    r = subprocess.run(args, cwd=BASE)
    if r.returncode not in (0, 1):  # extract_local exit 1 = >10% fail, usable
        sys.exit(f"step failed: {args} -> {r.returncode}")


def main():
    ads_file = os.path.join(BASE, "data", "raw", "jobs", f"{SLUG}_ads.json")
    ads = json.load(open(ads_file, encoding="utf-8"))
    print(f"raw corpus: {len(ads)} ads")

    # 1. gate (resumable; covers the EU additions)
    relevant, verdicts = ad_gate.gate_ads(SLUG, ads)
    gated_file = os.path.join(BASE, "data", "gated", f"{SLUG}_relevant.json")
    json.dump(relevant, open(gated_file, "w", encoding="utf-8"),
              ensure_ascii=False, indent=1)
    print(f"gate: {len(relevant)}/{len(ads)} relevant")

    # 2. extract missing (resumable via jsonl done-set)
    ext_file = os.path.join(BASE, "data", "evidence", f"{SLUG}.jsonl")
    sh([PY, os.path.join("pipeline", "extract_local.py"),
        "--ads", gated_file, "--out", ext_file])

    # 3+4. store gated-relevant + aggregate (regional market.md, hot-tech)
    extractions = [json.loads(l) for l in open(ext_file, encoding="utf-8")
                   if l.strip()]
    ads_dict = {a["job_id"]: a for a in relevant}
    cn = connect()
    n_jobs, n_ent = p3b.store_for_occupation(cn, ads_dict, extractions, SLUG)
    print(f"store: {n_jobs} jobs, {n_ent} entities (gated only)")
    p3c.aggregate_for_occupation(cn, SLUG, p2_generate.SKILLS_DIR)
    print("aggregate: market.md + hot-tech done")

    # 5. SKILL.md regeneration (clean intro, full competences, overlays)
    cur = cn.cursor()
    m = json.load(open(os.path.join(BASE, "skills", SLUG, "manifest.json"),
                       encoding="utf-8"))
    cur.execute("""SELECT e.concept_uri, e.code, e.preferred_label, e.alt_labels,
                          e.description, e.definition, e.isco_group,
                          c.onet_id, c.onet_title, c.match_type
                   FROM esco_occupation e
                   LEFT JOIN crosswalk_esco_onet c ON c.esco_uri = e.concept_uri
                   WHERE e.concept_uri = ?""", m["ids"]["esco_uri"])
    row = cur.fetchone()
    cols = [d[0] for d in cur.description]
    r = dict(zip(cols, row))
    # manual O*NET backfill (p7) survives regeneration
    if not r.get("onet_id") and m["ids"].get("onet_soc"):
        r["onet_id"] = m["ids"]["onet_soc"]
        r["onet_title"] = m["ids"].get("crosswalk_match", "manual")
        r["match_type"] = "manual nearest"
    r.setdefault("isco_label", None)
    p2_generate.gen_package(cn, r)
    print("p2: SKILL.md regenerated (overlays applied)")
    # p2 rewrites the manifest — restore fields later stages depend on
    m2 = json.load(open(os.path.join(BASE, "skills", SLUG, "manifest.json"),
                        encoding="utf-8"))
    for k in ("collar", "computer_work"):
        if k in m:
            m2[k] = m[k]
    m2["ids"]["onet_soc"] = m["ids"].get("onet_soc") or m2["ids"].get("onet_soc")
    if m["ids"].get("crosswalk_match"):
        m2["ids"]["crosswalk_match"] = m["ids"]["crosswalk_match"]
    json.dump(m2, open(os.path.join(BASE, "skills", SLUG, "manifest.json"),
                       "w", encoding="utf-8"), indent=2)
    # re-insert market sections lost by tools/skills regeneration
    p3c.aggregate_for_occupation(cn, SLUG, p2_generate.SKILLS_DIR)
    cn.close()

    # 6-7. tiered enrichment, provenance, stats
    sh([PY, os.path.join("pipeline", "p5_enrich_ai_skills.py"), SLUG])
    sh([PY, os.path.join("pipeline", "p3d_provenance.py")])
    sh([PY, os.path.join("pipeline", "gen_stats.py")])

    # 8. lint gate
    r2 = subprocess.run([PY, os.path.join("qa", "lint_package.py"), SLUG],
                        cwd=BASE)
    if r2.returncode != 0:
        sys.exit("LINT FAILED — package is not reference quality yet")
    print("LINT CLEAN — reference quality reached")


if __name__ == "__main__":
    main()