phase5-pipeline
This commit is contained in:
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user