skillfactor-pipeline/qa/lint_package.py

"""qa/lint_package.py — publish gate for skill packages (phase 3).

Checks (each returns a list of findings; any finding = lint failure):
  C1 contamination   designer/marketing tool names in the top-10 market
                     rankings of an engineering/ICT package
  C2 truncation      long bullet list that is exactly alphabetically sorted
                     and ends mid-alphabet (the silent essential[:15] defect)
  C3 counts          manifest.enrichment_ai_skills.total_skills must equal
                     the ai-skills.md table rows; stats.json flagship count
                     must match the manifest
  C4 provenance      every ai-skills.md source section needs repo, commit,
                     license, retrieval date; manifest needs esco_uri;
                     PROVENANCE.md must exist
  C5 grammar         singular-noun + base-verb agreement error in the intro
                     ("engineer apply ...")

Usage:
  python qa/lint_package.py <slug>            lint one package (exit 1 on fail)
  python qa/lint_package.py --all [--json F]  lint the catalog, write summary

Publish wiring: p4_publish refuses to push a failing package unless
SKIP_LINT=1 is set (used for explicitly-labeled v1-grade republishes).
"""
import argparse
import json
import os
import re
import sys

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

# C1: tools that signal a foreign role family inside an ICT/engineering package
CONTAMINANTS = {
    "figma", "sketch", "adobe creative", "adobe photoshop", "photoshop",
    "illustrator", "indesign", "canva", "adobe xd", "after effects",
    "mailchimp", "hubspot", "hootsuite", "google ads", "meta ads",
    "salesforce marketing", "interaction design", "user research",
    "product design", "visual design", "ux design", "ui design",
}
ENGINEERING_ISCO_PREFIXES = ("21", "25", "35")


def _read(base, rel):
    p = os.path.join(base, rel)
    return open(p, encoding="utf-8").read() if os.path.exists(p) else ""


def check_contamination(base, manifest):
    isco = str(manifest.get("ids", {}).get("isco_group") or "")
    if not isco.startswith(ENGINEERING_ISCO_PREFIXES):
        return []
    market = _read(base, os.path.join("references", "market.md"))
    if not market:
        return []
    findings = []
    for section in ("Tools", "Hard skills"):
        m = re.search(rf"(?ms)^## {section} — full market ranking$(.*?)(?=^## |\Z)",
                      market)
        if not m:
            continue
        rows = re.findall(r"(?m)^\|\s*(\d+)\s*\|\s*([^|]+)\|", m.group(1))
        top10 = [item.strip().lower() for rank, item in rows if int(rank) <= 10]
        for item in top10:
            if any(c in item for c in CONTAMINANTS):
                findings.append(f"C1 contamination: '{item}' in market top-10 "
                                f"({section}) of an engineering package")
    return findings


def check_truncation(base, manifest):
    findings = []
    skill_md = _read(base, "SKILL.md")
    for m in re.finditer(r"(?ms)^## ([^\n]+)$\n\n((?:- [^\n]+\n)+)", skill_md):
        title, block = m.group(1), m.group(2)
        items = [l[2:].strip().lower() for l in block.strip().splitlines()]
        if len(items) < 10:
            continue
        if items == sorted(items) and items[-1][:1] < "n":
            findings.append(f"C2 truncation: '{title}' list is alphabetically "
                            f"sorted and ends at '{items[-1][:30]}' — looks cut")
    return findings


def check_counts(base, manifest, slug):
    findings = []
    enr = manifest.get("enrichment_ai_skills") or {}
    claimed = enr.get("total_skills")
    ai_md = _read(base, os.path.join("references", "ai-skills.md"))
    if claimed is not None and ai_md:
        actual = len(re.findall(r"(?m)^\| `", ai_md))
        if actual != claimed:
            findings.append(f"C3 counts: manifest claims {claimed} ai-skills, "
                            f"ai-skills.md has {actual}")
    stats_p = os.path.join(BASE, "data", "stats.json")
    if os.path.exists(stats_p):
        st = json.load(open(stats_p, encoding="utf-8"))
        fl = st.get("flagship") or {}
        if fl.get("slug") == slug and claimed is not None \
                and fl.get("ai_skills_total") != claimed:
            findings.append(f"C3 counts: stats.json flagship says "
                            f"{fl.get('ai_skills_total')}, manifest {claimed}")
    return findings


def check_provenance(base, manifest):
    findings = []
    if not manifest.get("ids", {}).get("esco_uri"):
        findings.append("C4 provenance: manifest.ids.esco_uri missing")
    if not os.path.exists(os.path.join(base, "PROVENANCE.md")):
        findings.append("C4 provenance: PROVENANCE.md missing")
    ai_md = _read(base, os.path.join("references", "ai-skills.md"))
    if ai_md:
        for m in re.finditer(r"(?ms)^## Source: ([^\n]+)$(.*?)(?=^## |\Z)", ai_md):
            src, body = m.group(1).strip(), m.group(2)
            for field, pat in (("commit", r"commit `?\w+"),
                               ("license", r"License:"),
                               ("retrieval date", r"retrieved \d{4}-\d{2}-\d{2}")):
                if not re.search(pat, body):
                    findings.append(f"C4 provenance: source '{src}' lacks {field}")
    return findings


_SING_NOUNS = ("engineer", "designer", "manager", "technician", "analyst",
               "consultant", "officer", "specialist", "developer", "operator",
               "assistant", "advisor", "worker", "scientist", "teacher")
_BASE_VERBS = ("apply", "design", "create", "provide", "develop", "perform",
               "manage", "use", "work", "integrate", "analyse", "analyze",
               "coordinate", "support", "conduct", "oversee")


def check_grammar(base, manifest):
    skill_md = _read(base, "SKILL.md")
    m = re.search(r"(?ms)^# [^\n]+\n\n(.+?)(?=\n\n## )", skill_md)
    if not m:
        return []
    intro = m.group(1)
    first = intro.split(".")[0].lower()
    for noun in _SING_NOUNS:
        for verb in _BASE_VERBS:
            if re.search(rf"\b{noun} {verb}\b", first):
                return [f"C5 grammar: singular '{noun}' with plural verb "
                        f"'{verb}' in the intro"]
    return []


def lint(slug):
    base = os.path.join(SKILLS, slug)
    mp = os.path.join(base, "manifest.json")
    if not os.path.isfile(mp):
        return [f"C4 provenance: manifest.json missing"]
    try:
        manifest = json.load(open(mp, encoding="utf-8"))
    except ValueError:
        return ["C4 provenance: manifest.json unparseable"]
    findings = []
    findings += check_contamination(base, manifest)
    findings += check_truncation(base, manifest)
    findings += check_counts(base, manifest, slug)
    findings += check_provenance(base, manifest)
    findings += check_grammar(base, manifest)
    return findings


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("slug", nargs="?")
    ap.add_argument("--all", action="store_true")
    ap.add_argument("--json", help="write per-package results to this file")
    args = ap.parse_args()

    if args.slug:
        findings = lint(args.slug)
        for f in findings:
            print(f"FAIL {args.slug}: {f}")
        print(f"{args.slug}: {'CLEAN' if not findings else str(len(findings)) + ' finding(s)'}")
        sys.exit(1 if findings else 0)

    if not args.all:
        ap.error("slug or --all required")

    results = {}
    slugs = sorted(d for d in os.listdir(SKILLS)
                   if os.path.isdir(os.path.join(SKILLS, d)))
    for i, slug in enumerate(slugs, 1):
        results[slug] = lint(slug)
        if i % 500 == 0:
            print(f"... {i}/{len(slugs)}")
    if args.json:
        json.dump(results, open(args.json, "w", encoding="utf-8"), indent=1)
    dirty = {s: f for s, f in results.items() if f}
    by_class = {}
    for fs in results.values():
        for f in fs:
            by_class[f[:2]] = by_class.get(f[:2], 0) + 1
    print(f"\ncatalog: {len(slugs)} packages, {len(slugs) - len(dirty)} clean, "
          f"{len(dirty)} with findings")
    for cls, n in sorted(by_class.items()):
        print(f"  {cls}: {n} findings")
    sys.exit(0)


if __name__ == "__main__":
    main()