skillfactor-pipeline/pipeline/ad_gate.py

"""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/<slug>.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 <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/<slug>.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()