"""LinkedIn Jobs adapter — scroll to load cards, paginate via start= URL param.""" from __future__ import annotations 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 def _deduplicate_title(title: str) -> str: """Remove LinkedIn's accessibility text duplication. LinkedIn renders job titles as: Title Title with verification inner_text() returns both, so we get "Title Title with verification". Strategy: if the title starts with a repeated prefix word-by-word, keep the shorter part. """ words = title.split() n = len(words) # Try splitting at each midpoint from 1 to n//2 for split in range(1, n // 2 + 1): prefix = words[:split] rest = words[split:] # Exact full duplication: "Foo Bar Foo Bar" if rest == prefix: return " ".join(prefix) # Prefix repeats at start of rest: "Foo Foo with verification" if rest[:split] == prefix: return " ".join(prefix) return title @register("linkedin") class LinkedInAdapter(ManifestDrivenAdapter): def collect_cards(self, page, manifest) -> list[RawJobCard]: """Scroll each card into view to trigger lazy-loading, then extract.""" self._scroll_to_load(page) return super().collect_cards(page, manifest) def normalize(self, raw: RawJobCard) -> JobPostingCandidate: # Fix relative URLs → absolute if raw.url and raw.url.startswith("/"): raw = raw.model_copy(update={"url": "https://www.linkedin.com" + raw.url}) # Extract job ID from URL for stable fingerprinting (immune to text drift) import re as _re if raw.url and not raw.external_job_id: m = _re.search(r"/jobs/view/(\d+)", raw.url) if m: raw = raw.model_copy(update={"external_job_id": m.group(1)}) # Fix duplicated titles caused by LinkedIn's accessibility span # e.g. "Orthodontist Orthodontist with verification" → "Orthodontist" # e.g. "LOCUM Dentist ... LOCUM Dentist ..." → "LOCUM Dentist ..." if raw.title: raw = raw.model_copy(update={"title": _deduplicate_title(raw.title)}) return super().normalize(raw) def paginate(self, page, page_index: int, config) -> bool: """Navigate to next page by incrementing start= by 25.""" if page_index >= config.pagination.max_pages - 1: return False import re start = (page_index + 1) * 25 current_url = page.url # Replace existing start param or append it if "start=" in current_url: new_url = re.sub(r"start=\d+", f"start={start}", current_url) else: sep = "&" if "?" in current_url else "?" new_url = current_url + sep + f"start={start}" logger.info(f"[linkedin] Page {page_index + 2}: start={start}") page.goto(new_url, timeout=30000) try: page.wait_for_selector( "li[data-occludable-job-id]", timeout=15000, state="attached" ) except Exception: logger.debug("[linkedin] Wait selector timed out after pagination") return True @staticmethod def _scroll_to_load(page) -> None: """Scroll cards into view one by one, re-checking count as LinkedIn lazy-loads more.""" try: scrolled = 0 max_cards = 30 # LinkedIn shows 25 per page; cap with margin while scrolled < max_cards: count = page.evaluate( "document.querySelectorAll('li[data-occludable-job-id]').length" ) if scrolled >= count: break page.evaluate(f"""() => {{ const cards = document.querySelectorAll('li[data-occludable-job-id]'); if (cards[{scrolled}]) cards[{scrolled}].scrollIntoView({{behavior: 'instant', block: 'center'}}); }}""") page.wait_for_timeout(200) scrolled += 1 logger.debug(f"[linkedin] Scrolled through {scrolled} cards") except Exception as e: logger.debug(f"[linkedin] scroll_to_load failed: {e}")