skillfactor-pipeline/pipeline/p3b_store_evidence.py

"""Phase 3b — store per-ad skill extractions in the MSSQL evidence store.

Tables:
- evidence_job(job_id PK, title, employer, country, seniority, occupation_slug)
- evidence_entity(job_id, entity_type, entity)

Idempotent full-rebuild:  main() (recruiter standalone)
Idempotent per-occupation: store_for_occupation(cn, ads_dict, extractions, slug)
"""
import json
import os
import sys

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

BASE = os.path.join(os.path.dirname(__file__), "..")
ADS = os.path.join(BASE, "data", "raw", "jobs", "recruiter_ads.json")
EXTRACTIONS = os.path.join(BASE, "data", "evidence", "extractions.json")

CANONICAL = {
    "linkedin": "LinkedIn / LinkedIn Recruiter",
    "linkedin recruiter": "LinkedIn / LinkedIn Recruiter",
    "ms office": "Microsoft Office",
    "microsoft office suite": "Microsoft Office",
}


def norm(s: str) -> str:
    v = " ".join(str(s).strip().split())
    return CANONICAL.get(v.lower(), v)


def ensure_schema(cn):
    """Create evidence tables if they do not exist (idempotent)."""
    cur = cn.cursor()
    cur.execute("""
        IF OBJECT_ID('evidence_job','U') IS NULL
        CREATE TABLE evidence_job (
            job_id NVARCHAR(100) NOT NULL PRIMARY KEY,
            title NVARCHAR(400), employer NVARCHAR(400),
            country NVARCHAR(5), seniority NVARCHAR(20),
            occupation_slug NVARCHAR(200))
    """)
    cur.execute("""
        IF OBJECT_ID('evidence_entity','U') IS NULL
        CREATE TABLE evidence_entity (
            job_id NVARCHAR(100) NOT NULL,
            entity_type NVARCHAR(30) NOT NULL,
            entity NVARCHAR(300) NOT NULL)
    """)
    cur.execute("""
        IF NOT EXISTS (
            SELECT 1 FROM sys.indexes WHERE name='IX_evidence_entity'
            AND object_id = OBJECT_ID('evidence_entity'))
        CREATE INDEX IX_evidence_entity ON evidence_entity(entity_type, entity)
    """)
    cn.commit()


def store_for_occupation(cn, ads_dict, extractions, slug, max_retries=3):
    """Idempotent upsert for one occupation: delete its old rows, insert fresh.

    Returns (n_jobs_inserted, n_entities_inserted).
    Retries on SQL Server deadlock (error 1205).
    Silently skips any job_id that already exists under a different occupation
    (PK=job_id is globally unique; cross-occupation overlap is rare in practice).
    """
    import time as _time
    import pyodbc as _pyodbc

    for attempt in range(max_retries):
        try:
            cur = cn.cursor()
            existing_jids = [r[0] for r in cur.execute(
                "SELECT job_id FROM evidence_job WHERE occupation_slug=?", slug
            ).fetchall()]
            if existing_jids:
                for jid in existing_jids:
                    cur.execute("DELETE FROM evidence_entity WHERE job_id=?", jid)
                cur.execute("DELETE FROM evidence_job WHERE occupation_slug=?", slug)

            n_jobs = n_ent = 0
            for e in extractions:
                jid = e["job_id"]
                if jid not in ads_dict:
                    # not in the gated-relevant set (phase 2a) — extraction
                    # output stays on disk as raw asset, but is never stored
                    continue
                ad = ads_dict[jid]
                try:
                    cur.execute(
                        "INSERT INTO evidence_job "
                        "(job_id, title, employer, country, seniority, occupation_slug) "
                        "VALUES (?,?,?,?,?,?)",
                        jid, ad.get("title"), ad.get("employer"),
                        ad.get("country"), e.get("seniority"), slug)
                except _pyodbc.IntegrityError:
                    continue  # PK clash with another occupation — skip
                for etype in ("hard_skills", "tools", "methods", "responsibilities"):
                    for entity in e.get(etype, []):
                        cur.execute("INSERT INTO evidence_entity VALUES (?,?,?)",
                                    jid, etype, norm(entity))
                        n_ent += 1
                n_jobs += 1
            cn.commit()
            return n_jobs, n_ent

        except _pyodbc.Error as exc:
            if "1205" in str(exc) and attempt < max_retries - 1:
                try:
                    cn.rollback()
                except Exception:
                    pass
                _time.sleep(5 * (attempt + 1))
                continue
            raise


def main():
    if not (os.path.exists(ADS) and os.path.exists(EXTRACTIONS)):
        print("SKIP: ads or extractions missing — TODO in README")
        return
    ads = {a["job_id"]: a for a in json.load(open(ADS, encoding="utf-8"))}
    ext = json.load(open(EXTRACTIONS, encoding="utf-8"))

    cn = connect()
    cur = cn.cursor()
    cur.execute("IF OBJECT_ID('evidence_entity','U') IS NOT NULL DROP TABLE evidence_entity")
    cur.execute("IF OBJECT_ID('evidence_job','U') IS NOT NULL DROP TABLE evidence_job")
    cur.execute("""CREATE TABLE evidence_job (
        job_id NVARCHAR(100) NOT NULL PRIMARY KEY,
        title NVARCHAR(400), employer NVARCHAR(400),
        country NVARCHAR(5), seniority NVARCHAR(20),
        occupation_slug NVARCHAR(200))""")
    cur.execute("""CREATE TABLE evidence_entity (
        job_id NVARCHAR(100) NOT NULL,
        entity_type NVARCHAR(30) NOT NULL,
        entity NVARCHAR(300) NOT NULL)""")
    cur.execute("CREATE INDEX IX_evidence_entity ON evidence_entity(entity_type, entity)")

    n_ent = 0
    for e in ext:
        ad = ads.get(e["job_id"], {})
        cur.execute(
            "INSERT INTO evidence_job "
            "(job_id, title, employer, country, seniority, occupation_slug) VALUES (?,?,?,?,?,?)",
            e["job_id"], ad.get("title"), ad.get("employer"),
            ad.get("country"), e.get("seniority"), "recruitment-consultant")
        for etype in ("hard_skills", "tools", "methods", "responsibilities"):
            for entity in e.get(etype, []):
                cur.execute("INSERT INTO evidence_entity VALUES (?,?,?)",
                            e["job_id"], etype, norm(entity))
                n_ent += 1
    cn.commit()
    cn.close()
    print(f"evidence: {len(ext)} jobs, {n_ent} entities loaded")


if __name__ == "__main__":
    main()