skillfactor-pipeline/pipeline/p3d_provenance.py

"""Phase 3d — data provenance per skill package.

Counts the content items of every package by source category and makes the
split visible IN the git structure:
- skills/<slug>/PROVENANCE.md  — mermaid pie chart + table (Gitea renders it)
- manifest.json                — gets a "provenance" block
- prints the global totals (input for the landing-page donut)

Categories:
- esco    : occupation profile + competences (skills.md items, profile)
- onet    : tasks, detailed work activities, software (tasks.md/tools.md)
- jobads  : market-evidence items inside the <!-- market-evidence --> markers
- wiki_ai : AI-curated expert knowledge with cited web sources
            (glossary, literature, usecases, intake, quality, evals)
"""
import json
import os
import re

SKILLS_DIR = os.path.join(os.path.dirname(__file__), "..", "skills")
MARKER = "<!-- market-evidence -->"

LABELS = {
    "esco": "ESCO (occupation & competences)",
    "onet": "O*NET (tasks & tools)",
    "jobads": "Job boards (market evidence)",
    "wiki_ai": "Wikipedia & AI expert curation",
}


def count_items(text: str) -> int:
    """List entries + table rows + numbered items = content items."""
    n = 0
    for line in text.splitlines():
        s = line.strip()
        if s.startswith(("- ", "* ")) or re.match(r"^\d+\.\s", s) or (
            s.startswith("|") and not set(s) <= {"|", "-", " ", ":"}
        ):
            n += 1
    return n


def split_market(text: str):
    """Returns (text_without_market_sections, market_text)."""
    parts = re.split(re.escape(MARKER), text)
    if len(parts) >= 3:
        base = parts[0] + "".join(parts[4::2])  # outside the marker pairs
        market = "".join(parts[1::2])
        return base, market
    return text, ""


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


def package_provenance(base):
    counts = {"esco": 0, "onet": 0, "jobads": 0, "wiki_ai": 0}

    skills_txt, market1 = split_market(read(base, "references/skills.md"))
    tools_txt, market2 = split_market(read(base, "references/tools.md"))
    counts["esco"] += count_items(skills_txt) + 1  # +1 occupation profile
    counts["onet"] += count_items(tools_txt)
    counts["onet"] += count_items(read(base, "references/tasks.md"))
    counts["jobads"] += count_items(market1) + count_items(market2)
    counts["jobads"] += count_items(read(base, "references/market.md"))

    for rel in ("references/glossary.md", "references/literature.md",
                "references/usecases.md", "references/intake.md",
                "references/quality.md"):
        txt = read(base, rel)
        # usecases/glossary/literature zaehlen auch ihre H2-Eintraege
        counts["wiki_ai"] += count_items(txt) + len(re.findall(r"(?m)^## ", txt))
    evals = os.path.join(base, "evals")
    if os.path.isdir(evals):
        counts["wiki_ai"] += len([f for f in os.listdir(evals) if f.endswith(".md")]) * 2

    return counts


def write_provenance_md(base, slug, counts):
    total = sum(counts.values()) or 1
    pct = {k: round(100.0 * v / total, 1) for k, v in counts.items()}
    lines = [
        f"# Data provenance — {slug}",
        "",
        "Where the content of this skill package comes from, counted by",
        "content items (tasks, competences, tools, evidence entries, curated",
        "knowledge). Rendered live by Gitea:",
        "",
        "```mermaid",
        "pie showData",
        f'    title Content sources — {slug}',
    ]
    for k in ("esco", "onet", "jobads", "wiki_ai"):
        if counts[k]:
            lines.append(f'    "{LABELS[k]}" : {counts[k]}')
    lines += [
        "```",
        "",
        "| Source | Items | Share | Files |",
        "|---|---|---|---|",
        f"| {LABELS['esco']} | {counts['esco']} | {pct['esco']} % | references/profile.md, references/skills.md |",
        f"| {LABELS['onet']} | {counts['onet']} | {pct['onet']} % | references/tasks.md, references/tools.md |",
        f"| {LABELS['jobads']} | {counts['jobads']} | {pct['jobads']} % | references/market.md (full report) + \"Market evidence\" headline sections |",
        f"| {LABELS['wiki_ai']} | {counts['wiki_ai']} | {pct['wiki_ai']} % | glossary, literature, usecases, intake, quality, evals/ |",
        "",
        "Licensing: O*NET (USDOL/ETA, CC BY 4.0) · ESCO (© European Union) ·",
        "job-ad evidence via official APIs (JSearch/Adzuna) · Wikipedia content",
        "paraphrased with source URLs — never copied.",
        "",
    ]
    with open(os.path.join(base, "PROVENANCE.md"), "w", encoding="utf-8", newline="\n") as f:
        f.write("\n".join(lines))
    return pct


def main():
    total = {"esco": 0, "onet": 0, "jobads": 0, "wiki_ai": 0}
    slugs = sorted(d for d in os.listdir(SKILLS_DIR)
                   if os.path.isdir(os.path.join(SKILLS_DIR, d)))
    for slug in slugs:
        base = os.path.join(SKILLS_DIR, slug)
        counts = package_provenance(base)
        pct = write_provenance_md(base, slug, counts)
        mpath = os.path.join(base, "manifest.json")
        m = json.load(open(mpath, encoding="utf-8"))
        m["provenance"] = {"items": counts, "share_percent": pct,
                           "method": "content items per source category"}
        json.dump(m, open(mpath, "w", encoding="utf-8"), indent=2)
        for k in total:
            total[k] += counts[k]

    print(f"PROVENANCE.md written for {len(slugs)} packages")
    grand = sum(total.values())
    print("GLOBAL totals (landing donut input):")
    for k in ("esco", "onet", "jobads", "wiki_ai"):
        print(f"  {k}: {total[k]} ({round(100.0*total[k]/grand, 1)} %)")


if __name__ == "__main__":
    main()