skillfactor-pipeline/pipeline/p3c_aggregate.py

"""Phase 3c — aggregate job-ad evidence into market percentages and write them
into the recruiter package (references/tools.md + skills.md get a
"Market evidence" section; threshold >= 20 %, percentages + as-of date shown).

Idempotent: the market sections are fully replaced on each run.
"""
import os
import re
import sys
from datetime import date

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

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

SECTION_FOR = {
    "tools": "tools.md",
    "hard_skills": "skills.md",
    "methods": "skills.md",
    "responsibilities": "skills.md",
}
TITLES = {
    "tools": "Tools", "hard_skills": "Hard skills",
    "methods": "Methods", "responsibilities": "Responsibilities",
}


REPORT_MIN_JOBS = 3  # Vollreport: alles ab 3 Anzeigen (~3 %) — DAS ist der Markt-Juice


def build_market_report(cur, recruiter, total):
    """references/market.md — der vollständige Marktreport aus dem Evidence Store.
    Nur extrahierte, aggregierte Fakten (keine Anzeigen-Texte: Urheberrecht/ToS)."""
    stamp = date.today().isoformat()
    by_country = cur.execute(
        "SELECT country, COUNT(*) FROM evidence_job GROUP BY country ORDER BY COUNT(*) DESC").fetchall()
    by_seniority = cur.execute(
        "SELECT ISNULL(seniority,'n/a'), COUNT(*) FROM evidence_job GROUP BY seniority ORDER BY COUNT(*) DESC").fetchall()
    titles = cur.execute(
        "SELECT TOP 25 title, COUNT(*) FROM evidence_job GROUP BY title ORDER BY COUNT(*) DESC, title").fetchall()
    lines = [
        f"# Market evidence report — {recruiter}",
        "",
        f"Source: **{total} real job ads** (JSearch API, countries: "
        + ", ".join(f"{c or 'n/a'} {n}" for c, n in by_country)
        + f"), extracted into the MSSQL evidence store; as of {stamp}.",
        "This report contains extracted, aggregated facts only — no ad text is",
        "reproduced (copyright / platform terms).",
        "",
        "## Seniority distribution",
        "",
        "| Seniority | Ads | Share |",
        "|---|---|---|",
    ]
    for s, n in by_seniority:
        lines.append(f"| {s} | {n} | {round(100.0*n/total)} % |")
    for etype in ("tools", "hard_skills", "methods", "responsibilities"):
        rows = cur.execute("""
            SELECT entity, COUNT(DISTINCT job_id) AS jobs
            FROM evidence_entity WHERE entity_type = ?
            GROUP BY entity HAVING COUNT(DISTINCT job_id) >= ?
            ORDER BY jobs DESC, entity
        """, etype, REPORT_MIN_JOBS).fetchall()
        lines += ["", f"## {TITLES[etype]} — full market ranking", "",
                  "| # | Item | Ads | Share |", "|---|---|---|---|"]
        for i, (entity, jobs) in enumerate(rows, 1):
            lines.append(f"| {i} | {entity} | {jobs} | {round(100.0*jobs/total)} % |")
    lines += ["", "## Job title variants in the market", "",
              "| Title | Ads |", "|---|---|"]
    for t, n in titles:
        lines.append(f"| {t} | {n} |")
    lines += ["", f"Methodology: entities extracted per ad "
              f"({{hard_skills, tools, methods, responsibilities, seniority}}), "
              f"normalized, counted as DISTINCT ads per entity; report threshold "
              f"≥ {REPORT_MIN_JOBS} ads. Headline sections in skills.md/tools.md "
              "use the stricter ≥ 20 % threshold.", ""]
    path = os.path.join(SKILLS_DIR, recruiter, "references", "market.md")
    open(path, "w", encoding="utf-8", newline="\n").write("\n".join(lines))
    n_items = sum(1 for l in lines if l.startswith("| ") and not l.startswith("| #") and "---" not in l)
    print(f"market.md written ({n_items} evidence rows)")


def main():
    recruiter = next((d for d in sorted(os.listdir(SKILLS_DIR)) if "recruit" in d), None)
    if not recruiter:
        sys.exit("no recruiter package — run p2 first")
    cn = connect()
    cur = cn.cursor()
    try:
        total = cur.execute("SELECT COUNT(*) FROM evidence_job").fetchone()[0]
    except Exception:
        print("SKIP: no evidence tables yet (p3b not run) — TODO in README")
        return
    if not total:
        print("SKIP: evidence store empty")
        return

    rows = cur.execute("""
        SELECT entity_type, entity, COUNT(DISTINCT job_id) AS jobs
        FROM evidence_entity
        GROUP BY entity_type, entity
        HAVING COUNT(DISTINCT job_id) >= ?
        ORDER BY entity_type, jobs DESC
    """, max(1, int(total * THRESHOLD))).fetchall()
    build_market_report(cur, recruiter, total)
    cn.close()

    by_file = {}
    for etype, entity, jobs in rows:
        pct = round(100.0 * jobs / total)
        by_file.setdefault(SECTION_FOR[etype], {}).setdefault(etype, []).append((entity, pct))

    stamp = date.today().isoformat()
    for fname, groups in by_file.items():
        path = os.path.join(SKILLS_DIR, recruiter, "references", fname)
        text = open(path, encoding="utf-8").read()
        text = re.sub(rf"\n?{MARKER}.*?{MARKER}\n?", "", text, flags=re.S)
        block = [f"\n{MARKER}", "",
                 f"## Market evidence (job-ad analysis, {total} ads, as of {stamp})",
                 "",
                 f"Share of analyzed job ads mentioning the item "
                 f"(threshold ≥ {int(THRESHOLD*100)} %). Source: JSearch/Adzuna APIs.", ""]
        for etype in ("tools", "hard_skills", "methods", "responsibilities"):
            if etype in groups:
                block.append(f"### {TITLES[etype]}")
                block.append("")
                for entity, pct in groups[etype]:
                    block.append(f"- {entity} — **{pct} %**")
                block.append("")
        block.append(MARKER)
        open(path, "w", encoding="utf-8", newline="\n").write(text.rstrip() + "\n" + "\n".join(block) + "\n")
        print(f"market evidence written into {fname} ({sum(len(v) for v in groups.values())} items)")


if __name__ == "__main__":
    main()