5dfef33f0e
- intro: cached gemma3 rewrite to one clean paragraph (raw ESCO stays the provenance reference); overlays/<slug>/intro.md wins when present - competences: full essential list, the silent essential[:15] alphabetical cut is gone - hot technologies: derived by p3c from post-gate market ranking (omitted below 30 gated ads), no longer the alphabetical O*NET dump - standardized O*NET proxy caveat for non-exact crosswalk matches - overlays/<slug>/workflow.md preserves curated core workflows across regeneration (flagship overlay included) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PDKeXvpT6tENSvyQGLV1Uq
294 lines
13 KiB
Python
294 lines
13 KiB
Python
"""Phase 3c — aggregate job-ad evidence into market percentages.
|
|
|
|
Public API:
|
|
aggregate_for_occupation(cn, slug, skills_dir) -- batch runner entry point
|
|
main() -- recruiter standalone
|
|
"""
|
|
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 -->"
|
|
REPORT_MIN_JOBS = 3
|
|
|
|
# Phase 2b: regional reporting with minimum-corpus thresholds — below the
|
|
# minimum a region reports "insufficient evidence" instead of ranking noise.
|
|
REGIONS = {"US": ("us",), "UK": ("gb",), "EU/DACH": ("de", "at", "ch", "nl")}
|
|
REGION_MIN_ADS = 30
|
|
CORPUS_MIN_ADS = 100
|
|
|
|
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",
|
|
}
|
|
|
|
|
|
def _market_block(groups, total, stamp):
|
|
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)
|
|
return "\n".join(block)
|
|
|
|
|
|
def _write_market_report(cur, slug, total, skills_dir):
|
|
"""Write references/market.md for one occupation."""
|
|
stamp = date.today().isoformat()
|
|
by_country = cur.execute(
|
|
"SELECT country, COUNT(*) FROM evidence_job WHERE occupation_slug=? "
|
|
"GROUP BY country ORDER BY COUNT(*) DESC", slug).fetchall()
|
|
by_seniority = cur.execute(
|
|
"SELECT ISNULL(seniority,'n/a'), COUNT(*) FROM evidence_job "
|
|
"WHERE occupation_slug=? GROUP BY seniority ORDER BY COUNT(*) DESC", slug).fetchall()
|
|
titles = cur.execute(
|
|
"SELECT TOP 25 title, COUNT(*) FROM evidence_job WHERE occupation_slug=? "
|
|
"GROUP BY title ORDER BY COUNT(*) DESC, title", slug).fetchall()
|
|
|
|
lines = [
|
|
f"# Market evidence report — {slug}", "",
|
|
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 ee.entity, COUNT(DISTINCT ee.job_id) AS jobs
|
|
FROM evidence_entity ee
|
|
JOIN evidence_job ej ON ej.job_id=ee.job_id AND ej.occupation_slug=?
|
|
WHERE ee.entity_type=?
|
|
GROUP BY ee.entity HAVING COUNT(DISTINCT ee.job_id) >= ?
|
|
ORDER BY COUNT(DISTINCT ee.job_id) DESC, ee.entity
|
|
""", slug, 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)} % |")
|
|
|
|
# ── Regional breakdown (phase 2b) ────────────────────────────────────
|
|
lines += ["", "## Regional breakdown", ""]
|
|
if total < CORPUS_MIN_ADS:
|
|
lines += [f"> **Corpus note:** {total} relevant ads in total — below the "
|
|
f"{CORPUS_MIN_ADS}-ad target for a fully reliable ranking. "
|
|
"Percentages above should be read as indicative.", ""]
|
|
country_counts = dict((c or "", n) for c, n in by_country)
|
|
for region, ccs in REGIONS.items():
|
|
n_region = sum(country_counts.get(c, 0) for c in ccs)
|
|
lines += [f"### {region} ({', '.join(ccs)})", ""]
|
|
if n_region < REGION_MIN_ADS:
|
|
lines += [f"**Insufficient evidence** — {n_region} ads "
|
|
f"(minimum for a regional ranking: {REGION_MIN_ADS}). "
|
|
"No ranking is reported for this region.", ""]
|
|
continue
|
|
placeholders = ",".join("?" for _ in ccs)
|
|
lines += [f"{n_region} ads.", ""]
|
|
for etype in ("hard_skills", "tools"):
|
|
rows = cur.execute(f"""
|
|
SELECT TOP 10 ee.entity, COUNT(DISTINCT ee.job_id) AS jobs
|
|
FROM evidence_entity ee
|
|
JOIN evidence_job ej ON ej.job_id=ee.job_id
|
|
WHERE ej.occupation_slug=? AND ej.country IN ({placeholders})
|
|
AND ee.entity_type=?
|
|
GROUP BY ee.entity HAVING COUNT(DISTINCT ee.job_id) >= 2
|
|
ORDER BY COUNT(DISTINCT ee.job_id) DESC, ee.entity
|
|
""", slug, *ccs, etype).fetchall()
|
|
if rows:
|
|
lines += [f"**Top {TITLES[etype].lower()}:**", ""]
|
|
lines += [f"- {e} — {round(100.0*n/n_region)} % ({n} ads)"
|
|
for e, n in rows]
|
|
lines.append("")
|
|
sen = cur.execute(f"""
|
|
SELECT ISNULL(seniority,'n/a'), COUNT(*) FROM evidence_job
|
|
WHERE occupation_slug=? AND country IN ({placeholders})
|
|
GROUP BY seniority ORDER BY COUNT(*) DESC
|
|
""", slug, *ccs).fetchall()
|
|
if sen:
|
|
lines += ["**Seniority:** "
|
|
+ " · ".join(f"{s} {round(100.0*n/n_region)} %" for s, n in sen), ""]
|
|
|
|
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.", "",
|
|
]
|
|
|
|
refs_dir = os.path.join(skills_dir, slug, "references")
|
|
os.makedirs(refs_dir, exist_ok=True)
|
|
path = os.path.join(refs_dir, "market.md")
|
|
open(path, "w", encoding="utf-8", newline="\n").write("\n".join(lines))
|
|
return path
|
|
|
|
|
|
def aggregate_for_occupation(cn, slug, skills_dir, max_retries=3):
|
|
"""Aggregate evidence for one occupation: write market.md + update tools.md/skills.md.
|
|
|
|
Returns total ad count processed, or 0 if no evidence found.
|
|
Retries on SQL Server deadlock (error 1205) up to max_retries times.
|
|
Called from batch_run.py stage_evidence.
|
|
"""
|
|
import time as _time
|
|
import pyodbc as _pyodbc
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
cur = cn.cursor()
|
|
total = cur.execute(
|
|
"SELECT COUNT(*) FROM evidence_job WHERE occupation_slug=?", slug
|
|
).fetchone()[0]
|
|
if not total:
|
|
return 0
|
|
|
|
rows = cur.execute("""
|
|
SELECT ee.entity_type, ee.entity, COUNT(DISTINCT ee.job_id) AS jobs
|
|
FROM evidence_entity ee
|
|
JOIN evidence_job ej ON ej.job_id=ee.job_id AND ej.occupation_slug=?
|
|
GROUP BY ee.entity_type, ee.entity
|
|
HAVING COUNT(DISTINCT ee.job_id) >= ?
|
|
ORDER BY ee.entity_type, COUNT(DISTINCT ee.job_id) DESC
|
|
""", slug, max(1, int(total * THRESHOLD))).fetchall()
|
|
|
|
_write_market_report(cur, slug, total, skills_dir)
|
|
|
|
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))
|
|
|
|
# Phase 2d: "Hot technologies" in SKILL.md — from the post-gate
|
|
# market ranking, never the alphabetical O*NET dump. Omitted when
|
|
# the corpus is too thin for a ranking.
|
|
HOT_MARKER = "<!-- hot-tech -->"
|
|
skill_path = os.path.join(skills_dir, slug, "SKILL.md")
|
|
if os.path.exists(skill_path):
|
|
stext = open(skill_path, encoding="utf-8").read()
|
|
stext = re.sub(rf"\n?{HOT_MARKER}.*?{HOT_MARKER}\n?", "",
|
|
stext, flags=re.S)
|
|
if total >= 30:
|
|
hot_rows = cur.execute("""
|
|
SELECT TOP 10 ee.entity, COUNT(DISTINCT ee.job_id) AS jobs
|
|
FROM evidence_entity ee
|
|
JOIN evidence_job ej ON ej.job_id=ee.job_id
|
|
WHERE ej.occupation_slug=? AND ee.entity_type='tools'
|
|
GROUP BY ee.entity HAVING COUNT(DISTINCT ee.job_id) >= ?
|
|
ORDER BY COUNT(DISTINCT ee.job_id) DESC, ee.entity
|
|
""", slug, REPORT_MIN_JOBS).fetchall()
|
|
if hot_rows:
|
|
block = [f"\n{HOT_MARKER}", "",
|
|
"## Hot technologies",
|
|
"",
|
|
f"Top tools from {total} gated job ads "
|
|
f"(see references/market.md, as of "
|
|
f"{date.today().isoformat()}):", ""]
|
|
block += [f"- {e} — {round(100.0*n/total)} %"
|
|
for e, n in hot_rows]
|
|
block += ["", HOT_MARKER]
|
|
stext = re.sub(r"(?m)^---\n\*Sources:",
|
|
"\n".join(block) + "\n\n---\n*Sources:",
|
|
stext, count=1)
|
|
open(skill_path, "w", encoding="utf-8", newline="\n").write(stext)
|
|
|
|
stamp = date.today().isoformat()
|
|
for fname, groups in by_file.items():
|
|
path = os.path.join(skills_dir, slug, "references", fname)
|
|
if not os.path.exists(path):
|
|
continue
|
|
text = open(path, encoding="utf-8").read()
|
|
text = re.sub(rf"\n?{MARKER}.*?{MARKER}\n?", "", text, flags=re.S)
|
|
block = _market_block(groups, total, stamp)
|
|
open(path, "w", encoding="utf-8", newline="\n").write(
|
|
text.rstrip() + "\n" + block + "\n")
|
|
|
|
return total
|
|
|
|
except _pyodbc.Error as exc:
|
|
# SQL Server error 1205 = deadlock victim — retry after backoff
|
|
if "1205" in str(exc) and attempt < max_retries - 1:
|
|
_time.sleep(5 * (attempt + 1))
|
|
continue
|
|
raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Legacy entry point — recruiter-only standalone run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def build_market_report(cur, recruiter, total):
|
|
"""Kept for standalone use. Delegates to _write_market_report."""
|
|
_write_market_report(cur, recruiter, SKILLS_DIR)
|
|
n_items = 0
|
|
for etype in ("tools", "hard_skills", "methods", "responsibilities"):
|
|
n_items += cur.execute("""
|
|
SELECT COUNT(*) FROM (
|
|
SELECT ee.entity
|
|
FROM evidence_entity ee
|
|
JOIN evidence_job ej ON ej.job_id=ee.job_id AND ej.occupation_slug=?
|
|
WHERE ee.entity_type=?
|
|
GROUP BY ee.entity HAVING COUNT(DISTINCT ee.job_id) >= ?
|
|
) t
|
|
""", recruiter, etype, REPORT_MIN_JOBS).fetchone()[0]
|
|
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 WHERE occupation_slug=?", recruiter
|
|
).fetchone()[0]
|
|
if not total:
|
|
# fallback: count all if column missing or no slug set
|
|
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
|
|
|
|
n = aggregate_for_occupation(cn, recruiter, SKILLS_DIR)
|
|
cn.close()
|
|
print(f"market evidence written for {recruiter} ({n} ads)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|