""" Scraper Engine - Orchestrates job scraping with compliance checks. """ import asyncio import hashlib import logging from datetime import datetime, timezone from typing import Any from urllib.parse import urlparse import httpx from robotexclusionrulesparser import RobotExclusionRulesParser from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.models.job import Company, Job from app.models.scraper import JobSource, ScrapeRun, ScrapeRunStatus logger = logging.getLogger(__name__) class ComplianceChecker: """Check robots.txt and enforce crawl policies.""" def __init__(self): self._robots_cache: dict[str, tuple[RobotExclusionRulesParser, datetime]] = {} self._cache_ttl = 3600 # 1 hour async def check_robots(self, url: str, user_agent: str) -> tuple[bool, float | None]: """ Check if URL is allowed by robots.txt. Returns (is_allowed, crawl_delay). """ parsed = urlparse(url) robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt" parser = await self._get_robots(robots_url) if parser is None: # No robots.txt = assume allowed return True, None is_allowed = parser.is_allowed(user_agent, url) crawl_delay = parser.get_crawl_delay(user_agent) return is_allowed, crawl_delay async def _get_robots(self, robots_url: str) -> RobotExclusionRulesParser | None: """Fetch and cache robots.txt.""" now = datetime.now(timezone.utc) if robots_url in self._robots_cache: parser, cached_at = self._robots_cache[robots_url] if (now - cached_at).total_seconds() < self._cache_ttl: return parser try: async with httpx.AsyncClient(timeout=10) as client: response = await client.get(robots_url) if response.status_code == 200: parser = RobotExclusionRulesParser() parser.parse(response.text) self._robots_cache[robots_url] = (parser, now) return parser except Exception as e: logger.warning(f"Failed to fetch robots.txt from {robots_url}: {e}") return None class JobNormalizer: """Normalize and deduplicate job data.""" @staticmethod def compute_content_hash(job_data: dict) -> str: """Generate a content hash for deduplication.""" key_fields = f"{job_data.get('title', '')}-{job_data.get('company_name', '')}-{job_data.get('source_url', '')}" return hashlib.sha256(key_fields.encode()).hexdigest() @staticmethod def normalize_salary(salary_str: str | None) -> tuple[int | None, int | None, str]: """Parse salary strings into min/max/currency.""" if not salary_str: return None, None, "USD" # Basic normalization - expand with regex patterns # TODO: Handle "$100k-$150k", "100,000 - 150,000 GBP", etc. return None, None, "USD" @staticmethod def normalize_location(location_raw: str | None) -> list[dict] | None: """Parse location string into structured data.""" if not location_raw: return None # Basic normalization return [{"raw": location_raw}] @staticmethod def detect_remote_type(title: str, description: str | None, location: str | None) -> str | None: """Detect if job is remote/hybrid/onsite from text.""" text = f"{title} {description or ''} {location or ''}".lower() if "remote" in text: if "hybrid" in text: return "hybrid" return "remote" if "hybrid" in text: return "hybrid" if "on-site" in text or "onsite" in text or "in-office" in text: return "onsite" return None class ScraperEngine: """Main scraper orchestrator.""" def __init__(self, db: AsyncSession): self.db = db self.compliance = ComplianceChecker() self.normalizer = JobNormalizer() self.http_client = httpx.AsyncClient( timeout=30, headers={"User-Agent": settings.SCRAPER_USER_AGENT}, follow_redirects=True, ) async def run_source(self, source: JobSource) -> ScrapeRun: """Execute a scrape run for a source.""" run = ScrapeRun(source_id=source.id, started_at=datetime.now(timezone.utc)) self.db.add(run) await self.db.flush() try: # Compliance check is_allowed, crawl_delay = await self.compliance.check_robots( source.base_url, settings.SCRAPER_USER_AGENT ) run.robots_status = "allowed" if is_allowed else "denied" run.crawl_delay_used = crawl_delay or source.request_delay_seconds if not is_allowed: run.status = ScrapeRunStatus.FAILED run.error_message = "Blocked by robots.txt" run.completed_at = datetime.now(timezone.utc) return run # Route to appropriate scraper run.status = ScrapeRunStatus.RUNNING await self.db.flush() if source.source_type == "ats_api": jobs_data = await self._scrape_ats(source) elif source.source_type == "careers_page": jobs_data = await self._scrape_careers_page(source) elif source.source_type == "sitemap": jobs_data = await self._scrape_sitemap(source) else: jobs_data = [] # Process results for job_data in jobs_data: await self._upsert_job(job_data, source, run) run.status = ScrapeRunStatus.COMPLETED run.completed_at = datetime.now(timezone.utc) run.jobs_found = len(jobs_data) # Update source stats source.last_crawled_at = datetime.now(timezone.utc) source.last_success_at = datetime.now(timezone.utc) source.consecutive_failures = 0 source.total_jobs_found += run.jobs_new except Exception as e: run.status = ScrapeRunStatus.FAILED run.error_message = str(e) run.completed_at = datetime.now(timezone.utc) source.consecutive_failures += 1 logger.error(f"Scrape failed for source {source.name}: {e}") return run async def _scrape_ats(self, source: JobSource) -> list[dict]: """Scrape ATS APIs (Greenhouse, Lever, Ashby, etc.).""" ats_vendor = source.ats_vendor config = source.ats_config or {} if ats_vendor == "greenhouse": return await self._scrape_greenhouse(config) elif ats_vendor == "lever": return await self._scrape_lever(config) elif ats_vendor == "ashby": return await self._scrape_ashby(config) else: logger.warning(f"Unknown ATS vendor: {ats_vendor}") return [] async def _scrape_greenhouse(self, config: dict) -> list[dict]: """Scrape Greenhouse public API - fully compliant, public endpoint.""" board_token = config.get("board_token") if not board_token: return [] url = f"https://boards-api.greenhouse.io/v1/boards/{board_token}/jobs" response = await self.http_client.get(url, params={"content": "true"}) response.raise_for_status() data = response.json() jobs = [] for item in data.get("jobs", []): locations = [] for loc in item.get("offices", []): locations.append({"raw": loc.get("name")}) for loc in item.get("location", {}).get("name", "").split(","): if loc.strip(): locations.append({"raw": loc.strip()}) jobs.append({ "external_id": str(item["id"]), "title": item["title"], "description_html": item.get("content", ""), "source_url": item.get("absolute_url", ""), "locations": locations or None, "department": item.get("departments", [{}])[0].get("name") if item.get("departments") else None, "date_posted": item.get("updated_at"), "ats_vendor": "greenhouse", }) return jobs async def _scrape_lever(self, config: dict) -> list[dict]: """Scrape Lever public postings API.""" company_slug = config.get("company_slug") if not company_slug: return [] url = f"https://api.lever.co/v0/postings/{company_slug}" response = await self.http_client.get(url) response.raise_for_status() data = response.json() jobs = [] for item in data: jobs.append({ "external_id": item["id"], "title": item["text"], "description_html": item.get("descriptionPlain", ""), "source_url": item.get("hostedUrl", ""), "locations": [{"raw": item.get("categories", {}).get("location", "")}], "department": item.get("categories", {}).get("department"), "employment_type": item.get("categories", {}).get("commitment"), "date_posted": item.get("createdAt"), "ats_vendor": "lever", }) return jobs async def _scrape_ashby(self, config: dict) -> list[dict]: """Scrape Ashby public job board API.""" board_slug = config.get("board_slug") if not board_slug: return [] url = "https://api.ashbyhq.com/posting-api/job-board" response = await self.http_client.post( url, json={"jobBoardSlug": board_slug} ) response.raise_for_status() data = response.json() jobs = [] for item in data.get("jobs", []): jobs.append({ "external_id": item["id"], "title": item["title"], "description_html": item.get("descriptionHtml", ""), "source_url": item.get("jobUrl", ""), "locations": [{"raw": loc.get("locationName", "")} for loc in item.get("locations", [])], "department": item.get("departmentName"), "employment_type": item.get("employmentType"), "date_posted": item.get("publishedDate"), "ats_vendor": "ashby", }) return jobs async def _scrape_careers_page(self, source: JobSource) -> list[dict]: """Scrape a company careers page - requires custom parsing.""" # TODO: Implement with selectolax/BeautifulSoup + schema.org detection return [] async def _scrape_sitemap(self, source: JobSource) -> list[dict]: """Parse job listings from XML sitemaps.""" # TODO: Implement sitemap parsing return [] async def _upsert_job(self, job_data: dict, source: JobSource, run: ScrapeRun): """Insert or update a job, handling deduplication.""" content_hash = self.normalizer.compute_content_hash(job_data) # Check for existing job by external_id + source existing = await self.db.execute( select(Job).where( Job.external_id == job_data.get("external_id"), Job.source_domain == source.domain, ) ) existing_job = existing.scalar_one_or_none() if existing_job: # Update existing existing_job.date_seen_last = datetime.now(timezone.utc) if existing_job.content_hash != content_hash: # Content changed - update fields for key, value in job_data.items(): if hasattr(existing_job, key) and value is not None: setattr(existing_job, key, value) existing_job.content_hash = content_hash run.jobs_updated += 1 else: # Create new job remote_type = self.normalizer.detect_remote_type( job_data.get("title", ""), job_data.get("description_html"), str(job_data.get("locations", "")), ) job = Job( external_id=job_data.get("external_id"), source_url=job_data.get("source_url", source.base_url), source_domain=source.domain, source_type=source.source_type, ats_vendor=job_data.get("ats_vendor"), title=job_data["title"], company_name=source.name, description_html=job_data.get("description_html"), locations=job_data.get("locations"), remote_type=remote_type, employment_type=job_data.get("employment_type"), department=job_data.get("department"), date_posted=job_data.get("date_posted"), content_hash=content_hash, raw_payload=job_data, ) self.db.add(job) run.jobs_new += 1 async def close(self): """Cleanup resources.""" await self.http_client.aclose()