new scrapping

data/nsbe_companies.csv

@@ -0,0 +1,7 @@
+company,careers_url,ats_type,ats_identifier,priority
+Stripe,https://stripe.com/jobs/search,greenhouse,stripe,10
+Netflix,https://jobs.netflix.com/,lever,netflix,10
+Capital One,https://www.capitalonecareers.com/,workday,,9
+Chevron,https://chevron.wd5.myworkdayjobs.com/ChevronCareers,workday,,8
+ServiceNow,https://careers.servicenow.com/careers,smartrecruiters,,8
+ABB,https://global.abb/group/en/careers,generic,,7

src/cache.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterable, List, Sequence
+
+from src.models import JobPosting
+
+
+def load_cached_jobs(cache_path: str | Path) -> dict:
+    path = Path(cache_path)
+    if not path.exists():
+        return {"generated_at": "", "companies": [], "jobs": []}
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return {"generated_at": "", "companies": [], "jobs": []}
+
+
+def write_cached_jobs(cache_path: str | Path, jobs: Sequence[JobPosting], companies: Iterable[str]) -> Path:
+    path = Path(cache_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "companies": sorted(set(companies)),
+        "jobs": [job.to_dict() for job in jobs],
+    }
+    path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    return path
+
+
+def merge_cached_jobs(cache_path: str | Path, refreshed_jobs: Sequence[JobPosting], refreshed_companies: Iterable[str]) -> Path:
+    cached = load_cached_jobs(cache_path)
+    refreshed_set = set(refreshed_companies)
+    retained_jobs = [job for job in cached.get("jobs", []) if job.get("company") not in refreshed_set]
+    refreshed_payload = [job.to_dict() for job in refreshed_jobs]
+    merged_companies = set(cached.get("companies", [])) - refreshed_set
+    merged_companies.update(refreshed_set)
+
+    payload = {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "companies": sorted(merged_companies),
+        "jobs": retained_jobs + refreshed_payload,
+    }
+
+    path = Path(cache_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    return path

src/collect_jobs.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable, List, Optional
+
+from src import cache
+from src.collectors import generic, greenhouse, lever, smartrecruiters, workday
+from src.collectors.common import dedupe_jobs
+from src.company_loader import load_company_records
+from src.detectors.ats_detector import detect_ats_type, extract_ats_identifier
+from src.jobs.debug_utils import save_debug_html
+from src.models import CompanyRecord, JobPosting
+from src.resolver.jobs_page_resolver import ResolvedJobsPage, resolve_real_jobs_page
+
+LOGGER = logging.getLogger("career_fair_matcher.collect_jobs")
+
+CONNECTOR_REGISTRY = {
+    "greenhouse": greenhouse.collect,
+    "lever": lever.collect,
+    "workday": workday.collect,
+    "smartrecruiters": smartrecruiters.collect,
+    "generic": generic.collect,
+}
+
+
+@dataclass
+class CompanyCollectionOutcome:
+    company: str
+    original_url: str
+    resolved_url: str
+    fetch_method: str
+    ats_detected: str
+    connector_used: str
+    jobs_collected: int
+    failure_reason: str = ""
+
+
+def configure_logging(level: int = logging.INFO) -> None:
+    if LOGGER.handlers:
+        return
+    logging.basicConfig(level=level, format="%(message)s")
+
+
+def _log_company_event(outcome: CompanyCollectionOutcome) -> None:
+    LOGGER.info(json.dumps({
+        "company": outcome.company,
+        "original_careers_url": outcome.original_url,
+        "resolved_url": outcome.resolved_url,
+        "fetch_method": outcome.fetch_method,
+        "ats_detected": outcome.ats_detected,
+        "connector_used": outcome.connector_used,
+        "jobs_collected": outcome.jobs_collected,
+        "failure_reason": outcome.failure_reason,
+    }))
+
+
+def _save_resolution_snapshots(debug_dir: Path, company: CompanyRecord, resolved_page: ResolvedJobsPage) -> None:
+    for stage, html in resolved_page.html_snapshots.items():
+        save_debug_html(company.company, html, stage, debug_dir)
+    save_debug_html(company.company, resolved_page.html, "resolved", debug_dir)
+
+
+def _collect_from_connector(company: CompanyRecord, resolved_page: ResolvedJobsPage, ats_type: str) -> tuple[List[JobPosting], str, str]:
+    connector_name = ats_type if ats_type in CONNECTOR_REGISTRY else "generic"
+    connector = CONNECTOR_REGISTRY[connector_name]
+    jobs = connector(company, resolved_page)
+
+    failure_reason = ""
+    if not jobs and connector_name != "generic":
+        jobs = generic.collect(company, resolved_page)
+        failure_reason = "ATS_PARSE_FAILED" if not jobs else ""
+        connector_name = "generic"
+
+    if not jobs:
+        resolution_failure = resolved_page.failure_reason or "NO_JOBS_FOUND"
+        return [], connector_name, resolution_failure
+    return dedupe_jobs(jobs), connector_name, failure_reason
+
+
+def collect_jobs_for_company(company: CompanyRecord, debug_dir: Path) -> tuple[List[JobPosting], CompanyCollectionOutcome]:
+    """Resolve, detect, collect, normalize, and log a single curated company."""
+    resolved_page = resolve_real_jobs_page(company.careers_url)
+    _save_resolution_snapshots(debug_dir, company, resolved_page)
+
+    ats_type = detect_ats_type(resolved_page.url, resolved_page.html, company.ats_type)
+    if not company.ats_identifier:
+        company.ats_identifier = extract_ats_identifier(ats_type, resolved_page.url, resolved_page.html)
+
+    jobs, connector_used, failure_reason = _collect_from_connector(company, resolved_page, ats_type)
+    normalized_jobs = [
+        job if isinstance(job, JobPosting) else job
+        for job in jobs
+    ]
+    outcome = CompanyCollectionOutcome(
+        company=company.company,
+        original_url=company.careers_url,
+        resolved_url=resolved_page.url,
+        fetch_method=resolved_page.fetch_method,
+        ats_detected=ats_type,
+        connector_used=connector_used,
+        jobs_collected=len(normalized_jobs),
+        failure_reason=failure_reason,
+    )
+    _log_company_event(outcome)
+    return normalized_jobs, outcome
+
+
+def refresh_selected_companies(
+    *,
+    company_csv: str | Path,
+    cache_path: str | Path,
+    debug_dir: str | Path,
+    selected_companies: Optional[Iterable[str]] = None,
+    limit: Optional[int] = None,
+) -> tuple[List[JobPosting], List[CompanyCollectionOutcome], Path]:
+    """Refresh cached jobs for the curated list, preserving partial results on failure."""
+    configure_logging()
+    debug_path = Path(debug_dir)
+    companies = load_company_records(company_csv, limit=limit, selected_companies=set(selected_companies or []))
+
+    all_jobs: List[JobPosting] = []
+    outcomes: List[CompanyCollectionOutcome] = []
+
+    for company in companies:
+        try:
+            company_jobs, outcome = collect_jobs_for_company(company, debug_path)
+            all_jobs.extend(company_jobs)
+            outcomes.append(outcome)
+        except Exception as exc:
+            outcome = CompanyCollectionOutcome(
+                company=company.company,
+                original_url=company.careers_url,
+                resolved_url=company.careers_url,
+                fetch_method="requests",
+                ats_detected=company.ats_type or "generic",
+                connector_used="generic",
+                jobs_collected=0,
+                failure_reason=str(exc) or "REQUEST_FAILED",
+            )
+            outcomes.append(outcome)
+            _log_company_event(outcome)
+
+    refreshed_companies = [company.company for company in companies]
+    cache_file = cache.merge_cached_jobs(cache_path, dedupe_jobs(all_jobs), refreshed_companies)
+    return dedupe_jobs(all_jobs), outcomes, cache_file
+
+
+def main() -> None:
+    base_dir = Path(__file__).resolve().parent.parent
+    company_csv = base_dir / "data" / "nsbe_companies.csv"
+    cache_path = base_dir / "data" / "cached_jobs.json"
+    debug_dir = base_dir / "debug_html" / "collect_jobs"
+    refresh_selected_companies(company_csv=company_csv, cache_path=cache_path, debug_dir=debug_dir)
+
+
+if __name__ == "__main__":
+    main()

src/collectors/__init__.py

@@ -0,0 +1 @@
+"""ATS-specific and generic job collectors."""

src/collectors/common.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from typing import Iterable, List
+
+from src.models import CompanyRecord, JobPosting
+
+ENTRY_LEVEL_SIGNALS = [
+    "new grad",
+    "university",
+    "early career",
+    "associate",
+    "entry level",
+    "campus",
+    "intern",
+    "internship",
+]
+
+SENIOR_SIGNALS = ["senior", "staff", "principal", "manager", "director"]
+
+REMOTE_SIGNALS = ["remote", "hybrid remote", "work from home"]
+
+
+def is_remote_role(*texts: str) -> bool:
+    blob = " ".join(texts).lower()
+    return any(signal in blob for signal in REMOTE_SIGNALS)
+
+
+def is_entry_level_role(*texts: str) -> bool:
+    blob = " ".join(texts).lower()
+    if any(signal in blob for signal in SENIOR_SIGNALS):
+        return False
+    return any(signal in blob for signal in ENTRY_LEVEL_SIGNALS)
+
+
+def normalize_job_posting(
+    company_record: CompanyRecord,
+    *,
+    title: str,
+    location: str,
+    job_url: str,
+    description: str,
+    department: str = "",
+    source_ats: str = "generic",
+    resolved_url: str = "",
+    employment_type: str = "",
+    posted_date: str = "",
+    raw_payload: dict | None = None,
+) -> JobPosting:
+    """Normalize connector output into the shared JobPosting schema."""
+    return JobPosting(
+        company=company_record.company,
+        title=title.strip()[:160],
+        location=location.strip()[:160],
+        url=job_url.strip(),
+        department=department.strip()[:160],
+        description=description.strip()[:4000],
+        ats=source_ats,
+        source_ats=source_ats,
+        resolved_url=resolved_url or company_record.careers_url,
+        employment_type=employment_type.strip()[:80],
+        posted_date=posted_date.strip()[:80],
+        is_remote=is_remote_role(title, location, description),
+        is_entry_level=is_entry_level_role(title, description),
+        raw_payload=raw_payload or {},
+    )
+
+
+def dedupe_jobs(jobs: Iterable[JobPosting]) -> List[JobPosting]:
+    """Deduplicate jobs by normalized title and URL."""
+    seen = set()
+    deduped: List[JobPosting] = []
+    for job in jobs:
+        key = (job.title.strip().lower(), job.url.strip().lower())
+        if key in seen:
+            continue
+        seen.add(key)
+        deduped.append(job)
+    return deduped

src/collectors/generic.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List
+
+from src.collectors.common import dedupe_jobs, normalize_job_posting
+from src.jobs.extractor import ExtractionDiagnostics, extract_jobs_with_diagnostics
+from src.models import CompanyRecord, JobPosting
+from src.resolver.jobs_page_resolver import ResolvedJobsPage
+
+
+@dataclass
+class GenericCollectionDetails:
+    jobs: List[JobPosting] = field(default_factory=list)
+    failure_reason: str = "UNKNOWN"
+    diagnostics: ExtractionDiagnostics | None = None
+
+
+def collect(company_record: CompanyRecord, resolved_page: ResolvedJobsPage) -> List[JobPosting]:
+    """Generic fallback that parses resolved HTML and embedded JSON job data."""
+    return collect_with_details(company_record, resolved_page).jobs
+
+
+def collect_with_details(
+    company_record: CompanyRecord,
+    resolved_page: ResolvedJobsPage,
+    source_ats: str = "generic",
+) -> GenericCollectionDetails:
+    parsed_jobs, diagnostics = extract_jobs_with_diagnostics(
+        company_record,
+        resolved_page.html,
+        source_ats,
+        base_url=resolved_page.url,
+    )
+    normalized = [
+        normalize_job_posting(
+            company_record,
+            title=job.title,
+            location=job.location,
+            job_url=job.url,
+            description=job.description,
+            department=job.department,
+            source_ats=source_ats,
+            resolved_url=resolved_page.url,
+            raw_payload=job.raw_payload,
+        )
+        for job in parsed_jobs
+    ]
+    return GenericCollectionDetails(
+        jobs=dedupe_jobs(normalized),
+        failure_reason=diagnostics.failure_type,
+        diagnostics=diagnostics,
+    )

src/collectors/greenhouse.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from typing import List
+
+import requests
+
+from src.collectors.common import dedupe_jobs, normalize_job_posting
+from src.detectors.ats_detector import extract_ats_identifier
+from src.models import CompanyRecord, JobPosting
+from src.resolver.jobs_page_resolver import ResolvedJobsPage
+
+HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
+
+
+def collect(company_record: CompanyRecord, resolved_page: ResolvedJobsPage) -> List[JobPosting]:
+    """Collect Greenhouse jobs via the public boards API."""
+    identifier = company_record.ats_identifier or extract_ats_identifier("greenhouse", resolved_page.url, resolved_page.html)
+    if not identifier:
+        return []
+
+    api_url = f"https://boards-api.greenhouse.io/v1/boards/{identifier}/jobs?content=true"
+    try:
+        response = requests.get(api_url, headers=HEADERS, timeout=20)
+        response.raise_for_status()
+        payload = response.json()
+    except Exception:
+        return []
+
+    jobs = [
+        normalize_job_posting(
+            company_record,
+            title=item.get("title", ""),
+            location=((item.get("location") or {}).get("name") or ""),
+            job_url=item.get("absolute_url", resolved_page.url),
+            description=(item.get("content") or ""),
+            department=(item.get("department") or ""),
+            source_ats="greenhouse",
+            resolved_url=resolved_page.url,
+            posted_date=str(item.get("updated_at") or ""),
+            raw_payload=item,
+        )
+        for item in payload.get("jobs", [])
+        if item.get("title")
+    ]
+    return dedupe_jobs(jobs)

src/collectors/lever.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import List
+
+import requests
+
+from src.collectors.common import dedupe_jobs, normalize_job_posting
+from src.detectors.ats_detector import extract_ats_identifier
+from src.models import CompanyRecord, JobPosting
+from src.resolver.jobs_page_resolver import ResolvedJobsPage
+
+HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
+
+
+def collect(company_record: CompanyRecord, resolved_page: ResolvedJobsPage) -> List[JobPosting]:
+    """Collect Lever jobs via the public postings endpoint."""
+    identifier = company_record.ats_identifier or extract_ats_identifier("lever", resolved_page.url, resolved_page.html)
+    if not identifier:
+        return []
+
+    api_url = f"https://api.lever.co/v0/postings/{identifier}?mode=json"
+    try:
+        response = requests.get(api_url, headers=HEADERS, timeout=20)
+        response.raise_for_status()
+        payload = response.json()
+    except Exception:
+        return []
+
+    jobs = []
+    for item in payload:
+        categories = item.get("categories") or {}
+        jobs.append(
+            normalize_job_posting(
+                company_record,
+                title=item.get("text", ""),
+                location=categories.get("location", ""),
+                job_url=item.get("hostedUrl", resolved_page.url),
+                description=item.get("descriptionPlain", ""),
+                department=categories.get("team", ""),
+                source_ats="lever",
+                resolved_url=resolved_page.url,
+                employment_type=categories.get("commitment", ""),
+                raw_payload=item,
+            )
+        )
+    return dedupe_jobs(jobs)

src/collectors/smartrecruiters.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import List
+
+import requests
+
+from src.collectors.common import dedupe_jobs, normalize_job_posting
+from src.detectors.ats_detector import extract_ats_identifier
+from src.models import CompanyRecord, JobPosting
+from src.resolver.jobs_page_resolver import ResolvedJobsPage
+
+HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
+
+
+def collect(company_record: CompanyRecord, resolved_page: ResolvedJobsPage) -> List[JobPosting]:
+    """Collect SmartRecruiters jobs when a company identifier is available."""
+    identifier = company_record.ats_identifier or extract_ats_identifier("smartrecruiters", resolved_page.url, resolved_page.html)
+    if not identifier:
+        return []
+
+    api_url = f"https://api.smartrecruiters.com/v1/companies/{identifier}/postings"
+    try:
+        response = requests.get(api_url, headers=HEADERS, timeout=20)
+        response.raise_for_status()
+        payload = response.json()
+    except Exception:
+        return []
+
+    jobs = []
+    for item in payload.get("content", []):
+        jobs.append(
+            normalize_job_posting(
+                company_record,
+                title=item.get("name", ""),
+                location=(item.get("location") or {}).get("city", ""),
+                job_url=item.get("ref", resolved_page.url),
+                description=item.get("jobAd", {}).get("sections", "") if isinstance(item.get("jobAd"), dict) else "",
+                department=item.get("department", ""),
+                source_ats="smartrecruiters",
+                resolved_url=resolved_page.url,
+                employment_type=item.get("typeOfEmployment", ""),
+                posted_date=str(item.get("releasedDate") or ""),
+                raw_payload=item,
+            )
+        )
+    return dedupe_jobs(jobs)

src/collectors/workday.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from typing import List
+
+from src.collectors.common import dedupe_jobs
+from src.collectors.generic import collect_with_details
+from src.models import CompanyRecord, JobPosting
+from src.resolver.jobs_page_resolver import ResolvedJobsPage
+
+
+def collect(company_record: CompanyRecord, resolved_page: ResolvedJobsPage) -> List[JobPosting]:
+    """Best-effort Workday collection.
+
+    Workday endpoints vary widely by tenant, so this connector currently relies on the
+    resolved HTML plus embedded JSON extraction until a tenant-specific endpoint is configured.
+    """
+    details = collect_with_details(company_record, resolved_page, source_ats="workday")
+    return dedupe_jobs(details.jobs)

src/company_loader.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+from typing import Iterable, List, Optional, Set
+
+from src.models import CompanyRecord
+
+COMPANY_KEYS = ["company", "company list", "name"]
+CAREERS_KEYS = ["careers_url", "career url", "jobs_url", "direct links to company career/job openings page"]
+
+
+def _pick_value(row: dict[str, str], keys: Iterable[str]) -> str:
+    for key in keys:
+        value = row.get(key, "").strip()
+        if value:
+            return value
+    return ""
+
+
+def _parse_priority(value: str) -> int:
+    try:
+        return int(value.strip()) if value.strip() else 0
+    except Exception:
+        return 0
+
+
+def load_company_records(
+    csv_path: str | Path,
+    *,
+    limit: Optional[int] = None,
+    selected_companies: Optional[Set[str]] = None,
+) -> List[CompanyRecord]:
+    """Load the curated company list used by the targeted job collector."""
+    path = Path(csv_path)
+    if not path.exists():
+        raise FileNotFoundError(f"Company CSV not found: {path}")
+
+    companies: List[CompanyRecord] = []
+    with path.open("r", encoding="utf-8-sig", newline="") as handle:
+        reader = csv.DictReader(handle)
+        for raw_row in reader:
+            row = {str(key).strip().lower(): str(value or "").strip() for key, value in raw_row.items() if key}
+            company = _pick_value(row, COMPANY_KEYS)
+            careers_url = _pick_value(row, CAREERS_KEYS)
+            if not company or not careers_url:
+                continue
+            if selected_companies and company not in selected_companies:
+                continue
+
+            companies.append(
+                CompanyRecord(
+                    company=company,
+                    careers_url=careers_url,
+                    ats_type=row.get("ats_type", ""),
+                    ats_identifier=row.get("ats_identifier", ""),
+                    priority=_parse_priority(row.get("priority", "0")),
+                    source="curated",
+                    meta=row,
+                )
+            )
+
+    return companies[:limit] if limit else companies

src/detectors/__init__.py

@@ -0,0 +1 @@
+"""ATS detection helpers."""

src/detectors/ats_detector.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import re
+from urllib.parse import parse_qs, urlparse
+
+KNOWN_ATS = {
+    "greenhouse": ["greenhouse.io"],
+    "lever": ["lever.co", "jobs.lever.co"],
+    "workday": ["myworkdayjobs.com", "workday.com"],
+    "smartrecruiters": ["smartrecruiters.com"],
+    "icims": ["icims.com"],
+    "ashby": ["ashbyhq.com", "jobs.ashbyhq.com"],
+    "successfactors": ["successfactors.com", "career8.successfactors.com"],
+}
+
+
+def normalize_ats_type(value: str) -> str:
+    lowered = (value or "").strip().lower()
+    if lowered in KNOWN_ATS:
+        return lowered
+    return "generic"
+
+
+def detect_ats_type(url: str, html: str = "", declared_ats: str = "") -> str:
+    """Detect the ATS provider from explicit config, resolved URL, or HTML content."""
+    if declared_ats:
+        return normalize_ats_type(declared_ats)
+
+    blob = f"{url} {html}".lower()
+    for ats_type, patterns in KNOWN_ATS.items():
+        if any(pattern in blob for pattern in patterns):
+            return ats_type
+
+    return "generic"
+
+
+def extract_ats_identifier(ats_type: str, url: str, html: str = "") -> str:
+    """Best-effort extraction of ATS board identifiers for connector API usage."""
+    patterns = {
+        "greenhouse": r"greenhouse\.io/([^/?#]+)",
+        "lever": r"lever\.co/([^/?#]+)",
+        "smartrecruiters": r"smartrecruiters\.com/([^/?#]+)",
+        "ashby": r"ashbyhq\.com/([^/?#]+)",
+    }
+    if ats_type in patterns:
+        match = re.search(patterns[ats_type], f"{url} {html}")
+        return match.group(1) if match else ""
+
+    if ats_type == "successfactors":
+        parsed = urlparse(url)
+        query = parse_qs(parsed.query)
+        values = query.get("company") or query.get("_s.crb") or []
+        return values[0] if values else ""
+
+    if ats_type == "workday":
+        parsed = urlparse(url)
+        parts = [part for part in parsed.path.split("/") if part]
+        return "/".join(parts[:3]) if parts else ""
+
+    return ""

src/models.py

@@ -1,5 +1,5 @@
-from dataclasses import dataclass, field
-from typing import Dict, List
+from dataclasses import asdict, dataclass, field
+from typing import Any, Dict, List
 
 
 @dataclass
@@ -19,6 +19,9 @@ class CompanyRecord:
     careers_url: str = ""
     source: str = "default"
     meta: Dict[str, str] = field(default_factory=dict)
+    ats_type: str = ""
+    ats_identifier: str = ""
+    priority: int = 0
 
 
 @dataclass
@@ -30,6 +33,20 @@ class JobPosting:
     department: str = ""
     description: str = ""
     ats: str = "unknown"
+    source_ats: str = "unknown"
+    resolved_url: str = ""
+    employment_type: str = ""
+    posted_date: str = ""
+    is_remote: bool = False
+    is_entry_level: bool = False
+    failure_reason: str = ""
+    raw_payload: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        payload = asdict(self)
+        payload["job_url"] = payload.pop("url")
+        payload["source_ats"] = payload.get("source_ats") or payload.get("ats", "unknown")
+        return payload
 
 
 @dataclass

src/resolver/__init__.py

@@ -0,0 +1 @@
+"""Jobs page resolution utilities."""

src/resolver/jobs_page_resolver.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import importlib
+import re
+from dataclasses import dataclass, field
+from typing import Dict, List
+from urllib.parse import urljoin
+
+import requests
+from bs4 import BeautifulSoup
+
+HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
+ATS_LINK_PATTERNS = [
+    "greenhouse.io",
+    "jobs.lever.co",
+    "myworkdayjobs.com",
+    "smartrecruiters.com",
+    "icims.com",
+    "ashbyhq.com",
+    "successfactors.com",
+]
+JOB_CTA_HINTS = [
+    "jobs",
+    "job search",
+    "search jobs",
+    "view all jobs",
+    "open positions",
+    "current openings",
+    "careers search",
+    "all openings",
+    "explore jobs",
+]
+
+
+@dataclass
+class ResolvedJobsPage:
+    requested_url: str
+    url: str
+    html: str
+    fetch_method: str
+    final_url: str = ""
+    fallback_used: bool = False
+    failure_reason: str = ""
+    resolution_steps: List[str] = field(default_factory=list)
+    html_snapshots: Dict[str, str] = field(default_factory=dict)
+
+
+def find_ats_link(html: str, base_url: str) -> str:
+    """Return the first ATS-hosted link or iframe source found in the page."""
+    soup = BeautifulSoup(html or "", "html.parser")
+    for tag in soup.select("a[href], iframe[src]"):
+        target = tag.get("href") or tag.get("src") or ""
+        resolved = urljoin(base_url, target.strip())
+        if any(pattern in resolved.lower() for pattern in ATS_LINK_PATTERNS):
+            return resolved
+    return ""
+
+
+def find_redirect_url(html: str, base_url: str) -> str:
+    """Handle meta refresh and simple JavaScript redirects."""
+    soup = BeautifulSoup(html or "", "html.parser")
+    meta_refresh = soup.find("meta", attrs={"http-equiv": re.compile(r"refresh", re.I)})
+    if meta_refresh:
+        content = meta_refresh.get("content", "")
+        match = re.search(r"url\s*=\s*([^;]+)", content, re.I)
+        if match:
+            return urljoin(base_url, match.group(1).strip().strip("\"'"))
+
+    for pattern in [
+        r'location\.href\s*=\s*["\']([^"\']+)["\']',
+        r'window\.location\s*=\s*["\']([^"\']+)["\']',
+        r'window\.location\.href\s*=\s*["\']([^"\']+)["\']',
+    ]:
+        match = re.search(pattern, html or "", re.I)
+        if match:
+            return urljoin(base_url, match.group(1).strip())
+    return ""
+
+
+def find_job_list_url(html: str, base_url: str) -> str:
+    """Find a likely jobs listing URL from page CTAs, anchors, buttons, or iframes."""
+    soup = BeautifulSoup(html or "", "html.parser")
+    for tag in soup.select("a[href], button, iframe[src], [data-href], [formaction]"):
+        text = " ".join(tag.get_text(" ", strip=True).split()).lower()
+        target = tag.get("href") or tag.get("src") or tag.get("data-href") or tag.get("formaction") or ""
+        blob = f"{text} {target}".lower()
+        if any(hint in blob for hint in JOB_CTA_HINTS):
+            return urljoin(base_url, target.strip()) if target else ""
+    return ""
+
+
+def looks_like_shell_page(html: str) -> bool:
+    """Detect shell/search pages that do not yet expose job content."""
+    lowered = (html or "").lower()
+    if not lowered:
+        return True
+    shell_markers = ["search jobs", "view jobs", "career search", "keyword", "join our talent community"]
+    job_markers = ["apply now", "job id", "req id", "posted", "department"]
+    return sum(marker in lowered for marker in shell_markers) >= 2 and sum(marker in lowered for marker in job_markers) <= 1
+
+
+def looks_js_heavy(html: str) -> bool:
+    """Detect pages that likely require browser execution."""
+    lowered = (html or "").lower()
+    if len(lowered) < 1500:
+        return True
+    return any(marker in lowered for marker in ["enable javascript", "loading", "__next", "app-root", "hydration"])
+
+
+def _request_page(url: str, timeout: int = 12) -> ResolvedJobsPage:
+    response = requests.get(url, headers=HEADERS, timeout=timeout)
+    response.raise_for_status()
+    resolved_url = response.url or url
+    return ResolvedJobsPage(
+        requested_url=url,
+        url=resolved_url,
+        final_url=resolved_url,
+        html=response.text,
+        fetch_method="requests",
+        resolution_steps=[f"requests:{url} -> {resolved_url}"],
+        html_snapshots={"requests": response.text},
+    )
+
+
+def _playwright_resolve(url: str) -> ResolvedJobsPage:
+    try:
+        sync_api = importlib.import_module("playwright.sync_api")
+        sync_playwright = getattr(sync_api, "sync_playwright")
+
+        with sync_playwright() as playwright:
+            browser = playwright.chromium.launch(headless=True)
+            page = browser.new_page()
+            page.goto(url, wait_until="domcontentloaded", timeout=20000)
+
+            try:
+                page.wait_for_load_state("networkidle", timeout=5000)
+            except Exception:
+                pass
+
+            for cta in ["View Jobs", "Search Jobs", "Open Positions", "Current Openings", "Explore Jobs", "See All Jobs"]:
+                try:
+                    page.get_by_text(cta, exact=False).first.click(timeout=1500)
+                    page.wait_for_timeout(800)
+                    break
+                except Exception:
+                    continue
+
+            for _ in range(3):
+                page.mouse.wheel(0, 1800)
+                page.wait_for_timeout(400)
+
+            html = page.content()
+            final_url = page.url
+            browser.close()
+    except Exception:
+        return ResolvedJobsPage(requested_url=url, url=url, final_url=url, html="", fetch_method="playwright", fallback_used=True, failure_reason="REQUEST_FAILED")
+
+    return ResolvedJobsPage(
+        requested_url=url,
+        url=final_url,
+        final_url=final_url,
+        html=html,
+        fetch_method="playwright",
+        fallback_used=True,
+        resolution_steps=[f"playwright:{url} -> {final_url}"],
+        html_snapshots={"playwright": html},
+    )
+
+
+def resolve_real_jobs_page(careers_url: str) -> ResolvedJobsPage:
+    """Resolve the company careers landing page to the real jobs page before collection."""
+    if not careers_url:
+        return ResolvedJobsPage(requested_url="", url="", final_url="", html="", fetch_method="none", failure_reason="REQUEST_FAILED")
+
+    visited = set()
+    current_url = careers_url
+    steps: List[str] = []
+    snapshots: Dict[str, str] = {}
+    last_page = ResolvedJobsPage(requested_url=careers_url, url=careers_url, final_url=careers_url, html="", fetch_method="requests")
+
+    for step in range(4):
+        if current_url in visited:
+            break
+        visited.add(current_url)
+
+        try:
+            last_page = _request_page(current_url)
+            steps.extend(last_page.resolution_steps)
+            snapshots[f"step{step + 1}_requests"] = last_page.html
+        except Exception:
+            fallback = _playwright_resolve(current_url)
+            fallback.failure_reason = fallback.failure_reason or "REQUEST_FAILED"
+            fallback.resolution_steps = steps + fallback.resolution_steps
+            fallback.html_snapshots.update(snapshots)
+            return fallback
+
+        next_url = find_redirect_url(last_page.html, last_page.url) or find_ats_link(last_page.html, last_page.url) or find_job_list_url(last_page.html, last_page.url)
+        if next_url and next_url not in visited:
+            steps.append(f"discovered:{last_page.url} -> {next_url}")
+            current_url = next_url
+            continue
+
+        if looks_like_shell_page(last_page.html) or looks_js_heavy(last_page.html):
+            fallback = _playwright_resolve(last_page.url)
+            fallback.failure_reason = fallback.failure_reason or "JS_PAGE"
+            fallback.resolution_steps = steps + fallback.resolution_steps
+            fallback.html_snapshots.update(snapshots)
+            return fallback
+
+        break
+
+    last_page.requested_url = careers_url
+    last_page.resolution_steps = steps
+    last_page.html_snapshots.update(snapshots)
+    if looks_like_shell_page(last_page.html):
+        last_page.failure_reason = "SHELL_PAGE"
+    return last_page