"""GovernmentJobs.com main search adapter — URL keyword search, 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("govtjobs") class GovtJobsAdapter(ManifestDrivenAdapter): def collect_cards(self, page, manifest) -> list[RawJobCard]: """Extract job cards from the search results page.""" raw_data = page.evaluate("""() => { const cards = document.querySelectorAll('h3 a[href*="/jobs/"]'); return Array.from(cards).map(a => { const cardDiv = a.closest('h3')?.parentElement; const generics = cardDiv ? Array.from(cardDiv.children).filter(el => el.tagName !== 'H3') : []; return { title: a.innerText.trim(), url: a.href, company: generics[0]?.innerText?.trim() || null, location: generics[1]?.innerText?.trim() || null, salary: generics[2]?.innerText?.trim() || null }; }).filter(j => j.title && j.url.includes('/jobs/')); }""") cards = [] for item in raw_data: cards.append(RawJobCard( title=item["title"], company=item["company"], location=item["location"], salary_text=item["salary"], url=item["url"], )) logger.info(f"[govtjobs] Extracted {len(cards)} cards") return cards def paginate(self, page, page_index: int, config) -> bool: """Navigate to next page using URL ?page=N parameter.""" has_next = page.evaluate("""() => { const links = Array.from(document.querySelectorAll('a')); const next = links.find(a => a.textContent.trim() === 'Next' || (a.getAttribute('aria-label') || '').toLowerCase().includes('next page') ); return !!(next && !next.hasAttribute('disabled') && next.getAttribute('aria-disabled') !== 'true'); }""") if not has_next: return False next_page_num = page_index + 2 # page_index 0 → go to page 2 current_url = page.url if re.search(r'[?&]page=\d+', current_url): next_url = re.sub(r'(page=)\d+', f'page={next_page_num}', current_url) elif '?' in current_url: next_url = current_url + f'&page={next_page_num}' else: next_url = current_url + f'?page={next_page_num}' logger.debug(f"[govtjobs] Navigating to page {next_page_num}: {next_url}") page.goto(next_url, timeout=30000) try: page.wait_for_selector( "h3 a[href*='/jobs/']", timeout=15000, state="attached" ) except Exception: logger.debug("[govtjobs] No jobs found on next page — stopping") return False return True def normalize(self, raw: RawJobCard) -> JobPostingCandidate: return super().normalize(raw)