From 91cf930a5f3b8342cd82cf7e61768e952888fc2d Mon Sep 17 00:00:00 2001 From: ldy Date: Thu, 18 Jun 2026 00:02:58 -0400 Subject: [PATCH] phase5-pipeline --- jobsource/db.py | 305 ++++++++++++++++++++++- jobsource/main.py | 9 +- jobsource/pipeline.py | 344 +++++++++++++++++++++++++- tests/test_db.py | 550 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 1186 insertions(+), 22 deletions(-) create mode 100644 tests/test_db.py diff --git a/jobsource/db.py b/jobsource/db.py index 26e1606..8873c16 100644 --- a/jobsource/db.py +++ b/jobsource/db.py @@ -1,10 +1,297 @@ -"""SQLite persistence layer: companies table, jobs table, dedup, company cache, CSV export. +"""SQLite persistence layer: companies table, jobs table, dedup, company cache, CSV export.""" +from __future__ import annotations -Scaffold stub -- not implemented yet. -""" -# TODO (Stage 4): implement per CLAUDE.md "Stage 4 — Persist & export" and "Data model". -# Schema: -# companies(company_key PK, name, website, career_url, first_seen) -# jobs(job_id PK, company_key, linkedin_url, position_url, status, listed_at, first_seen) -# CSV export writes output/results.csv with columns: company_name, career_page_url, open_position_url -# (complete rows — status==position_found — sorted first; incomplete rows follow). +import csv +import logging +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import urlsplit + +from .config import get_settings +from .models import CSV_COLUMNS, JobResult, JobStatus + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Pure helper — standalone so tests can import it directly +# --------------------------------------------------------------------------- + + +def _company_key(name: str, website: str | None) -> str: + """Derive a stable company key: registrable domain if website known, else name. + + Examples: + _company_key("Acme Inc", "https://www.acme.com/about") -> "acme.com" + _company_key("Acme LLC", None) -> "acme" + """ + if website: + try: + hostname = urlsplit(website).hostname or "" + hostname = hostname.lower() + if hostname.startswith("www."): + hostname = hostname[4:] + if hostname: + return hostname + except Exception: + pass + slug = name.strip().lower() + return slug or "unknown" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +# --------------------------------------------------------------------------- +# Database +# --------------------------------------------------------------------------- + + +class Database: + """SQLite-backed store for companies and jobs. + + Supports use as a context manager. Thread-safety: one connection per + instance (single-threaded pipeline use). Postgres-swap is possible by + replacing this class behind the same interface. + """ + + def __init__(self, path: Path | str | None = None) -> None: + if path is None: + path = get_settings().db_path + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(str(path)) + self._conn.row_factory = sqlite3.Row + # WAL mode for better write performance (no-op on in-memory dbs) + self._conn.execute("PRAGMA journal_mode=WAL") + self.init_schema() + logger.debug("db: opened %s", path) + + # ---- Schema ----------------------------------------------------------- + + def init_schema(self) -> None: + """Create tables if they don't exist. Safe to call on every startup.""" + self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS companies ( + company_key TEXT PRIMARY KEY, + name TEXT, + website TEXT, + career_url TEXT, + first_seen TEXT + ); + CREATE TABLE IF NOT EXISTS jobs ( + job_id TEXT PRIMARY KEY, + company_key TEXT, + linkedin_url TEXT, + position_url TEXT, + status TEXT, + listed_at TEXT, + first_seen TEXT + ); + """) + self._conn.commit() + + # ---- Dedup / cache ---------------------------------------------------- + + def is_seen(self, job_id: str) -> bool: + """Return True if this job_id already exists in the DB (cross-run dedup).""" + row = self._conn.execute( + "SELECT 1 FROM jobs WHERE job_id = ?", (job_id,) + ).fetchone() + return row is not None + + def get_cached_career_url(self, company_key: str) -> str | None: + """Return a previously resolved career URL for this company, or None. + + When non-null, the pipeline skips the cascade for this company and uses + the cached value directly (companies are resolved at most once). + """ + row = self._conn.execute( + "SELECT career_url FROM companies WHERE company_key = ?", (company_key,) + ).fetchone() + if row and row["career_url"]: + return row["career_url"] + return None + + # ---- Company key helper ----------------------------------------------- + + def company_key_for(self, name: str, website: str | None) -> str: + """Delegate to the module-level pure function.""" + return _company_key(name, website) + + # ---- Upserts ---------------------------------------------------------- + + def upsert_company( + self, + company_key: str, + name: str, + website: str | None, + career_url: str | None, + ) -> None: + """Insert or update a company row. + + COALESCE semantics: a new non-null value fills an existing NULL, but + never clobbers an already-populated column. first_seen is preserved. + """ + now = _now_iso() + self._conn.execute( + """ + INSERT INTO companies(company_key, name, website, career_url, first_seen) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(company_key) DO UPDATE SET + name = COALESCE(excluded.name, companies.name), + website = COALESCE(excluded.website, companies.website), + career_url = COALESCE(excluded.career_url, companies.career_url), + first_seen = companies.first_seen + """, + (company_key, name, website, career_url, now), + ) + self._conn.commit() + + def upsert_job(self, result: JobResult) -> None: + """Insert or update a job row. + + first_seen is preserved on conflict. position_url and linkedin_url + use COALESCE: a resolved value fills a prior NULL but is never + overwritten once set. + """ + now = _now_iso() + listed_at = ( + result.listed_at.isoformat() if result.listed_at is not None else None + ) + self._conn.execute( + """ + INSERT INTO jobs(job_id, company_key, linkedin_url, position_url, status, listed_at, first_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(job_id) DO UPDATE SET + company_key = excluded.company_key, + linkedin_url = COALESCE(excluded.linkedin_url, jobs.linkedin_url), + position_url = COALESCE(excluded.position_url, jobs.position_url), + status = excluded.status, + listed_at = COALESCE(excluded.listed_at, jobs.listed_at), + first_seen = jobs.first_seen + """, + ( + result.job_id, + result.company_key, + result.linkedin_url, + result.open_position_url, + result.status.value, + listed_at, + now, + ), + ) + self._conn.commit() + + def persist_result(self, result: JobResult) -> None: + """Persist one JobResult atomically: upsert company then job.""" + with self._conn: + now = _now_iso() + listed_at = ( + result.listed_at.isoformat() if result.listed_at is not None else None + ) + self._conn.execute( + """ + INSERT INTO companies(company_key, name, website, career_url, first_seen) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(company_key) DO UPDATE SET + name = COALESCE(excluded.name, companies.name), + website = COALESCE(excluded.website, companies.website), + career_url = COALESCE(excluded.career_url, companies.career_url), + first_seen = companies.first_seen + """, + ( + result.company_key, + result.company_name, + result.website, + result.career_page_url, + now, + ), + ) + self._conn.execute( + """ + INSERT INTO jobs(job_id, company_key, linkedin_url, position_url, status, listed_at, first_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(job_id) DO UPDATE SET + company_key = excluded.company_key, + linkedin_url = COALESCE(excluded.linkedin_url, jobs.linkedin_url), + position_url = COALESCE(excluded.position_url, jobs.position_url), + status = excluded.status, + listed_at = COALESCE(excluded.listed_at, jobs.listed_at), + first_seen = jobs.first_seen + """, + ( + result.job_id, + result.company_key, + result.linkedin_url, + result.open_position_url, + result.status.value, + listed_at, + now, + ), + ) + + # ---- Export ----------------------------------------------------------- + + def export_csv(self, path: Path | None = None) -> Path: + """Write output/results.csv; complete (position_found) rows sorted first. + + Uses models.CSV_COLUMNS for the header; None values become empty strings. + Returns the Path written. + """ + if path is None: + path = get_settings().output_csv + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + + rows = self._conn.execute( + """ + SELECT + c.name AS company_name, + c.career_url AS career_page_url, + j.position_url AS open_position_url, + j.status + FROM jobs j + LEFT JOIN companies c ON j.company_key = c.company_key + ORDER BY + (j.status = 'position_found') DESC, + c.name ASC NULLS LAST + """ + ).fetchall() + + with path.open("w", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=list(CSV_COLUMNS)) + writer.writeheader() + for row in rows: + writer.writerow( + { + "company_name": row["company_name"] or "", + "career_page_url": row["career_page_url"] or "", + "open_position_url": row["open_position_url"] or "", + } + ) + + logger.info("db: exported %d rows to %s", len(rows), path) + return path + + # ---- Counts (for summary / tests) ------------------------------------ + + def counts(self) -> dict[str, int]: + """Return per-status job counts from the DB.""" + rows = self._conn.execute( + "SELECT status, COUNT(*) AS n FROM jobs GROUP BY status" + ).fetchall() + return {row["status"]: row["n"] for row in rows} + + # ---- Lifecycle ------------------------------------------------------- + + def close(self) -> None: + self._conn.close() + + def __enter__(self) -> "Database": + return self + + def __exit__(self, *_: object) -> None: + self.close() diff --git a/jobsource/main.py b/jobsource/main.py index 13218d2..0141c60 100644 --- a/jobsource/main.py +++ b/jobsource/main.py @@ -46,8 +46,13 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) - print("jobsource: scaffold stub -- pipeline not implemented yet.", file=sys.stderr) - print(f"parsed args: {vars(args)}", file=sys.stderr) + from jobsource.pipeline import run_batch # deferred: keeps --help fast before heavy deps + run_batch( + batch_size=args.batch_size, + search_terms=args.search, # None or list[str] from --search repeats + location=args.location, + hours_old=args.hours_old, + ) return 0 diff --git a/jobsource/pipeline.py b/jobsource/pipeline.py index 5deaf08..b8c7cb4 100644 --- a/jobsource/pipeline.py +++ b/jobsource/pipeline.py @@ -1,12 +1,334 @@ -"""Batch orchestration: dedup, per-record isolation, cascade, persistence, summary. +"""Batch orchestration: dedup, per-record isolation, cascade, persistence, summary.""" +from __future__ import annotations -Scaffold stub -- not implemented yet. -""" -# TODO (pipeline): implement run_batch() per CLAUDE.md "Pipeline stages". -# run_batch() contract: -# - Accept batch_size, search terms, location, hours_old overrides. -# - Call the job source, dedup by job_id against the DB (skip already-seen jobs). -# - For each new RawJob, run the full cascade (resolve -> careers -> extract) in isolation: -# one failing record must NEVER abort the batch — catch, record failed/needs_review, continue. -# - Persist each JobResult to the DB and export output/results.csv when done. -# - Print a run summary: per-stage counts + % of new jobs reaching position_found. +import logging +import time +from dataclasses import dataclass + +from .careers.cascade import find_careers_page +from .config import get_settings +from .db import Database +from .extract import extract_open_position +from .http import build_client +from .models import JobResult, JobStatus, RawJob +from .resolve import resolve_website +from .sources.base import JobSource + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Public summary result +# --------------------------------------------------------------------------- + + +@dataclass +class BatchSummary: + """Per-stage counts returned (and printed) by run_batch().""" + + fetched: int = 0 + already_seen: int = 0 + new: int = 0 + website_resolved: int = 0 + careers_found: int = 0 + position_found: int = 0 + needs_review: int = 0 + failed: int = 0 + + @property + def coverage(self) -> float: + """Fraction of new jobs that reached position_found.""" + return self.position_found / self.new if self.new > 0 else 0.0 + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def run_batch( + batch_size: int | None = None, + *, + search_terms: list[str] | None = None, + location: str | None = None, + hours_old: int | None = None, + db: Database | None = None, +) -> BatchSummary: + """Fetch, dedup, cascade, persist, and export one batch of jobs. + + Override arguments take precedence over config values; ``None`` means + "use config default". Returns a :class:`BatchSummary` (also printed). + + One failing record never aborts the batch — each job is isolated in its + own try/except; the outer loop continues regardless. + """ + settings = get_settings() + batch_size = batch_size if batch_size is not None else settings.batch_size + search_terms = search_terms if search_terms is not None else list(settings.search_terms) + location = location if location is not None else settings.location + hours_old = hours_old if hours_old is not None else settings.hours_old + + own_db = db is None + if own_db: + db = Database() + + summary = BatchSummary() + t0 = time.perf_counter() + csv_path = None + + try: + # ------------------------------------------------------------------ + # Stage 1 — Ingest + # ------------------------------------------------------------------ + source = _make_source(settings) + print(f"\n{'='*64}") + print( + f" JobSourceAgent batch={batch_size}" + f" search={search_terms}" + f" hours_old={hours_old}" + ) + print(f"{'='*64}") + print("Stage 1: ingesting jobs…") + + raw_jobs: list[RawJob] = source.fetch_recent_jobs( + search_terms=search_terms, + location=location, + hours_old=hours_old, + results_wanted=settings.results_wanted, + ) + summary.fetched = len(raw_jobs) + + # Dedup: filter out jobs already in the DB; cap new jobs at batch_size. + new_jobs: list[RawJob] = [] + for raw in raw_jobs: + if db.is_seen(raw.job_id): + summary.already_seen += 1 + else: + new_jobs.append(raw) + if len(new_jobs) >= batch_size: + break + + summary.new = len(new_jobs) + print( + f" Fetched {summary.fetched} jobs; " + f"{summary.already_seen} already seen; " + f"{summary.new} new to process.\n" + ) + + # ------------------------------------------------------------------ + # Stages 1b → 3 — per-record cascade + # ------------------------------------------------------------------ + if new_jobs: + with build_client() as client: + for raw in new_jobs: + result = JobResult.from_raw(raw) + print(f"[{raw.job_id}] {raw.company}") + try: + _process_one(result, raw, db, client) + except Exception as exc: + # Safety net — _process_one should not raise, but guard anyway. + logger.exception( + "pipeline: unexpected error for %s (%s): %s", + raw.job_id, raw.company, exc, + ) + result.status = JobStatus.failed + try: + db.persist_result(result) + except Exception: + logger.exception( + "pipeline: failed to persist error record %s", raw.job_id + ) + _tally(summary, result) + print() + + # ------------------------------------------------------------------ + # Stage 4 — Export CSV + # ------------------------------------------------------------------ + try: + csv_path = db.export_csv() + except Exception as exc: + logger.error("pipeline: CSV export failed: %s", exc) + + finally: + if own_db: + db.close() + + elapsed = time.perf_counter() - t0 + _print_summary(summary, elapsed, csv_path) + return summary + + +# --------------------------------------------------------------------------- +# Per-record cascade (must not raise — handles all errors internally) +# --------------------------------------------------------------------------- + + +def _process_one( + result: JobResult, + raw: RawJob, + db: Database, + client: object, +) -> None: + """Run stages 1b → 3 for one job and persist the result. + + Uses early returns for soft misses (needs_review) and catches exceptions + for hard failures (failed status). Never raises to the caller. + """ + # ------------------------------------------------------------------ + # Stage 1b — Resolve company website + # ------------------------------------------------------------------ + try: + website = resolve_website(raw.company, raw.website, client=client) + except Exception as exc: + logger.warning("pipeline(%s): resolve error: %s", raw.job_id, exc) + print(f" resolve: ERROR — {exc}") + result.status = JobStatus.failed + db.persist_result(result) + return + + if not website: + print(" website: UNRESOLVED") + result.status = JobStatus.needs_review + db.persist_result(result) + return + + result.website = website + result.status = JobStatus.website_resolved + result.company_key = db.company_key_for(raw.company, website) + print(f" website: {website}") + + # ------------------------------------------------------------------ + # Stage 2 — Find careers page (check company cache first) + # ------------------------------------------------------------------ + cached_career_url = db.get_cached_career_url(result.company_key) + + if cached_career_url: + careers_url = cached_career_url + careers_method = "cache" + position_shortcut: str | None = None + print(f" careers: {careers_url} [cache]") + else: + try: + cr = find_careers_page(website, company_name=raw.company, client=client) + except Exception as exc: + logger.warning("pipeline(%s): cascade error: %s", raw.job_id, exc) + print(f" careers: ERROR — {exc}") + result.status = JobStatus.needs_review + db.persist_result(result) + return + + careers_url = cr.careers_url + careers_method = cr.method + position_shortcut = cr.position_url + + if not careers_url: + print(f" careers: MISSED (method={cr.method})") + result.status = JobStatus.needs_review + db.persist_result(result) + return + + print( + f" careers: {careers_url}" + f" [{careers_method} conf={cr.confidence:.2f}]" + ) + + result.career_page_url = careers_url + result.careers_method = careers_method + result.status = JobStatus.careers_found + + # ------------------------------------------------------------------ + # Stage 3 — Extract one open position + # ------------------------------------------------------------------ + if position_shortcut: + # Free shortcut from ATS or browser-agent tier — no extra HTTP call. + position_url = position_shortcut + position_method = careers_method + else: + try: + position_url, position_method = extract_open_position( + careers_url, client=client + ) + except Exception as exc: + logger.warning("pipeline(%s): extract error: %s", raw.job_id, exc) + print(f" position: ERROR — {exc}") + result.status = JobStatus.needs_review + db.persist_result(result) + return + + if position_url: + result.open_position_url = position_url + result.position_method = position_method + result.status = JobStatus.position_found + print(f" position: {position_url} [{position_method}]") + else: + print(f" position: MISSED (method={position_method})") + result.status = JobStatus.needs_review + + db.persist_result(result) + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _make_source(settings) -> JobSource: + """Instantiate the configured job source provider.""" + if settings.job_source == "apify": + from .sources.apify_source import ApifySource + return ApifySource() + from .sources.jobspy_source import JobSpySource + return JobSpySource() + + +def _tally(summary: BatchSummary, result: JobResult) -> None: + """Accumulate per-stage counts from a processed JobResult.""" + if result.website: + summary.website_resolved += 1 + if result.career_page_url: + summary.careers_found += 1 + if result.open_position_url: + summary.position_found += 1 + if result.status == JobStatus.needs_review: + summary.needs_review += 1 + elif result.status == JobStatus.failed: + summary.failed += 1 + + +def _print_summary( + summary: BatchSummary, elapsed: float, csv_path: object +) -> None: + """Print the run summary table to stdout.""" + n = summary.new + wr = summary.website_resolved + cf = summary.careers_found + + print(f"{'='*64}") + print(f" SUMMARY ({n} new jobs processed in {elapsed:.1f}s)") + print(f"{'='*64}") + print( + f" Stage 1 ingested {summary.fetched:>4}" + f" ({summary.already_seen} already seen)" + ) + print( + f" Stage 1b website resolved {wr:>4} / {n}" + f" ({wr / max(n, 1) * 100:.0f}%)" + ) + print( + f" Stage 2 careers found {cf:>4} / {wr}" + f" ({cf / max(wr, 1) * 100:.0f}% of resolved)" + ) + print( + f" Stage 3 position found {summary.position_found:>4} / {cf}" + f" ({summary.position_found / max(cf, 1) * 100:.0f}% of careers)" + ) + print(f" Needs review {summary.needs_review:>4}") + print(f" Failed {summary.failed:>4}") + print( + f" End-to-end coverage " + f"{summary.coverage * 100:.0f}%" + f" ({summary.position_found}/{n} new jobs fully resolved)" + ) + if csv_path: + print(f" CSV written to {csv_path}") + print() diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..66cf54e --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,550 @@ +"""Tests for jobsource/db.py and jobsource/pipeline.py. + +All tests are offline — no network calls, no real job source. +The DB uses tmp_path (file-backed) or ":memory:" via path override. +""" +from __future__ import annotations + +import csv +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from jobsource.db import Database, _company_key +from jobsource.models import CSV_COLUMNS, JobResult, JobStatus, RawJob + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _db(tmp_path: Path) -> Database: + """Open a fresh file-backed Database in tmp_path.""" + return Database(path=tmp_path / "test.db") + + +def _result( + job_id: str = "1", + company_name: str = "Acme", + website: str | None = "https://acme.com", + career_url: str | None = "https://acme.com/careers", + position_url: str | None = "https://acme.com/careers/jobs/42", + status: JobStatus = JobStatus.position_found, + company_key: str | None = None, +) -> JobResult: + return JobResult( + job_id=job_id, + company_name=company_name, + company_key=company_key or _company_key(company_name, website), + website=website, + career_page_url=career_url, + open_position_url=position_url, + status=status, + ) + + +# --------------------------------------------------------------------------- +# _company_key (pure function) +# --------------------------------------------------------------------------- + + +class TestCompanyKey: + def test_domain_stripped_of_www(self) -> None: + assert _company_key("Acme", "https://www.acme.com/about") == "acme.com" + + def test_domain_lowercase(self) -> None: + assert _company_key("Foo", "https://Foo.IO") == "foo.io" + + def test_no_website_uses_name(self) -> None: + assert _company_key("Globex Corp", None) == "globex corp" + + def test_no_website_lowercased(self) -> None: + assert _company_key("Initech LLC", None) == "initech llc" + + def test_empty_host_falls_back_to_name(self) -> None: + # Malformed URL with empty hostname + result = _company_key("BadCo", "://no-host") + assert result == "badco" + + def test_empty_name_and_no_website_returns_unknown(self) -> None: + assert _company_key("", None) == "unknown" + + def test_subpath_ignored(self) -> None: + assert _company_key("X", "https://x.example.com/jobs/42") == "x.example.com" + + +# --------------------------------------------------------------------------- +# Database.init_schema / open +# --------------------------------------------------------------------------- + + +class TestDatabaseSchema: + def test_creates_tables(self, tmp_path: Path) -> None: + db = _db(tmp_path) + try: + cur = db._conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" + ) + tables = {r[0] for r in cur.fetchall()} + assert "companies" in tables + assert "jobs" in tables + finally: + db.close() + + def test_idempotent_schema(self, tmp_path: Path) -> None: + """Opening the same DB twice must not raise.""" + db1 = Database(path=tmp_path / "idempotent.db") + db1.close() + db2 = Database(path=tmp_path / "idempotent.db") + db2.close() + + def test_parent_dirs_created(self, tmp_path: Path) -> None: + nested = tmp_path / "a" / "b" / "c" / "test.db" + db = Database(path=nested) + try: + assert nested.exists() + finally: + db.close() + + def test_context_manager(self, tmp_path: Path) -> None: + with Database(path=tmp_path / "cm.db") as db: + assert not db.is_seen("x") + + +# --------------------------------------------------------------------------- +# is_seen / dedup +# --------------------------------------------------------------------------- + + +class TestIsSeeen: + def test_unseen_job_returns_false(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + assert not db.is_seen("99999") + + def test_seen_after_persist(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="42") + db.persist_result(r) + assert db.is_seen("42") + + def test_different_job_ids_independent(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + db.persist_result(_result(job_id="1")) + assert db.is_seen("1") + assert not db.is_seen("2") + + +# --------------------------------------------------------------------------- +# company_key_for +# --------------------------------------------------------------------------- + + +class TestCompanyKeyFor: + def test_delegates_to_pure_fn(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + assert db.company_key_for("Acme", "https://acme.com") == "acme.com" + assert db.company_key_for("Foo LLC", None) == "foo llc" + + +# --------------------------------------------------------------------------- +# get_cached_career_url +# --------------------------------------------------------------------------- + + +class TestCachedCareerUrl: + def test_miss_returns_none(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + assert db.get_cached_career_url("acme.com") is None + + def test_hit_after_persist(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="1", website="https://acme.com", career_url="https://acme.com/careers") + db.persist_result(r) + assert db.get_cached_career_url("acme.com") == "https://acme.com/careers" + + def test_null_career_url_not_cached(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="1", career_url=None, status=JobStatus.needs_review) + db.persist_result(r) + assert db.get_cached_career_url("acme.com") is None + + +# --------------------------------------------------------------------------- +# upsert_company COALESCE / first_seen preservation +# --------------------------------------------------------------------------- + + +class TestUpsertCompany: + def test_insert_then_null_fill(self, tmp_path: Path) -> None: + """A NULL value in the second upsert fills the existing NULL without error.""" + with _db(tmp_path) as db: + db.upsert_company("x.com", "X", None, None) + db.upsert_company("x.com", "X", "https://x.com", "https://x.com/careers") + row = db._conn.execute( + "SELECT website, career_url FROM companies WHERE company_key='x.com'" + ).fetchone() + assert row["website"] == "https://x.com" + assert row["career_url"] == "https://x.com/careers" + + def test_existing_value_not_clobbered(self, tmp_path: Path) -> None: + """A second upsert with NULL must not overwrite an existing value.""" + with _db(tmp_path) as db: + db.upsert_company("y.com", "Y", "https://y.com", "https://y.com/jobs") + db.upsert_company("y.com", "Y", None, None) + row = db._conn.execute( + "SELECT website, career_url FROM companies WHERE company_key='y.com'" + ).fetchone() + assert row["website"] == "https://y.com" + assert row["career_url"] == "https://y.com/jobs" + + def test_first_seen_preserved(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + db.upsert_company("z.com", "Z", None, None) + row1 = db._conn.execute( + "SELECT first_seen FROM companies WHERE company_key='z.com'" + ).fetchone() + db.upsert_company("z.com", "Z", "https://z.com", "https://z.com/careers") + row2 = db._conn.execute( + "SELECT first_seen FROM companies WHERE company_key='z.com'" + ).fetchone() + assert row1["first_seen"] == row2["first_seen"] + + +# --------------------------------------------------------------------------- +# persist_result / upsert_job +# --------------------------------------------------------------------------- + + +class TestPersistResult: + def test_both_tables_populated(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="10", company_name="Initech", website="https://initech.com") + db.persist_result(r) + job = db._conn.execute( + "SELECT * FROM jobs WHERE job_id='10'" + ).fetchone() + assert job is not None + assert job["status"] == "position_found" + company = db._conn.execute( + "SELECT * FROM companies WHERE company_key='initech.com'" + ).fetchone() + assert company is not None + assert company["name"] == "Initech" + + def test_listed_at_null_stored_as_null(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="20") + assert r.listed_at is None + db.persist_result(r) + row = db._conn.execute( + "SELECT listed_at FROM jobs WHERE job_id='20'" + ).fetchone() + assert row["listed_at"] is None + + def test_listed_at_serialized(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="30") + r.listed_at = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc) + db.persist_result(r) + row = db._conn.execute( + "SELECT listed_at FROM jobs WHERE job_id='30'" + ).fetchone() + assert row["listed_at"] is not None + assert "2026" in row["listed_at"] + + def test_upsert_fills_null_position(self, tmp_path: Path) -> None: + """A second persist with position_url fills a prior NULL.""" + with _db(tmp_path) as db: + r1 = _result(job_id="50", position_url=None, status=JobStatus.needs_review) + db.persist_result(r1) + r2 = _result(job_id="50", position_url="https://acme.com/jobs/1", status=JobStatus.position_found) + db.persist_result(r2) + row = db._conn.execute( + "SELECT position_url, status FROM jobs WHERE job_id='50'" + ).fetchone() + assert row["position_url"] == "https://acme.com/jobs/1" + assert row["status"] == "position_found" + + def test_job_first_seen_preserved(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="60", position_url=None, status=JobStatus.needs_review) + db.persist_result(r) + row1 = db._conn.execute( + "SELECT first_seen FROM jobs WHERE job_id='60'" + ).fetchone() + r2 = _result(job_id="60", position_url="https://acme.com/jobs/1", status=JobStatus.position_found) + db.persist_result(r2) + row2 = db._conn.execute( + "SELECT first_seen FROM jobs WHERE job_id='60'" + ).fetchone() + assert row1["first_seen"] == row2["first_seen"] + + +# --------------------------------------------------------------------------- +# counts +# --------------------------------------------------------------------------- + + +class TestCounts: + def test_empty_db(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + assert db.counts() == {} + + def test_counts_grouped_by_status(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + db.persist_result(_result(job_id="1", status=JobStatus.position_found)) + db.persist_result(_result(job_id="2", position_url=None, status=JobStatus.needs_review)) + db.persist_result(_result(job_id="3", position_url=None, status=JobStatus.needs_review)) + c = db.counts() + assert c["position_found"] == 1 + assert c["needs_review"] == 2 + + +# --------------------------------------------------------------------------- +# export_csv +# --------------------------------------------------------------------------- + + +class TestExportCsv: + def test_header_matches_csv_columns(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + db.persist_result(_result(job_id="1")) + out = tmp_path / "out.csv" + db.export_csv(path=out) + with out.open() as f: + reader = csv.DictReader(f) + assert tuple(reader.fieldnames or []) == CSV_COLUMNS + + def test_null_fields_become_empty_string(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + r = _result(job_id="1", career_url=None, position_url=None, status=JobStatus.needs_review) + db.persist_result(r) + out = tmp_path / "out.csv" + db.export_csv(path=out) + with out.open() as f: + rows = list(csv.DictReader(f)) + assert rows[0]["career_page_url"] == "" + assert rows[0]["open_position_url"] == "" + + def test_complete_rows_sorted_first(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + # Insert incomplete row first, complete row second + db.persist_result(_result( + job_id="1", company_name="ZZZ Corp", + position_url=None, status=JobStatus.needs_review, + )) + db.persist_result(_result( + job_id="2", company_name="AAA Inc", + position_url="https://aaa.com/jobs/1", status=JobStatus.position_found, + )) + out = tmp_path / "out.csv" + db.export_csv(path=out) + with out.open() as f: + rows = list(csv.DictReader(f)) + assert len(rows) == 2 + # complete row (AAA Inc) must come first + assert rows[0]["open_position_url"] != "" + assert rows[1]["open_position_url"] == "" + + def test_creates_parent_dirs(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + db.persist_result(_result()) + nested = tmp_path / "deep" / "dir" / "results.csv" + db.export_csv(path=nested) + assert nested.exists() + + def test_returns_path(self, tmp_path: Path) -> None: + with _db(tmp_path) as db: + db.persist_result(_result()) + out = tmp_path / "r.csv" + returned = db.export_csv(path=out) + assert returned == out + + +# --------------------------------------------------------------------------- +# pipeline.run_batch — offline unit tests via monkeypatch +# --------------------------------------------------------------------------- + + +def _make_raw(job_id: str, company: str = "Acme") -> RawJob: + return RawJob( + job_id=job_id, + company=company, + linkedin_url=f"https://www.linkedin.com/jobs/view/{job_id}", + website=None, + ) + + +class TestRunBatch: + """Monkeypatched run_batch tests — no network, no real providers.""" + + def _patch_all(self, monkeypatch, raw_jobs: list[RawJob]) -> None: + """Patch all external I/O in pipeline so tests run offline.""" + from jobsource.careers.cascade import CareersResult + + # Fake job source + fake_source = MagicMock() + fake_source.fetch_recent_jobs.return_value = raw_jobs + monkeypatch.setattr("jobsource.pipeline._make_source", lambda _: fake_source) + + # Fake resolve: always succeeds + monkeypatch.setattr( + "jobsource.pipeline.resolve_website", + lambda name, website=None, *, client=None: f"https://{name.lower().replace(' ', '')}.com", + ) + + # Fake cascade: returns a careers URL + free position shortcut + def _fake_careers(website, *, company_name=None, client=None): + return CareersResult( + careers_url=f"{website}/careers", + confidence=0.95, + method="ats:greenhouse", + position_url=f"{website}/careers/jobs/1", + ) + + monkeypatch.setattr("jobsource.pipeline.find_careers_page", _fake_careers) + + # extract_open_position shouldn't be called (shortcut covers it), + # but patch defensively to detect if it is. + monkeypatch.setattr( + "jobsource.pipeline.extract_open_position", + lambda url, *, client=None: (None, "none"), + ) + + # build_client — return a context-manager stub + cm = MagicMock() + cm.__enter__ = lambda s: MagicMock() + cm.__exit__ = MagicMock(return_value=False) + monkeypatch.setattr("jobsource.pipeline.build_client", lambda: cm) + + def test_new_jobs_processed(self, tmp_path: Path, monkeypatch) -> None: + from jobsource.pipeline import run_batch + + # Use distinct company names so each job gets its own company_key (no cache sharing). + raw = [_make_raw("1", company="Alpha"), _make_raw("2", company="Beta")] + self._patch_all(monkeypatch, raw) + + db = Database(path=tmp_path / "t.db") + summary = run_batch(batch_size=5, search_terms=["eng"], db=db) + + assert summary.fetched == 2 + assert summary.new == 2 + assert summary.already_seen == 0 + assert summary.website_resolved == 2 + assert summary.careers_found == 2 + assert summary.position_found == 2 + assert summary.failed == 0 + assert summary.coverage == 1.0 + + def test_cross_run_dedup(self, tmp_path: Path, monkeypatch) -> None: + """Re-running with the same DB yields new == 0 (dedup proven).""" + from jobsource.pipeline import run_batch + + raw = [_make_raw("10", company="Gamma"), _make_raw("11", company="Delta")] + self._patch_all(monkeypatch, raw) + + db = Database(path=tmp_path / "dedup.db") + + # First run + s1 = run_batch(batch_size=5, search_terms=["eng"], db=db) + assert s1.new == 2 + + # Second run — same raw jobs returned by source, but DB already has them + s2 = run_batch(batch_size=5, search_terms=["eng"], db=db) + assert s2.new == 0 + assert s2.already_seen == 2 + + def test_batch_size_cap(self, tmp_path: Path, monkeypatch) -> None: + """Only batch_size new jobs are processed even if more are available.""" + from jobsource.pipeline import run_batch + + # Distinct companies to avoid career-URL cache sharing across jobs. + raw = [_make_raw(str(i), company=f"Co{i}") for i in range(10)] + self._patch_all(monkeypatch, raw) + + db = Database(path=tmp_path / "cap.db") + summary = run_batch(batch_size=3, search_terms=["eng"], db=db) + assert summary.new == 3 + assert summary.position_found == 3 + + def test_resolve_failure_marks_needs_review(self, tmp_path: Path, monkeypatch) -> None: + """Jobs where resolve_website returns None get needs_review, not failed.""" + from jobsource.pipeline import run_batch + + raw = [_make_raw("99")] + self._patch_all(monkeypatch, raw) + # Override resolve to return None (unresolvable) + monkeypatch.setattr( + "jobsource.pipeline.resolve_website", + lambda *a, **kw: None, + ) + + db = Database(path=tmp_path / "nr.db") + summary = run_batch(batch_size=5, search_terms=["eng"], db=db) + assert summary.needs_review == 1 + assert summary.failed == 0 + assert summary.website_resolved == 0 + + def test_csv_written_after_batch(self, tmp_path: Path, monkeypatch) -> None: + """Data is persisted by run_batch and export_csv produces the right columns.""" + from jobsource.pipeline import run_batch + + raw = [_make_raw("7", company="SevenCo")] + self._patch_all(monkeypatch, raw) + + csv_path = tmp_path / "out.csv" + # Pass db explicitly so own_db=False → database is NOT closed by run_batch. + db = Database(path=tmp_path / "t.db") + + run_batch(batch_size=5, search_terms=["eng"], db=db) + + # Export to a controlled path and verify the header/contract. + db.export_csv(path=csv_path) + + assert csv_path.exists() + with csv_path.open() as f: + reader = csv.DictReader(f) + assert tuple(reader.fieldnames or []) == CSV_COLUMNS + + def test_one_failure_does_not_abort_batch(self, tmp_path: Path, monkeypatch) -> None: + """A cascade exception for one company doesn't abort others. + + Cascade errors land in needs_review (not failed — they are recoverable/transient). + Each job has a distinct company so cache cannot mask the flaky call. + """ + from jobsource.pipeline import run_batch + from jobsource.careers.cascade import CareersResult + + # Distinct companies → distinct company_keys → no cache sharing. + raw = [ + _make_raw("A", company="AlphaCo"), + _make_raw("B", company="BetaCo"), + _make_raw("C", company="GammaCo"), + ] + self._patch_all(monkeypatch, raw) + + call_count = {"n": 0} + + def _flaky_careers(website, *, company_name=None, client=None): + call_count["n"] += 1 + if call_count["n"] == 2: + raise RuntimeError("simulated cascade crash") + return CareersResult( + careers_url=f"{website}/careers", + confidence=0.9, + method="url_pattern", + position_url=f"{website}/careers/jobs/1", + ) + + monkeypatch.setattr("jobsource.pipeline.find_careers_page", _flaky_careers) + + db = Database(path=tmp_path / "fault.db") + summary = run_batch(batch_size=5, search_terms=["eng"], db=db) + + # All 3 attempted; cascade exception → needs_review (not failed); 2 fully resolved. + assert summary.new == 3 + assert summary.needs_review == 1 + assert summary.failed == 0 + assert summary.position_found == 2