"""USAJOBS adapter — waits for JS render, handles relative URLs, URL-based pagination.""" from __future__ import annotations import re from loguru import logger from gimme_job.adapters.base import ManifestDrivenAdapter from gimme_job.adapters.registry import register from gimme_job.models.dto import JobPostingCandidate, RawJobCard @register("usajobs") class UsaJobsAdapter(ManifestDrivenAdapter): _BASE_URL = "https://www.usajobs.gov" def _wait_for_results(self, page) -> None: """Wait for JS-rendered results or zero-results heading.""" try: page.wait_for_selector( "#search-results .border, h2.font-normal", timeout=12000, state="attached", ) except Exception: logger.debug("[usajobs] Results wait timed out — may be zero results") def collect_cards(self, page, manifest): self._wait_for_results(page) return super().collect_cards(page, manifest) def paginate(self, page, page_index: int, config) -> bool: """Navigate to next page by incrementing the &p= URL parameter.""" max_pages = config.pagination.max_pages if page_index >= max_pages - 1: return False current_url = page.url next_page = page_index + 2 # page_index is 0-based; URL param is 1-based # Replace &p=N or add &p=N if re.search(r"[?&]p=\d+", current_url): next_url = re.sub(r"(p=)\d+", f"p={next_page}", current_url) else: sep = "&" if "?" in current_url else "?" next_url = f"{current_url}{sep}p={next_page}" logger.debug(f"[usajobs] Paginating to page {next_page}: {next_url}") page.goto(next_url, timeout=30000) self._wait_for_results(page) return True def normalize(self, raw: RawJobCard) -> JobPostingCandidate: candidate = super().normalize(raw) # USAJOBS job links are relative: /job/123456 → absolute URL if candidate.job_url and candidate.job_url.startswith("/"): candidate.job_url = self._BASE_URL + candidate.job_url return candidate