diff --git a/job_detail_parser.py b/job_detail_parser.py index 225d269..93d97cc 100644 --- a/job_detail_parser.py +++ b/job_detail_parser.py @@ -5,7 +5,10 @@ import re import logging -from typing import List, Dict, Optional +import random +import time +from itertools import cycle +from typing import List, Dict, Optional, Sequence from bs4 import BeautifulSoup import requests from job_details_models import JobDetails, PageParseError, NetworkError @@ -14,20 +17,89 @@ class JobDetailParser: """Parser for extracting detailed job information from Stepstone job pages""" - - def __init__(self): - self.headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'} - + + DEFAULT_USER_AGENTS: Sequence[str] = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 " + "(KHTML, like Gecko) Version/17.0 Safari/605.1.15", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + ) + + def __init__( + self, + *, + min_delay: float = 0.5, + max_delay: float = 1.5, + max_retries: int = 3, + backoff_factor: float = 2.0, + user_agents: Optional[Sequence[str]] = None, + session: Optional[requests.Session] = None, + request_timeout: int = 15, + ): + if min_delay < 0 or max_delay < 0: + raise ValueError("Delay values must be non-negative") + if max_delay < min_delay: + raise ValueError("max_delay must be greater than or equal to min_delay") + + self.min_delay = min_delay + self.max_delay = max_delay + self.max_retries = max_retries + self.backoff_factor = backoff_factor + self.request_timeout = request_timeout + + agents = tuple(user_agents) if user_agents else self.DEFAULT_USER_AGENTS + if not agents: + raise ValueError("At least one User-Agent must be provided") + self._user_agent_cycle = cycle(agents) + + self.base_headers = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + } + self.headers = self.base_headers.copy() + + self.session = session or requests.Session() + + def _build_headers(self) -> Dict[str, str]: + headers = self.base_headers.copy() + headers["User-Agent"] = next(self._user_agent_cycle) + self.headers = headers + return headers + + def _apply_throttle(self) -> float: + if self.max_delay == 0: + return 0.0 + delay = random.uniform(self.min_delay, self.max_delay) + if delay > 0: + time.sleep(delay) + return delay + + def _calculate_backoff(self, attempt: int) -> float: + backoff = self.min_delay * (self.backoff_factor ** attempt) + return max(backoff, 0.0) + def fetch_job_page(self, url: str) -> str: """Fetch the HTML content of a job page""" - try: - response = requests.get(url, headers=self.headers, timeout=15) - response.raise_for_status() - return response.text - except requests.RequestException as e: - logger.error(f"Network error fetching job page {url}: {e}") - raise NetworkError(f"Failed to fetch job page: {str(e)}") - + for attempt in range(1, self.max_retries + 1): + try: + self._apply_throttle() + response = self.session.get( + url, + headers=self._build_headers(), + timeout=self.request_timeout, + ) + response.raise_for_status() + return response.text + except requests.RequestException as e: + logger.error(f"Network error fetching job page {url}: {e}") + if attempt == self.max_retries: + raise NetworkError(f"Failed to fetch job page: {str(e)}") + backoff = self._calculate_backoff(attempt) + if backoff > 0: + time.sleep(backoff) + def parse_job_details(self, url: str) -> JobDetails: """Parse comprehensive job details from a Stepstone job page""" try: diff --git a/stepstone_server.py b/stepstone_server.py index 4fd0320..fb234e6 100644 --- a/stepstone_server.py +++ b/stepstone_server.py @@ -9,8 +9,11 @@ import asyncio import json import logging +import random +import time from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Any, Dict, List, Optional +from itertools import cycle +from typing import Any, Dict, List, Optional, Sequence import requests from bs4 import BeautifulSoup import re @@ -38,110 +41,184 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger("stepstone-server") +DEFAULT_USER_AGENTS: Sequence[str] = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "+ + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 "+ + "(KHTML, like Gecko) Version/17.0 Safari/605.1.15", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "+ + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", +) + + class StepstoneJobScraper: - """Job scraper for Stepstone.de""" - - def __init__(self): - self.headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'} - + """Job scraper for Stepstone.de with throttling and retry safeguards.""" + + def __init__( + self, + *, + min_delay: float = 0.5, + max_delay: float = 1.5, + max_retries: int = 3, + backoff_factor: float = 2.0, + user_agents: Optional[Sequence[str]] = None, + session: Optional[requests.Session] = None, + request_timeout: int = 10, + ): + if min_delay < 0 or max_delay < 0: + raise ValueError("Delay values must be non-negative") + if max_delay < min_delay: + raise ValueError("max_delay must be greater than or equal to min_delay") + + self.min_delay = min_delay + self.max_delay = max_delay + self.max_retries = max_retries + self.backoff_factor = backoff_factor + self.request_timeout = request_timeout + + agents = tuple(user_agents) if user_agents else DEFAULT_USER_AGENTS + if not agents: + raise ValueError("At least one User-Agent must be provided") + self._user_agent_cycle = cycle(agents) + + # Store the last used headers for observability/testing + self.base_headers = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + } + self.headers = self.base_headers.copy() + + self.session = session or requests.Session() + + def _build_headers(self) -> Dict[str, str]: + headers = self.base_headers.copy() + headers["User-Agent"] = next(self._user_agent_cycle) + self.headers = headers + return headers + + def _apply_throttle(self) -> float: + if self.max_delay == 0: + return 0.0 + delay = random.uniform(self.min_delay, self.max_delay) + if delay > 0: + time.sleep(delay) + return delay + + def _calculate_backoff(self, attempt: int) -> float: + backoff = self.min_delay * (self.backoff_factor ** attempt) + return max(backoff, 0.0) + def fetch_job_listings(self, url: str) -> List[Dict[str, str]]: """Fetch job listings from a Stepstone URL""" - try: - response = requests.get(url, headers=self.headers, timeout=10) - response.raise_for_status() - - soup = BeautifulSoup(response.text, 'html.parser') - container = soup.find('div', id='app-unifiedResultlist') - - if not container: - logger.warning(f"No job container found for URL: {url}") - return [] - - jobs = [] - seen_links = set() - - for article in container.find_all('article', attrs={'data-testid': 'job-item'}): - # Find all links in the article - all_links = article.find_all('a', href=True) - - job_link = None - job_title = None - - # First, look for job posting links with the correct pattern - for link_elem in all_links: - href = link_elem.get('href', '') - - # Check for actual job posting URLs (contain stellenangebote and inline.html) - if re.search(r'/stellenangebote--.*--\d+-inline\.html', href): - job_link = link_elem - job_title = link_elem.get_text(strip=True) - break - - # If no job posting link found, look for relative links starting with /stellenangebote - if not job_link: + for attempt in range(1, self.max_retries + 1): + try: + self._apply_throttle() + response = self.session.get( + url, + headers=self._build_headers(), + timeout=self.request_timeout, + ) + response.raise_for_status() + + soup = BeautifulSoup(response.text, 'html.parser') + container = soup.find('div', id='app-unifiedResultlist') + + if not container: + logger.warning(f"No job container found for URL: {url}") + return [] + + jobs = [] + seen_links = set() + + for article in container.find_all('article', attrs={'data-testid': 'job-item'}): + # Find all links in the article + all_links = article.find_all('a', href=True) + + job_link = None + job_title = None + + # First, look for job posting links with the correct pattern for link_elem in all_links: href = link_elem.get('href', '') - # Skip company profile links and external links - if '/cmp/' in href or href.startswith('http'): - continue - # Look for job posting links that start with /stellenangebote - if href.startswith('/stellenangebote') and 'inline.html' in href: + + # Check for actual job posting URLs (contain stellenangebote and inline.html) + if re.search(r'/stellenangebote--.*--\d+-inline\.html', href): job_link = link_elem job_title = link_elem.get_text(strip=True) break - - if not job_link: - continue - - # Extract job title from h2/h3 if available, otherwise use link text - if not job_title or len(job_title) < 5: - title_elem = (article.find('h2') or - article.find('h3') or - article.find('span', attrs={'data-testid': re.compile('job-title')})) - if title_elem: - job_title = title_elem.get_text(strip=True) - - title = job_title if job_title and len(job_title) > 0 else "Unknown Title" - - link = job_link['href'] - - # Ensure absolute URL - if not link.startswith("http"): - link = f"https://www.stepstone.de{link}" - - # Skip duplicates and company profile links - if link in seen_links or '/cmp/' in link: - continue - seen_links.add(link) - - # Extract company information - company_elem = (article.find('span', class_=re.compile('company|employer')) or - article.find('a', attrs={'data-testid': re.compile('company|employer')}) or - article.find('span', attrs={'data-testid': re.compile('company|employer')})) - company = company_elem.get_text(strip=True) if company_elem else "Unknown Company" - - # Extract short description - desc_elem = (article.find('p', class_=re.compile('description|snippet|teaser')) or - article.find('div', class_=re.compile('description|snippet|teaser')) or - article.find('span', class_=re.compile('description|snippet|teaser'))) - description = desc_elem.get_text(strip=True)[:200] + "..." if desc_elem and desc_elem.get_text(strip=True) else "No description available" - - jobs.append({ - "title": title, - "company": company, - "description": description, - "link": link - }) - - logger.info(f"Found {len(jobs)} jobs for URL: {url}") - return jobs - - except requests.RequestException as e: - logger.error(f"Request failed for URL {url}: {e}") - return [] - except Exception as e: - logger.error(f"Unexpected error scraping URL {url}: {e}") - return [] + + # If no job posting link found, look for relative links starting with /stellenangebote + if not job_link: + for link_elem in all_links: + href = link_elem.get('href', '') + # Skip company profile links and external links + if '/cmp/' in href or href.startswith('http'): + continue + # Look for job posting links that start with /stellenangebote + if href.startswith('/stellenangebote') and 'inline.html' in href: + job_link = link_elem + job_title = link_elem.get_text(strip=True) + break + + if not job_link: + continue + + # Extract job title from h2/h3 if available, otherwise use link text + if not job_title or len(job_title) < 5: + title_elem = (article.find('h2') or + article.find('h3') or + article.find('span', attrs={'data-testid': re.compile('job-title')})) + if title_elem: + job_title = title_elem.get_text(strip=True) + + title = job_title if job_title and len(job_title) > 0 else "Unknown Title" + + link = job_link['href'] + + # Ensure absolute URL + if not link.startswith("http"): + link = f"https://www.stepstone.de{link}" + + # Skip duplicates and company profile links + if link in seen_links or '/cmp/' in link: + continue + seen_links.add(link) + + # Extract company information + company_elem = (article.find('span', class_=re.compile('company|employer')) or + article.find('a', attrs={'data-testid': re.compile('company|employer')}) or + article.find('span', attrs={'data-testid': re.compile('company|employer')})) + company = company_elem.get_text(strip=True) if company_elem else "Unknown Company" + + # Extract short description + desc_elem = (article.find('p', class_=re.compile('description|snippet|teaser')) or + article.find('div', class_=re.compile('description|snippet|teaser')) or + article.find('span', class_=re.compile('description|snippet|teaser'))) + description = desc_elem.get_text(strip=True)[:200] + "..." if desc_elem and desc_elem.get_text(strip=True) else "No description available" + + jobs.append({ + "title": title, + "company": company, + "description": description, + "link": link + }) + + logger.info(f"Found {len(jobs)} jobs for URL: {url}") + return jobs + + except requests.RequestException as e: + logger.warning(f"Request attempt {attempt} failed for URL {url}: {e}") + if attempt == self.max_retries: + logger.error(f"Exceeded retry attempts for URL {url}") + return [] + backoff = self._calculate_backoff(attempt) + if backoff > 0: + time.sleep(backoff) + except Exception as e: + logger.error(f"Unexpected error scraping URL {url}: {e}") + return [] + + return [] def build_search_url(self, term: str, zip_code: str = "40210", radius: int = 5) -> str: """Build Stepstone search URL""" @@ -349,6 +426,8 @@ async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent if not jobs: formatted_output.append("No jobs found for this search term.") + formatted_output.append("Try adjusting your search terms to broaden the results.") + formatted_output.append("Consider refining your search terms or expanding the radius for more matches.") else: for i, job in enumerate(jobs, 1): formatted_output.append(f"\n{i}. {job['title']}") @@ -387,7 +466,7 @@ async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent )] elif name == "get_job_details": # Extract parameters - query = arguments.get("query") + query = arguments.get("query") or arguments.get("job_query") session_id = arguments.get("session_id") job_index = arguments.get("job_index") @@ -402,7 +481,10 @@ async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent if not query and job_index is None: return [types.TextContent( type="text", - text="Error: provide either a query string or a job_index to identify the job" + text=( + "Error: provide either a query string or a job_index to identify the job " + "(job_query is also accepted)" + ) )] try: diff --git a/tests/test_stepstone_scraper.py b/tests/test_stepstone_scraper.py index 23b3584..87e13d5 100644 --- a/tests/test_stepstone_scraper.py +++ b/tests/test_stepstone_scraper.py @@ -3,11 +3,12 @@ from urllib.parse import unquote import pytest +import requests sys.path.append(str(Path(__file__).resolve().parents[1])) from job_detail_parser import JobDetailParser -from job_details_models import JobDetails +from job_details_models import JobDetails, NetworkError from stepstone_server import StepstoneJobScraper @@ -130,3 +131,95 @@ def test_parse_job_details_extracts_expected_fields(monkeypatch): assert details.contact_info["phone"] == "+49 123 456789" assert details.contact_info["contact_person"] == "Anna Schmidt" assert details.job_url == "https://www.stepstone.de/job/123" + + +class DummyResponse: + def __init__(self, text: str, status_code: int = 200): + self.text = text + self.status_code = status_code + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"HTTP {self.status_code}") + + +def test_fetch_job_listings_applies_throttling_and_backoff(monkeypatch): + html = """ +
+
+ Example Role + Example Corp +

Exciting work

+
+
+ """ + + scraper = StepstoneJobScraper( + min_delay=0.1, + max_delay=0.1, + max_retries=3, + backoff_factor=2.0, + user_agents=("UA-1", "UA-2"), + ) + + sleep_calls = [] + + def fake_sleep(duration): + sleep_calls.append(duration) + + monkeypatch.setattr("stepstone_server.time.sleep", fake_sleep) + monkeypatch.setattr("stepstone_server.random.uniform", lambda a, b: 0.1) + + headers_seen = [] + call_count = {"value": 0} + + def fake_get(self, url, headers, timeout): + call_count["value"] += 1 + headers_seen.append(headers["User-Agent"]) + if call_count["value"] < 3: + raise requests.RequestException("temporary error") + return DummyResponse(html) + + monkeypatch.setattr(requests.Session, "get", fake_get, raising=False) + + results = scraper.fetch_job_listings("https://example.com/jobs") + + assert len(results) == 1 + assert call_count["value"] == 3 + assert headers_seen == ["UA-1", "UA-2", "UA-1"] + assert sleep_calls == [pytest.approx(0.1), pytest.approx(0.2), pytest.approx(0.1), pytest.approx(0.4), pytest.approx(0.1)] + + +def test_fetch_job_page_uses_backoff_and_rotating_agents(monkeypatch): + parser = JobDetailParser( + min_delay=0.1, + max_delay=0.1, + max_retries=2, + backoff_factor=2.0, + user_agents=("UA-1", "UA-2"), + ) + + sleep_calls = [] + + def fake_sleep(duration): + sleep_calls.append(duration) + + monkeypatch.setattr("job_detail_parser.time.sleep", fake_sleep) + monkeypatch.setattr("job_detail_parser.random.uniform", lambda a, b: 0.1) + + headers_seen = [] + call_count = {"value": 0} + + def fake_get(self, url, headers, timeout): + call_count["value"] += 1 + headers_seen.append(headers["User-Agent"]) + raise requests.RequestException("failure") + + monkeypatch.setattr(requests.Session, "get", fake_get, raising=False) + + with pytest.raises(NetworkError): + parser.fetch_job_page("https://example.com/job") + + assert call_count["value"] == 2 + assert headers_seen == ["UA-1", "UA-2"] + assert sleep_calls == [pytest.approx(0.1), pytest.approx(0.2), pytest.approx(0.1)]