"""Ad relevance gate — runs BEFORE extraction (phase 2a of the quality program). Two stages per ad: 1. Deterministic title pre-filter: role-family blocklist (designer, recruiter, sales, marketing, instructor, ...). A family term only blocks if it does NOT occur in the target occupation's own label/alt labels (a designer occupation must keep designer ads). Saves LLM tokens. 2. LLM gate (Ollama gemma3, temperature 0, structured output): judges ad title + first ~500 chars against the ESCO description as ground truth -> {"relevant": bool, "reason": str}. Artifacts (all resumable, raw data never modified): data/gated/.jsonl one verdict per ad: {job_id, relevant, stage, reason, title} data/rejected-ads.jsonl global log of rejections (slug, job_id, title, stage, reason, gated_at) — provenance only, no ad text beyond the title. Library use (batch_run): from ad_gate import gate_ads CLI re-classification: python pipeline/ad_gate.py --slug python pipeline/ad_gate.py --all """ import argparse import json import os import sys import time from datetime import datetime import requests from dotenv import load_dotenv BASE = os.path.join(os.path.dirname(__file__), "..") load_dotenv(os.path.join(BASE, ".env")) PROMPT_FILE = os.path.join(os.path.dirname(__file__), "prompts", "ad_gate.txt") GATED_DIR = os.path.join(BASE, "data", "gated") REJECTED_LOG = os.path.join(BASE, "data", "rejected-ads.jsonl") SNIPPET_CHARS = 500 GATE_SCHEMA = { "type": "object", "properties": { "relevant": {"type": "boolean"}, "reason": {"type": "string"}, }, "required": ["relevant", "reason"], } # Role families that leak into keyword-fetched corpora. A term is skipped if # it appears in the target occupation's own label/alt labels. BLOCK_FAMILIES = { "designer": ["designer", "design lead", "ux", "ui/ux", "ui &", "graphic", "art director", "illustrator", "creative director"], "recruiter": ["recruiter", "talent acquisition", "sourcer", "headhunter"], "sales": ["sales", "account executive", "account manager", "business development"], "marketing": ["marketing", "growth hacker", "seo ", "social media", "content creator", "brand manager"], "instructor": ["instructor", "teacher", "tutor", "professor", "lecturer", "trainer", "coach"], "management": ["product manager", "product owner", "project manager", "scrum master", "program manager"], } def _occ_meta(slug): """Label, alt labels and ESCO description for the gate prompt (from DB, fallback: package files).""" label, alts, desc = slug.replace("-", " "), "", "" try: mp = os.path.join(BASE, "skills", slug, "manifest.json") m = json.load(open(mp, encoding="utf-8")) label = m.get("title", label) uri = m.get("ids", {}).get("esco_uri") if uri: sys.path.insert(0, os.path.dirname(__file__)) from db import connect cn = connect() cur = cn.cursor() cur.execute("SELECT preferred_label, alt_labels, description " "FROM esco_occupation WHERE concept_uri=?", uri) row = cur.fetchone() cn.close() if row: label = row[0] or label alts = row[1] or "" desc = row[2] or "" except Exception: pass return label, alts, desc def title_block_family(title, occ_text): """Return the blocking family name, or None. occ_text = label + alts lowercased; family terms present there do not block.""" t = " " + (title or "").lower() + " " for family, terms in BLOCK_FAMILIES.items(): for term in terms: if term in t and term.strip() not in occ_text: return family return None def llm_gate(title, snippet, label, alts, desc, url, model, retries=3): prompt = (open(PROMPT_FILE, encoding="utf-8").read() .replace("{LABEL}", label) .replace("{ALTS}", alts[:400] or "-") .replace("{DESCRIPTION}", desc[:1200] or "-") .replace("{TITLE}", title[:200]) .replace("{SNIPPET}", (snippet or "")[:SNIPPET_CHARS])) body = {"model": model, "prompt": prompt, "stream": False, "format": GATE_SCHEMA, "options": {"temperature": 0, "num_ctx": 4096, "num_predict": 160}} err = "no attempt" for attempt in range(retries): try: r = requests.post(f"{url}/api/generate", json=body, timeout=120) r.raise_for_status() rec = json.loads(r.json().get("response", "")) if isinstance(rec.get("relevant"), bool): return rec, None err = "schema mismatch" except (requests.RequestException, json.JSONDecodeError, ValueError) as exc: err = f"{type(exc).__name__}: {exc}" time.sleep(3 * (attempt + 1)) return None, err def gate_ads(slug, ads, url=None, model=None, log=print): """Gate a list of raw ads for one occupation. Returns (accepted_ads, verdicts). Resumable via data/gated/.jsonl; raw ads untouched.""" url = (url or os.environ.get("OLLAMA_URL", "")).rstrip("/") model = model or os.environ.get("OLLAMA_MODEL", "") os.makedirs(GATED_DIR, exist_ok=True) gated_path = os.path.join(GATED_DIR, f"{slug}.jsonl") verdicts = {} if os.path.exists(gated_path): with open(gated_path, encoding="utf-8") as f: for line in f: if line.strip(): v = json.loads(line) verdicts[v["job_id"]] = v label, alts, desc = _occ_meta(slug) occ_text = (label + " " + alts).lower() new_rejects = [] with open(gated_path, "a", encoding="utf-8") as gf: for ad in ads: jid = ad.get("job_id") if not jid or jid in verdicts: continue title = ad.get("title", "") family = title_block_family(title, occ_text) if family: v = {"job_id": jid, "relevant": False, "stage": "blocklist", "reason": f"role family: {family}", "title": title[:200]} else: rec, err = llm_gate(title, ad.get("description", ""), label, alts, desc, url, model) if rec is None: # Gate unavailable -> fail OPEN but mark it, so the run # is repeatable once Ollama is back (verdict not stored). log(f" gate error {slug}/{jid}: {err} — ad passes unGATED") continue v = {"job_id": jid, "relevant": bool(rec["relevant"]), "stage": "llm", "reason": str(rec.get("reason", ""))[:300], "title": title[:200]} verdicts[jid] = v gf.write(json.dumps(v, ensure_ascii=False) + "\n") gf.flush() if not v["relevant"]: new_rejects.append(v) if new_rejects: with open(REJECTED_LOG, "a", encoding="utf-8") as rf: for v in new_rejects: rf.write(json.dumps({ "slug": slug, "job_id": v["job_id"], "title": v["title"], "stage": v["stage"], "reason": v["reason"], "gated_at": datetime.now().isoformat(timespec="seconds"), }, ensure_ascii=False) + "\n") accepted = [a for a in ads if verdicts.get(a.get("job_id"), {}).get("relevant", True)] n_rej = sum(1 for v in verdicts.values() if not v["relevant"]) log(f" gate {slug}: {len(ads)} ads -> {len(accepted)} relevant, " f"{n_rej} rejected ({sum(1 for v in verdicts.values() if v['stage'] == 'blocklist' and not v['relevant'])} via blocklist)") return accepted, verdicts def _reclassify(slugs): raw_dir = os.path.join(BASE, "data", "raw", "jobs") for i, slug in enumerate(slugs, 1): path = os.path.join(raw_dir, f"{slug}_ads.json") if not os.path.exists(path): continue ads = json.load(open(path, encoding="utf-8")) print(f"[{i}/{len(slugs)}] {slug} ({len(ads)} ads)") gate_ads(slug, ads) def main(): ap = argparse.ArgumentParser() ap.add_argument("--slug") ap.add_argument("--all", action="store_true", help="re-classify every stored raw ad file") args = ap.parse_args() if args.slug: _reclassify([args.slug]) elif args.all: raw_dir = os.path.join(BASE, "data", "raw", "jobs") slugs = sorted(f[:-9] for f in os.listdir(raw_dir) if f.endswith("_ads.json")) _reclassify(slugs) else: ap.error("--slug or --all required") if __name__ == "__main__": main()