You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
168 lines
5.4 KiB
Python
168 lines
5.4 KiB
Python
"""Indeed job site adapter."""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import TYPE_CHECKING
|
|
|
|
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, SearchQuery
|
|
|
|
if TYPE_CHECKING:
|
|
from playwright.sync_api import Page
|
|
|
|
from gimme_job.models.manifest import SiteManifest
|
|
|
|
|
|
@register("indeed")
|
|
class IndeedAdapter(ManifestDrivenAdapter):
|
|
"""Indeed-specific adapter with custom card extraction and URL normalization."""
|
|
|
|
def prepare(self, page: "Page", config: "SiteManifest") -> None:
|
|
"""Navigate to Indeed search URL with fromage=1 for last 24h."""
|
|
url = config.build_start_url(
|
|
type("Q", (), {
|
|
"keywords_urlencoded": page.__class__.__name__, # placeholder
|
|
})()
|
|
)
|
|
# Actually build URL using the query stored in the manifest
|
|
pass # URL is built in orchestrator before adapter.prepare() is called
|
|
|
|
def apply_search(self, page: "Page", query: SearchQuery) -> None:
|
|
"""Indeed uses URL params — no form input needed."""
|
|
pass
|
|
|
|
def apply_filters(self, page: "Page", query: SearchQuery) -> None:
|
|
"""Indeed date filter is already in the URL (fromage=1). Nothing to click."""
|
|
pass
|
|
|
|
def collect_cards(self, page: "Page", config: "SiteManifest") -> list[RawJobCard]:
|
|
"""Extract Indeed job cards."""
|
|
cards = []
|
|
|
|
# Try container selectors
|
|
container_selectors = config.extract.container_selectors or [
|
|
"div.job_seen_beacon",
|
|
"div[data-jk]",
|
|
"td.resultContent",
|
|
]
|
|
|
|
container_elements = []
|
|
for selector in container_selectors:
|
|
try:
|
|
elements = page.query_selector_all(selector)
|
|
if elements:
|
|
container_elements = elements
|
|
logger.debug(f"[indeed] Container: '{selector}' ({len(elements)} items)")
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if not container_elements:
|
|
logger.warning("[indeed] No job cards found")
|
|
return cards
|
|
|
|
for el in container_elements:
|
|
try:
|
|
card = self._extract_card(el)
|
|
if card and card.title:
|
|
cards.append(card)
|
|
except Exception as e:
|
|
logger.debug(f"[indeed] Card extraction error: {e}")
|
|
|
|
logger.info(f"[indeed] Extracted {len(cards)} cards")
|
|
return cards
|
|
|
|
def _extract_card(self, el) -> RawJobCard | None:
|
|
def get_text(*selectors: str) -> str | None:
|
|
for sel in selectors:
|
|
try:
|
|
child = el.query_selector(sel)
|
|
if child:
|
|
text = child.inner_text()
|
|
if text and text.strip():
|
|
return text.strip()
|
|
except Exception:
|
|
continue
|
|
return None
|
|
|
|
def get_attr(selector: str, attr: str) -> str | None:
|
|
try:
|
|
child = el.query_selector(selector)
|
|
if child:
|
|
val = child.get_attribute(attr)
|
|
return val.strip() if val else None
|
|
except Exception:
|
|
return None
|
|
|
|
title = get_text(
|
|
"h2.jobTitle a span[id^='jobTitle']",
|
|
"h2.jobTitle span",
|
|
"h2 a span",
|
|
"h2[class*='jobTitle'] a",
|
|
"h2",
|
|
)
|
|
if not title:
|
|
return None
|
|
|
|
company = get_text(
|
|
"span[data-testid='company-name']",
|
|
".company",
|
|
"span.companyName",
|
|
)
|
|
location = get_text(
|
|
"div[data-testid='text-location']",
|
|
".companyLocation",
|
|
)
|
|
posted_text = get_text(
|
|
"span[data-testid='myJobsStateDate']",
|
|
"span.date",
|
|
)
|
|
salary = get_text(
|
|
"div.salary-snippet-container",
|
|
"div[data-testid='attribute_snippet_testid']",
|
|
)
|
|
|
|
# URL: prefer the job card link
|
|
url = get_attr("h2.jobTitle a", "href") or get_attr("a[data-jk]", "href")
|
|
if url and url.startswith("/"):
|
|
url = "https://www.indeed.com" + url
|
|
if url:
|
|
url = self._clean_indeed_url(url)
|
|
|
|
raw_text = None
|
|
try:
|
|
raw_text = el.inner_text()
|
|
except Exception:
|
|
pass
|
|
|
|
return RawJobCard(
|
|
title=title,
|
|
company=company,
|
|
location=location,
|
|
posted_text=posted_text,
|
|
url=url,
|
|
salary_text=salary,
|
|
raw_text=raw_text,
|
|
)
|
|
|
|
def _clean_indeed_url(self, url: str) -> str:
|
|
"""Strip Indeed tracking parameters, keep the /viewjob?jk= part."""
|
|
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
|
|
try:
|
|
parsed = urlparse(url)
|
|
if parsed.path == "/rc/clk" or "/clk" in parsed.path:
|
|
# Extract jk parameter and build clean URL
|
|
params = parse_qs(parsed.query)
|
|
jk = params.get("jk", [None])[0]
|
|
if jk:
|
|
return f"https://www.indeed.com/viewjob?jk={jk}"
|
|
return url
|
|
except Exception:
|
|
return url
|
|
|
|
def normalize(self, raw: RawJobCard) -> JobPostingCandidate:
|
|
return super().normalize(raw)
|