"""Cheap-LLM link classification for careers page and job links (Stage 2, tier 5 / Stage 3, tier 4). Two typed classification functions backed by Pydantic AI: 1. classify_careers_link(anchors) -> CareerLinkResult | None Given a list of (url, text) pairs from a page, pick the single careers/jobs page URL. 2. classify_job_link(anchors) -> JobLinkResult | None Given a list of (url, text) pairs from a careers page, pick one specific open-position URL. Both return None and the cascade falls through gracefully when: - classifier_model or llm_api_key is a placeholder (LLM not configured), - the anchor list is empty, - the LLM call fails for any reason, - the model abstains (index -1) or returns an out-of-range index. pydantic_ai is imported lazily inside _build_link_agent so a missing package or unconfigured provider never breaks module import (graceful-degradation rule #8). extract_anchors() is also exported here so cascade.py and extract.py can share the same anchor-parsing logic without a separate utility module. """ from __future__ import annotations import logging from urllib.parse import urljoin from bs4 import BeautifulSoup from pydantic import BaseModel, Field from ..config import get_settings logger = logging.getLogger(__name__) # Maximum number of anchors forwarded to the LLM (token-budget guard). _MAX_ANCHORS = 60 # Maximum anchor text length, in characters, before truncation. _MAX_ANCHOR_TEXT = 80 # --------------------------------------------------------------------------- # Data models # --------------------------------------------------------------------------- class Anchor(BaseModel): """One hyperlink extracted from a page — the LLM's input row.""" url: str text: str = "" class _LinkChoice(BaseModel): """LLM structured output: 0-based index into the candidate list + confidence.""" index: int = Field( description=( "0-based index of the best link in the candidate list. " "Use -1 if no link qualifies." ) ) confidence: float = Field(default=0.0, ge=0.0, le=1.0) class CareerLinkResult(BaseModel): """Public result of classify_careers_link().""" careers_url: str confidence: float = 0.0 class JobLinkResult(BaseModel): """Public result of classify_job_link().""" job_url: str confidence: float = 0.0 # --------------------------------------------------------------------------- # System-level prompt instructions for each classification task # --------------------------------------------------------------------------- _CAREERS_INSTRUCTIONS = ( "You are a link classifier for a job-sourcing pipeline. " "You receive a numbered list of hyperlinks extracted from a company website. " "Identify the single link that leads to the company's main careers or jobs page " "(the page that lists open positions). " "Respond with the 0-based index of the best match and your confidence (0.0–1.0). " "If none of the links leads to a careers or jobs page, respond with index -1." ) _JOB_INSTRUCTIONS = ( "You are a link classifier for a job-sourcing pipeline. " "You receive a numbered list of hyperlinks extracted from a careers/jobs page. " "Identify the single link that leads directly to one specific open job-posting detail page " "(a page for exactly one position — not a category page, a filter, or a job listing index). " "Respond with the 0-based index of the best match and your confidence (0.0–1.0). " "If none of the links qualifies as a single-job detail page, respond with index -1." ) # --------------------------------------------------------------------------- # Shared anchor extractor (used by cascade.py Tier 5 and extract.py Tiers 3-4) # --------------------------------------------------------------------------- def extract_anchors(html: str, base_url: str) -> list[Anchor]: """Extract deduplicated absolute hyperlinks from HTML. Skips mailto:, tel:, # (fragment-only), and javascript: hrefs. Converts relative hrefs to absolute using base_url. Returns only http(s) links in first-seen order, URL-deduplicated. Truncates anchor text to _MAX_ANCHOR_TEXT characters. """ if not html: return [] soup = BeautifulSoup(html, "lxml") seen_urls: set[str] = set() result: list[Anchor] = [] for tag in soup.find_all("a", href=True): href: str = tag["href"].strip() if not href or href.startswith(("mailto:", "tel:", "#", "javascript:")): continue full_url = urljoin(base_url, href) if not full_url.startswith(("http://", "https://")): continue if full_url in seen_urls: continue seen_urls.add(full_url) text = tag.get_text(separator=" ", strip=True)[:_MAX_ANCHOR_TEXT] result.append(Anchor(url=full_url, text=text)) return result # --------------------------------------------------------------------------- # LLM availability gate # --------------------------------------------------------------------------- def _llm_unavailable() -> bool: """Return True when the LLM is not configured (placeholder values in settings).""" s = get_settings() return ( s.classifier_model.startswith("PLACEHOLDER") or s.llm_api_key.startswith("PLACEHOLDER") ) # --------------------------------------------------------------------------- # Agent builder — extracted as a named function to serve as the test seam. # Monkeypatch this function in tests to avoid real LLM calls. # --------------------------------------------------------------------------- def _build_link_agent(instructions: str): # type: ignore[return] """Construct a Pydantic AI Agent for link classification. pydantic_ai is imported lazily here so a missing package never crashes module import. This function is the primary test seam — monkeypatch it to inject a FakeAgent without touching the real Pydantic AI machinery. """ from pydantic_ai import Agent # lazy import return Agent( get_settings().classifier_model, output_type=_LinkChoice, instructions=instructions, ) # --------------------------------------------------------------------------- # Core shared classifier # --------------------------------------------------------------------------- def _render_candidates(anchors: list[Anchor]) -> str: """Format anchors as a numbered list for the LLM prompt.""" lines = [f"[{i}] {a.url} | {a.text}" for i, a in enumerate(anchors)] return "\n".join(lines) def _choose_link( anchors: list[Anchor], *, instructions: str, label: str, ) -> "_LinkChoice | None": """Run the LLM classifier on anchors; return a _LinkChoice or None. Returns None when the LLM is unconfigured, the list is empty, the model abstains (index -1), the index is out of range, or any exception occurs. """ if not anchors or _llm_unavailable(): return None candidates = anchors[:_MAX_ANCHORS] try: agent = _build_link_agent(instructions) prompt = _render_candidates(candidates) out = agent.run_sync(prompt).output if isinstance(out, _LinkChoice) and 0 <= out.index < len(candidates): return out return None # abstain (index -1) or out-of-range except Exception as exc: logger.warning("classify_llm(%s): tier error: %s", label, exc) return None # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def classify_careers_link(anchors: list[Anchor]) -> CareerLinkResult | None: """Pick the careers/jobs page URL from a list of extracted anchors (Stage 2, tier 5). Returns a CareerLinkResult on success, or None on any no-op / failure. Never raises. """ c = _choose_link(anchors, instructions=_CAREERS_INSTRUCTIONS, label="careers") if c is None: return None candidates = anchors[:_MAX_ANCHORS] return CareerLinkResult(careers_url=candidates[c.index].url, confidence=c.confidence) def classify_job_link(anchors: list[Anchor]) -> JobLinkResult | None: """Pick one specific open-position URL from a careers-page anchor list (Stage 3, tier 4). Returns a JobLinkResult on success, or None on any no-op / failure. Never raises. """ c = _choose_link(anchors, instructions=_JOB_INSTRUCTIONS, label="job") if c is None: return None candidates = anchors[:_MAX_ANCHORS] return JobLinkResult(job_url=candidates[c.index].url, confidence=c.confidence)