skillfactor-pipeline/pipeline/evidence_report.py

"""Evidence-Auswertung: coverage report over the job-ad evidence stage.

Reads progress.json + the MSSQL evidence store and writes
docs/evidence-report.md — runnable at any time (interim or final state).

Sections:
- coverage: fetched/extracted/aggregated/failed/thin-market per status
- volume: ads, unique employers, entities by type
- budget: JSearch requests spent vs. lifetime cap
- distributions: seniority, country, top occupations by ads
- extraction engines: Claude (in-session, flagship + QA) vs Ollama (mass run)

Usage:  python pipeline/evidence_report.py [--push]
        --push additionally commits the report into skills-core/marketplace
                as EVIDENCE.md (uses p4_publish push helpers).
"""
import json
import os
import sys
from collections import Counter
from datetime import date

sys.path.insert(0, os.path.dirname(__file__))
import progress
from db import connect

BASE = os.path.join(os.path.dirname(__file__), "..")
OUT = os.path.join(BASE, "docs", "evidence-report.md")

# The recruiter flagship (350 ads) was extracted in-session by Claude before
# the local mass pipeline existed; everything after runs on Ollama gemma3:27b.
CLAUDE_EXTRACTED_SLUGS = {"recruitment-consultant"}


def q(cur, sql, *args):
    cur.execute(sql, *args)
    return cur.fetchall()


def main():
    state = progress.load()
    occ = state["occupations"]
    ev = Counter(s.get("evidence") for s in occ.values())
    ads_per_occ = {slug: s.get("ads", 0) for slug, s in occ.items() if s.get("ads")}
    thin = [(slug, s.get("error", "")) for slug, s in occ.items()
            if s.get("evidence") == "failed" and "thin market" in (s.get("error") or "")]
    other_failed = [(slug, s.get("error", "")) for slug, s in occ.items()
                    if s.get("evidence") == "failed" and "thin market" not in (s.get("error") or "")]

    cn = connect()
    cur = cn.cursor()
    n_jobs = q(cur, "SELECT COUNT(*) FROM evidence_job")[0][0]
    n_occ_jobs = q(cur, "SELECT COUNT(DISTINCT occupation_slug) FROM evidence_job")[0][0]
    n_emp = q(cur, "SELECT COUNT(DISTINCT employer) FROM evidence_job WHERE employer IS NOT NULL AND employer<>''")[0][0]
    n_ent = q(cur, "SELECT COUNT(*) FROM evidence_entity")[0][0]
    ent_types = q(cur, "SELECT entity_type, COUNT(*) FROM evidence_entity GROUP BY entity_type ORDER BY 2 DESC")
    seniority = q(cur, "SELECT ISNULL(NULLIF(seniority,''),'n/a'), COUNT(*) FROM evidence_job GROUP BY ISNULL(NULLIF(seniority,''),'n/a') ORDER BY 2 DESC")
    country = q(cur, "SELECT ISNULL(NULLIF(country,''),'?'), COUNT(*) FROM evidence_job GROUP BY ISNULL(NULLIF(country,''),'?') ORDER BY 2 DESC")
    top_occ = q(cur, "SELECT TOP 20 occupation_slug, COUNT(*) FROM evidence_job GROUP BY occupation_slug ORDER BY 2 DESC")
    cn.close()

    claude_jobs = sum(n for s, n in
                      [(slug, ads_per_occ.get(slug, 0)) for slug in CLAUDE_EXTRACTED_SLUGS])
    # recruiter ads live in the DB — count them exactly
    cn = connect(); cur = cn.cursor()
    claude_db = q(cur, "SELECT COUNT(*) FROM evidence_job WHERE occupation_slug IN ('recruitment-consultant')")[0][0]
    cn.close()
    ollama_db = n_jobs - claude_db

    spent = state.get("jsearch_requests_total", 0)
    cap = progress.JSEARCH_BUDGET_TOTAL
    total_occ = len(occ)

    L = []
    L.append(f"# Job-ad evidence — coverage report")
    L.append("")
    L.append(f"_Generated {date.today().isoformat()} by `pipeline/evidence_report.py`. "
             f"Re-run any time for the current state._")
    L.append("")
    L.append("## Coverage")
    L.append("")
    L.append("| Evidence status | Occupations |")
    L.append("|---|---|")
    for k in ("done", "fetching", "extracting", "aggregating", "failed", "pending"):
        if ev.get(k):
            L.append(f"| {k} | {ev[k]} |")
    L.append(f"| **total catalog** | **{total_occ}** |")
    L.append("")
    L.append(f"- Occupations with ads in the evidence store: **{n_occ_jobs}**")
    L.append(f"- Thin markets (fewer than 5 usable ads): **{len(thin)}**")
    L.append(f"- Failed for other reasons: **{len(other_failed)}**")
    L.append("")
    L.append("## Volume")
    L.append("")
    L.append(f"| Metric | Value |")
    L.append(f"|---|---|")
    L.append(f"| Job ads stored | {n_jobs:,} |")
    L.append(f"| Distinct employers | {n_emp:,} |")
    L.append(f"| Extracted entities | {n_ent:,} |")
    for t, n in ent_types:
        L.append(f"| — {t} | {n:,} |")
    L.append("")
    L.append("## API budget")
    L.append("")
    L.append(f"- JSearch requests spent: **{spent:,} / {cap:,}** "
             f"({round(100.0*spent/cap, 1)} % of the lifetime budget)")
    L.append("- Every request is counted BEFORE the HTTP call "
             "(`progress.spend_request`, hard cap).")
    L.append("")
    L.append("## Distributions")
    L.append("")
    L.append("**Seniority** (per ad):")
    L.append("")
    L.append("| Seniority | Ads |")
    L.append("|---|---|")
    for s, n in seniority:
        L.append(f"| {s} | {n:,} |")
    L.append("")
    L.append("**Country**:")
    L.append("")
    L.append("| Country | Ads |")
    L.append("|---|---|")
    for c, n in country:
        L.append(f"| {c} | {n:,} |")
    L.append("")
    L.append("**Top 20 occupations by ads:**")
    L.append("")
    L.append("| Occupation | Ads |")
    L.append("|---|---|")
    for slug, n in top_occ:
        L.append(f"| {slug} | {n:,} |")
    L.append("")
    L.append("## Extraction engines")
    L.append("")
    L.append("| Engine | Ads extracted | Share | Role |")
    L.append("|---|---|---|---|")
    tot = max(1, n_jobs)
    L.append(f"| Claude (in-session) | {claude_db:,} | {round(100.0*claude_db/tot,1)} % | "
             f"flagship package (recruitment-consultant), prompt/validator design, "
             f"2 % QA spot checks on every batch |")
    L.append(f"| Ollama gemma3:27b (RTX-3090 box) | {ollama_db:,} | {round(100.0*ollama_db/tot,1)} % | "
             f"mass extraction of the full catalog (`extract_local.py`, "
             f"strict JSON schema, validator with final word) |")
    L.append("")
    L.append("_Thin-market occupations (kept for transparency):_")
    L.append("")
    for slug, err in sorted(thin)[:50]:
        L.append(f"- {slug} — {err}")
    if len(thin) > 50:
        L.append(f"- … and {len(thin)-50} more")
    L.append("")

    os.makedirs(os.path.dirname(OUT), exist_ok=True)
    with open(OUT, "w", encoding="utf-8", newline="\n") as f:
        f.write("\n".join(L))
    print(f"report written: {OUT}")
    print(f"coverage: {n_occ_jobs}/{total_occ} occupations, {n_jobs} ads, "
          f"{n_ent} entities, budget {spent}/{cap}")

    if "--push" in sys.argv:
        import shutil, tempfile
        import p4_publish as p4
        tmp = tempfile.mkdtemp(prefix="sfev_")
        try:
            # marketplace repo: add/refresh EVIDENCE.md next to README
            import subprocess
            url = p4.GITEA_URL.replace("://", f"://gitadmin:{p4.TOKEN}@") + "/skills-core/marketplace.git"
            subprocess.run(["git", "clone", "--depth", "1", url, tmp],
                           check=True, capture_output=True, text=True)
            shutil.copy(OUT, os.path.join(tmp, "EVIDENCE.md"))
            def g(*a):
                subprocess.run(["git", *a], cwd=tmp, check=True, capture_output=True, text=True)
            g("config", "user.name", "skillfactor-pipeline")
            g("config", "user.email", "pipeline@noreply.zeiterfassung.cloud")
            g("add", "-A")
            r = subprocess.run(["git", "commit", "-m", "docs: evidence coverage report"],
                               cwd=tmp, capture_output=True, text=True)
            if r.returncode == 0:
                g("push", "origin", "HEAD")
                print("EVIDENCE.md pushed to skills-core/marketplace")
            else:
                print("no changes to push")
        finally:
            shutil.rmtree(tmp, ignore_errors=True)


if __name__ == "__main__":
    main()