"""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()