""" Universal Web Scraper - scrapes ANY website's career/job pages. Uses multiple strategies: schema.org, meta tags, CSS heuristics, and LLM extraction. No platform-specific code - works on any site with job listings. """ import hashlib import json import logging import re from datetime import datetime, timezone from typing import Any from urllib.parse import urljoin, urlparse import httpx from selectolax.parser import HTMLParser from app.core.config import settings logger = logging.getLogger(__name__) class UniversalJobExtractor: """ Extracts job postings from ANY webpage using multiple strategies: 1. Schema.org JSON-LD (most reliable) 2. OpenGraph / meta tags 3. Microdata attributes 4. CSS/HTML heuristic patterns 5. LLM-based extraction (fallback for complex pages) """ def __init__(self, http_client: httpx.AsyncClient | None = None): self.client = http_client or httpx.AsyncClient( timeout=30, headers={"User-Agent": settings.SCRAPER_USER_AGENT}, follow_redirects=True, ) async def extract_from_url(self, url: str) -> dict: """ Fetch a URL and extract job posting data using all available strategies. Returns structured job data or empty dict if no job found. """ try: response = await self.client.get(url) if response.status_code != 200: return {"error": f"HTTP {response.status_code}", "url": url} html = response.text return self.extract_from_html(html, url) except Exception as e: logger.error(f"Failed to fetch {url}: {e}") return {"error": str(e), "url": url} def extract_from_html(self, html: str, page_url: str) -> dict: """Extract job data from HTML content using all strategies.""" tree = HTMLParser(html) result = {} # Strategy 1: JSON-LD (highest confidence) jsonld = self._extract_jsonld(tree, page_url) if jsonld and jsonld.get("title"): jsonld["extraction_method"] = "json_ld" jsonld["confidence"] = "high" return jsonld # Strategy 2: Meta tags (OpenGraph, Twitter Cards) meta = self._extract_meta_tags(tree, page_url) if meta and meta.get("title"): meta["extraction_method"] = "meta_tags" meta["confidence"] = "medium" return meta # Strategy 3: Microdata microdata = self._extract_microdata(tree, page_url) if microdata and microdata.get("title"): microdata["extraction_method"] = "microdata" microdata["confidence"] = "medium" return microdata # Strategy 4: Heuristic CSS selectors heuristic = self._extract_heuristic(tree, page_url) if heuristic and heuristic.get("title"): heuristic["extraction_method"] = "heuristic" heuristic["confidence"] = "low" return heuristic # Strategy 5: Raw text extraction for LLM processing raw_text = self._extract_raw_text(tree) return { "extraction_method": "raw_text", "confidence": "needs_llm", "raw_text": raw_text[:5000], "url": page_url, "title": self._guess_title(tree), } def _extract_jsonld(self, tree: HTMLParser, page_url: str) -> dict | None: """Extract from schema.org JSON-LD.""" for script in tree.css('script[type="application/ld+json"]'): try: data = json.loads(script.text()) jobs = self._find_job_postings(data) if jobs: return self._normalize_schema_job(jobs[0], page_url) except (json.JSONDecodeError, TypeError): continue return None def _find_job_postings(self, data: Any) -> list[dict]: """Recursively find JobPosting objects in JSON-LD.""" results = [] if isinstance(data, dict): if data.get("@type") == "JobPosting" or "JobPosting" in str(data.get("@type", "")): results.append(data) if "@graph" in data: for item in data["@graph"]: results.extend(self._find_job_postings(item)) elif isinstance(data, list): for item in data: results.extend(self._find_job_postings(item)) return results def _normalize_schema_job(self, data: dict, page_url: str) -> dict: """Normalize schema.org JobPosting to internal format.""" # Location location = "" job_location = data.get("jobLocation") if job_location: if isinstance(job_location, list): job_location = job_location[0] if isinstance(job_location, dict): addr = job_location.get("address", {}) if isinstance(addr, dict): location = ", ".join(filter(None, [ addr.get("addressLocality"), addr.get("addressRegion"), addr.get("addressCountry") ])) elif isinstance(addr, str): location = addr # Remote remote_type = None if "TELECOMMUTE" in str(data.get("jobLocationType", "")).upper(): remote_type = "remote" elif data.get("applicantLocationRequirements"): remote_type = "remote" # Salary salary_min, salary_max, salary_currency = None, None, "USD" base_salary = data.get("baseSalary") or data.get("estimatedSalary") if isinstance(base_salary, dict): value = base_salary.get("value", {}) if isinstance(value, dict): salary_min = value.get("minValue") salary_max = value.get("maxValue") salary_currency = base_salary.get("currency", "USD") # Company company = "" org = data.get("hiringOrganization", {}) if isinstance(org, dict): company = org.get("name", "") # Employment type emp_type = data.get("employmentType", "") if isinstance(emp_type, list): emp_type = emp_type[0] if emp_type else "" type_map = {"FULL_TIME": "full_time", "PART_TIME": "part_time", "CONTRACTOR": "contract", "INTERN": "internship"} employment_type = type_map.get(emp_type.upper(), None) return { "title": data.get("title", ""), "company_name": company, "description": data.get("description", ""), "location": location, "remote_type": remote_type, "employment_type": employment_type, "salary_min": int(salary_min) if salary_min else None, "salary_max": int(salary_max) if salary_max else None, "salary_currency": salary_currency, "date_posted": data.get("datePosted"), "valid_through": data.get("validThrough"), "apply_url": data.get("url") or page_url, "source_url": page_url, "skills": self._extract_skills_from_schema(data), "education": data.get("educationRequirements"), "experience": data.get("experienceRequirements"), } def _extract_skills_from_schema(self, data: dict) -> list[str] | None: """Extract skills from schema.org data.""" skills = data.get("skills") if skills: if isinstance(skills, str): return [s.strip() for s in skills.split(",") if s.strip()] if isinstance(skills, list): return skills return None def _extract_meta_tags(self, tree: HTMLParser, page_url: str) -> dict | None: """Extract from OpenGraph and meta tags.""" title = None description = None # OG tags og_title = tree.css_first('meta[property="og:title"]') og_desc = tree.css_first('meta[property="og:description"]') og_type = tree.css_first('meta[property="og:type"]') if og_title: title = og_title.attributes.get("content", "") if og_desc: description = og_desc.attributes.get("content", "") # Check if it's likely a job page page_title = tree.css_first("title") title_text = page_title.text() if page_title else "" if not self._looks_like_job(title or title_text, page_url): return None if not title: title = title_text # Try to extract company from various places company = "" og_site = tree.css_first('meta[property="og:site_name"]') if og_site: company = og_site.attributes.get("content", "") return { "title": self._clean_title(title), "company_name": company, "description": description or "", "source_url": page_url, "location": "", } if title else None def _extract_microdata(self, tree: HTMLParser, page_url: str) -> dict | None: """Extract from HTML microdata attributes.""" job_scope = tree.css_first('[itemtype*="JobPosting"]') if not job_scope: return None title = "" company = "" location = "" description = "" title_el = job_scope.css_first('[itemprop="title"]') if title_el: title = title_el.text().strip() org_el = job_scope.css_first('[itemprop="hiringOrganization"] [itemprop="name"]') if org_el: company = org_el.text().strip() loc_el = job_scope.css_first('[itemprop="jobLocation"] [itemprop="addressLocality"]') if loc_el: location = loc_el.text().strip() desc_el = job_scope.css_first('[itemprop="description"]') if desc_el: description = desc_el.text().strip() return { "title": title, "company_name": company, "description": description, "location": location, "source_url": page_url, } if title else None def _extract_heuristic(self, tree: HTMLParser, page_url: str) -> dict | None: """Extract using CSS selector heuristics that work across many sites.""" title = "" company = "" location = "" description = "" # Title selectors (ordered by specificity) title_selectors = [ 'h1[class*="job-title"]', 'h1[class*="posting-title"]', 'h1[class*="position"]', '[data-qa*="job-title"]', '[data-testid*="job-title"]', '.job-title h1', '.posting-headline h2', '.job-header h1', 'h1[class*="title"]', 'article h1', 'main h1', 'h1', ] for sel in title_selectors: el = tree.css_first(sel) if el and el.text().strip() and len(el.text().strip()) > 3: title = el.text().strip() break # Company selectors company_selectors = [ '[class*="company-name"]', '[class*="employer"]', '[class*="organization"]', '[data-qa*="company"]', '[data-testid*="company"]', '.company-name', '.employer-name', ] for sel in company_selectors: el = tree.css_first(sel) if el and el.text().strip(): company = el.text().strip() break if not company: # Try to get from domain parsed = urlparse(page_url) company = parsed.netloc.replace("www.", "").split(".")[0].title() # Location selectors location_selectors = [ '[class*="location"]', '[class*="job-location"]', '[data-qa*="location"]', '[data-testid*="location"]', '.location', '.job-location', ] for sel in location_selectors: el = tree.css_first(sel) if el and el.text().strip(): location = el.text().strip() break # Description desc_selectors = [ '[class*="job-description"]', '[class*="description"]', '[class*="posting-content"]', '[data-qa*="description"]', '.job-description', '.description', 'article', ] for sel in desc_selectors: el = tree.css_first(sel) if el and len(el.text().strip()) > 100: description = el.text().strip()[:5000] break if not title: return None return { "title": self._clean_title(title), "company_name": company, "description": description, "location": location, "source_url": page_url, } def _extract_raw_text(self, tree: HTMLParser) -> str: """Extract clean text from page for LLM processing.""" # Remove script, style, nav, footer for tag in tree.css("script, style, nav, footer, header, aside"): tag.decompose() body = tree.css_first("main") or tree.css_first("article") or tree.css_first("body") if body: return re.sub(r'\s+', ' ', body.text()).strip()[:5000] return "" def _guess_title(self, tree: HTMLParser) -> str: """Best guess at page title.""" h1 = tree.css_first("h1") if h1: return h1.text().strip() title = tree.css_first("title") if title: return title.text().strip().split("|")[0].split("-")[0].strip() return "" def _clean_title(self, title: str) -> str: """Clean common title suffixes.""" # Remove company names and site names from title for sep in [" | ", " - ", " — ", " · ", " at "]: parts = title.split(sep) if len(parts) > 1: title = parts[0].strip() break return title.strip() def _looks_like_job(self, text: str, url: str) -> bool: """Heuristic: does this look like a job posting?""" text_lower = (text + " " + url).lower() job_signals = ["job", "career", "position", "role", "opening", "hiring", "apply", "engineer", "developer", "manager", "designer", "analyst"] return any(signal in text_lower for signal in job_signals) class UniversalCrawler: """ Crawls any website's career section to discover job listing URLs. Works without any site-specific configuration. """ def __init__(self): self.client = httpx.AsyncClient( timeout=30, headers={"User-Agent": settings.SCRAPER_USER_AGENT}, follow_redirects=True, ) self.extractor = UniversalJobExtractor(self.client) async def discover_jobs(self, careers_url: str, max_pages: int = 50) -> list[dict]: """ Discover job listings from a careers page. Strategy: 1. Fetch the careers page 2. Find all links that look like individual job postings 3. Extract job data from each """ try: response = await self.client.get(careers_url) if response.status_code != 200: return [] tree = HTMLParser(response.text) base_url = f"{urlparse(careers_url).scheme}://{urlparse(careers_url).netloc}" # Find job listing links job_urls = set() for link in tree.css("a[href]"): href = link.attributes.get("href", "") if not href: continue full_url = urljoin(base_url, href) # Check if it looks like a job posting URL if self._is_job_url(full_url, careers_url): job_urls.add(full_url) # Also check for JSON API endpoints embedded in the page api_jobs = await self._check_embedded_apis(tree, base_url) # Extract from discovered URLs (limit to max_pages) jobs = list(api_jobs) for url in list(job_urls)[:max_pages - len(jobs)]: job_data = await self.extractor.extract_from_url(url) if job_data and job_data.get("title") and not job_data.get("error"): jobs.append(job_data) return jobs except Exception as e: logger.error(f"Discovery failed for {careers_url}: {e}") return [] async def _check_embedded_apis(self, tree: HTMLParser, base_url: str) -> list[dict]: """Check for embedded API calls that load job data (common in SPAs).""" jobs = [] # Look for data attributes or script tags with JSON job data for script in tree.css("script"): text = script.text() or "" # Look for JSON arrays of jobs if '"jobs"' in text or '"postings"' in text or '"positions"' in text: try: # Try to extract JSON from script json_match = re.search(r'(\[[\s\S]*?"title"[\s\S]*?\])', text) if json_match: data = json.loads(json_match.group(1)) if isinstance(data, list) and len(data) > 0 and "title" in data[0]: for item in data[:50]: jobs.append({ "title": item.get("title", ""), "company_name": item.get("company", item.get("company_name", "")), "location": item.get("location", ""), "source_url": item.get("url", item.get("absolute_url", base_url)), "extraction_method": "embedded_json", "confidence": "medium", }) except (json.JSONDecodeError, TypeError): pass return jobs def _is_job_url(self, url: str, base_careers_url: str) -> bool: """Determine if a URL is likely an individual job posting.""" url_lower = url.lower() base_domain = urlparse(base_careers_url).netloc # Must be same domain if urlparse(url).netloc != base_domain: return False # URL patterns that indicate individual job pages job_patterns = [ r"/jobs?/\d+", r"/jobs?/[a-f0-9-]{8,}", # /job/123 or /jobs/uuid r"/positions?/\d+", r"/positions?/[a-z0-9-]+", r"/openings?/", r"/opportunities?/", r"/careers?/[^/]+/[^/]+", # /careers/engineering/senior-dev r"/job-details", r"/job-posting", ] # Must match at least one pattern if not any(re.search(pattern, url_lower) for pattern in job_patterns): # Also check if it's a child page of the careers URL if not url.startswith(base_careers_url): return False # And has more path depth if url.rstrip("/") == base_careers_url.rstrip("/"): return False # Exclude common non-job pages exclude_patterns = ["/blog", "/about", "/contact", "/faq", "/privacy", "/terms", "/login", "/signup", "/search"] if any(pattern in url_lower for pattern in exclude_patterns): return False return True async def close(self): await self.client.aclose()