a2f17bdc12
Compile complete (10 professions x 150 attributed entries). The lint gate correctly blocked the v1-grade carrier packages (truncation defect); regen_skillmd.py brings them to reference SKILL.md quality (full competences, clean intro, market sections preserved) - all 9 published through the gate. Global stackx provenance: 1,350 items, measured. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PDKeXvpT6tENSvyQGLV1Uq
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
"""Regenerate SKILL.md (+ taxonomy references) for given packages with the
|
|
hardened p2 generator — full competences, clean intro, overlays — then
|
|
re-insert market sections, refresh enrichment links, provenance and stats,
|
|
and publish through the lint gate.
|
|
|
|
Used to bring the practitioner-QA carrier packages to lint-clean without a
|
|
full evidence regeneration (their ad corpora arrive with the ongoing crawl).
|
|
|
|
Usage: python pipeline/regen_skillmd.py <slug> [<slug> ...]
|
|
"""
|
|
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 p2_generate # noqa: E402
|
|
import p3c_aggregate as p3c # noqa: E402
|
|
from db import connect # noqa: E402
|
|
|
|
PY = sys.executable
|
|
|
|
|
|
def regen(cn, slug):
|
|
mp = os.path.join(BASE, "skills", slug, "manifest.json")
|
|
m = json.load(open(mp, encoding="utf-8"))
|
|
cur = cn.cursor()
|
|
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"])
|
|
rows = cur.fetchall()
|
|
r = dict(zip([d[0] for d in cur.description], rows[0]))
|
|
cur.close()
|
|
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)
|
|
# restore post-generation manifest fields
|
|
m2 = json.load(open(mp, encoding="utf-8"))
|
|
for k in ("collar", "computer_work"):
|
|
if k in m:
|
|
m2[k] = m[k]
|
|
if m["ids"].get("onet_soc"):
|
|
m2["ids"]["onet_soc"] = m["ids"]["onet_soc"]
|
|
if m["ids"].get("crosswalk_match"):
|
|
m2["ids"]["crosswalk_match"] = m["ids"]["crosswalk_match"]
|
|
json.dump(m2, open(mp, "w", encoding="utf-8"), indent=2)
|
|
# market/hot-tech sections (no-op when the slug has no evidence yet)
|
|
p3c.aggregate_for_occupation(cn, slug, p2_generate.SKILLS_DIR)
|
|
|
|
|
|
def main():
|
|
slugs = sys.argv[1:]
|
|
if not slugs:
|
|
sys.exit("slugs required")
|
|
cn = connect()
|
|
for slug in slugs:
|
|
regen(cn, slug)
|
|
subprocess.run([PY, os.path.join(BASE, "pipeline",
|
|
"p5_enrich_ai_skills.py"), slug],
|
|
cwd=BASE)
|
|
print(f"regenerated: {slug}")
|
|
cn.close()
|
|
subprocess.run([PY, os.path.join(BASE, "pipeline", "p3d_provenance.py")],
|
|
cwd=BASE)
|
|
subprocess.run([PY, os.path.join(BASE, "pipeline", "gen_stats.py")],
|
|
cwd=BASE)
|
|
for slug in slugs:
|
|
subprocess.run([PY, os.path.join(BASE, "pipeline", "p4_publish.py"),
|
|
slug], cwd=BASE)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|