anky2002

feat: Complete project scaffold - backend (FastAPI) + frontend (Next.js 15) + scraper + LLM gateway

8f2ab79 verified 5 months ago

13.3 kB

"""
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()

	"""
	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()