scripts/verify_scrapling_integration.py

"""Offline verification for the Scrapling anti-block integration.

Deterministic, no network: validates the fetch-layer shim + proxy rotation, the
Greenhouse/Lever/Ashby ATS parsers (via canned JSON), graceful degradation when
Scrapling isn't installed, and the ui/config/deps wiring. Run:

    python scripts/verify_scrapling_integration.py
"""
import os
import sys
import json
import types

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))

_ok = True


def check(label, cond, extra=""):
    global _ok
    _ok = _ok and bool(cond)
    print(f"  [{'PASS' if cond else 'FAIL'}] {label}" + (f"  {extra}" if extra else ""))


def read(rel):
    with open(os.path.join(REPO, rel), encoding="utf-8") as f:
        return f.read()


print("=" * 70)
print("Scrapling anti-block integration verification")
print("=" * 70)

# ── [A] fetch layer: shim + graceful import ──────────────────────────────────
print("[A] Fetch layer (requests.Response-compatible shim + graceful degrade)")
from src.scrapers import fetch  # noqa: E402

check("scrapling_available() returns a bool (no import crash)",
      isinstance(fetch.scrapling_available(), bool),
      f"scrapling={fetch.scrapling_available()} stealthy={fetch.stealthy_available()}")

r = fetch.FetchResponse(200, '{"a": 1}', url="http://x")
check("FetchResponse.ok true for 200", r.ok is True)
check("FetchResponse.json() parses body", r.json() == {"a": 1})
check("FetchResponse.content is bytes", isinstance(r.content, bytes))
check("FetchResponse non-200 not ok", fetch.FetchResponse(403, "").ok is False)

# ── [B] proxy rotation (the real anti-IP-block hook) ─────────────────────────
print("[B] Proxy env hook + round-robin rotation")
os.environ["SCRAPER_PROXIES"] = "http://p1:1,http://p2:2"
rotated = [fetch._next_proxy() for _ in range(4)]
check("rotates configured proxies round-robin",
      rotated == ["http://p1:1", "http://p2:2", "http://p1:1", "http://p2:2"],
      str(rotated))
os.environ.pop("SCRAPER_PROXIES", None)
check("no proxy configured -> None", fetch._next_proxy() is None)

# ── [C] ATS parsers (Greenhouse/Lever/Ashby) via canned JSON ─────────────────
print("[C] Company ATS parsers normalize to Job + filter non-PM")
from src.scrapers.company_ats import CompanyATSScraper  # noqa: E402

GH = json.dumps({"jobs": [
    {"id": 1, "title": "Senior Product Manager", "absolute_url": "https://gh/1",
     "updated_at": "2026-06-01T00:00:00Z", "location": {"name": "Bangalore, India"},
     "content": "<p>Own the roadmap</p>"},
    {"id": 2, "title": "Software Engineer", "absolute_url": "https://gh/2",
     "location": {"name": "Remote"}, "content": "x"},
]})
LEVER = json.dumps([
    {"id": "a", "text": "Product Manager, Growth", "hostedUrl": "https://lever/a",
     "categories": {"location": "Remote"}, "descriptionPlain": "Lead growth",
     "createdAt": 1700000000000},
])
ASHBY = json.dumps({"jobs": [
    {"id": "z", "title": "AI Product Manager", "jobUrl": "https://ashby/z",
     "location": "Remote", "descriptionPlain": "Build AI products",
     "publishedAt": "2026-05-01"},
]})


def fake_get(self, url, **kw):
    if "greenhouse" in url:
        return fetch.FetchResponse(200, GH)
    if "lever" in url:
        return fetch.FetchResponse(200, LEVER)
    if "ashby" in url:
        return fetch.FetchResponse(200, ASHBY)
    return None


boards = [
    {"provider": "greenhouse", "token": "x", "company": "GH Co"},
    {"provider": "lever", "token": "y", "company": "Lever Co"},
    {"provider": "ashby", "token": "z", "company": "Ashby Co"},
]
s = CompanyATSScraper(boards=boards)
s._get = types.MethodType(fake_get, s)
jobs = s.search("product manager", "remote", max_results=25)

titles = [j.title for j in jobs]
check("3 PM jobs across 3 providers (engineer filtered out)", len(jobs) == 3, str(titles))
check("non-PM 'Software Engineer' excluded", "Software Engineer" not in titles)
gh = next((j for j in jobs if j.company == "GH Co"), None)
check("greenhouse HTML description stripped + unescaped",
      gh is not None and gh.description == "Own the roadmap",
      gh.description if gh else "none")
lev = next((j for j in jobs if j.company == "Lever Co"), None)
check("lever createdAt(ms) -> YYYY-MM-DD posted_date",
      lev is not None and len(lev.posted_date) == 10 and lev.posted_date.startswith("20"),
      lev.posted_date if lev else "none")
check("all ATS jobs carry platform=CompanyATS + url",
      all(j.platform == "CompanyATS" and j.url for j in jobs))

# location filter rejects mismatched city when a specific location is given
s2 = CompanyATSScraper(boards=[{"provider": "ashby", "token": "z", "company": "Ashby Co"}])
ASHBY_SF = json.dumps({"jobs": [
    {"id": "z", "title": "Product Manager", "jobUrl": "https://ashby/sf",
     "location": "Munich, Germany", "descriptionPlain": "d", "publishedAt": "2026-05-01"},
]})
s2._get = types.MethodType(lambda self, url, **kw: fetch.FetchResponse(200, ASHBY_SF), s2)
check("location filter drops a non-matching city for a specific location",
      len(s2.search("product manager", "Hyderabad, India", 25)) == 0)

# ── [D] base scraper uses the fetch layer ────────────────────────────────────
print("[D] BaseScraper._get routes through the stealth fetch layer")
base_src = read("src/scrapers/base.py")
check("base imports the fetch layer", "from . import fetch" in base_src)
check("base._get calls fetch.get(", "fetch.get(" in base_src)
check("base._get supports use_browser/solve_cloudflare",
      "use_browser" in base_src and "solve_cloudflare" in base_src)

gd_src = read("src/scrapers/glassdoor.py")
check("glassdoor tries StealthyFetcher (Cloudflare) before Playwright",
      "fetch_browser_html" in gd_src)

# ── [E] config + ui + deps wiring ────────────────────────────────────────────
print("[E] config / ui / dependency wiring")
import config  # noqa: E402
check("config.SCRAPER present with proxies + impersonate",
      isinstance(getattr(config, "SCRAPER", None), dict)
      and "proxies" in config.SCRAPER and "impersonate" in config.SCRAPER)
check("config.COMPANY_ATS_BOARDS exists (list)",
      isinstance(getattr(config, "COMPANY_ATS_BOARDS", None), list))

ui_src = read("ui.py")
check("ui adds company_ats to legacy keys (not routed to ever-jobs)",
      '"company_ats"' in ui_src and "_legacy_keys" in ui_src)
check("ui registers CompanyATSScraper", "CompanyATSScraper" in ui_src)

plat_src = read("src/ever_jobs_bridge/platforms.py")
check("platforms registry exposes company_ats option", '"company_ats"' in plat_src)
check("company_ats is an India default platform",
      "company_ats" in plat_src and "INDIA_DEFAULT_PLATFORMS" in plat_src)

reqs = read("requirements.txt")
check("requirements.txt pins scrapling[fetchers]", "scrapling[fetchers]" in reqs)
docker = read("Dockerfile")
check("Dockerfile installs the Camoufox stealth browser", "scrapling install" in docker)

print("-" * 70)
print("PASS - Scrapling integration wired + ATS parsers verified"
      if _ok else "FAIL - see above")
sys.exit(0 if _ok else 1)