Upload main.py

main.py

@@ -0,0 +1,2338 @@
+import asyncio
+import json
+import os
+import random
+import tempfile
+import urllib.parse
+import time
+import logging
+from datetime import datetime, timezone
+import concurrent.futures
+import re
+from typing import Optional
+
+from dotenv import load_dotenv
+load_dotenv()  # loads .env from project root → SERPER_API_KEY etc.
+
+from openai import OpenAI
+
+import httpx
+from fastapi import FastAPI, HTTPException, Query, BackgroundTasks
+from pydantic import BaseModel
+from starlette.responses import StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+from bs4 import BeautifulSoup
+
+from school_classifier import classify_job, SCHOOLS, PROGRAM_KEYWORDS
+from locations import INDIAN_METROS
+from database import save_scraped_jobs, get_jobs_by_timeframe, get_jobs_in_timeframe, cleanup_old_jobs, get_binned_jobs, update_company_ratings, _connect
+
+# Primary engine: Selenium with undetected-chromedriver
+import undetected_chromedriver as uc
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.common.by import By
+
+# Fallback engine: Playwright with stealth
+from playwright.async_api import async_playwright
+from playwright_stealth import Stealth
+
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.interval import IntervalTrigger
+
+scheduler = AsyncIOScheduler()
+
+# ── Logging setup ─────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("internscrapper")
+logger.setLevel(logging.INFO)
+
+app = FastAPI(title="LinkedIn Public Job Scraper")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ── Shared resources ──────────────────────────────────────────
+class ScrapeSession:
+    def __init__(self):
+        self.history: list[str] = []
+        self.queues: list[asyncio.Queue] = []
+        self.is_active = False
+        self.task: Optional[asyncio.Task] = None
+
+    def cancel(self):
+        self.is_active = False
+        if self.task and not self.task.done():
+            self.task.cancel()
+
+global_scrape = ScrapeSession()
+
+executor = concurrent.futures.ThreadPoolExecutor(max_workers=6)
+_scrape_lock = asyncio.Lock()
+
+# ── Auto-scrape state (updated live so the admin UI can poll progress) ────
+_auto_scrape_state: dict = {
+    "is_running": False,
+    "current_school": None,
+    "completed": [],
+    "failed": [],
+    "started_at": None,
+    "finished_at": None,
+    "cancel_requested": False,
+}
+_auto_scrape_task: Optional[asyncio.Task] = None
+
+# ── Auto-scrape history (in-memory log of all sweep runs) ─────
+_auto_scrape_history: list[dict] = []
+
+# HuggingFace Spaces sets PORT=7860 in its environment.
+# Locally, uvicorn defaults to 8000 unless overridden.
+_SELF_PORT = int(os.environ.get("PORT", "8000"))
+_SELF_BASE_URL = f"http://localhost:{_SELF_PORT}"
+
+se_driver = None           # Selenium undetected-chromedriver
+pw_browser = None          # Playwright browser
+pw_stealth_ctx = None      # Playwright stealth context manager
+
+# Cookie warm-up cache — browser-derived cookies with a TTL.
+_warm_cookies: dict = {}
+_warm_cookies_ts: float = 0.0
+_COOKIE_TTL_SECONDS = 600  # 10-minute TTL before re-warming
+
+
+# ── Lifecycle ─────────────────────────────────────────────────
+
+def _get_chrome_major_version() -> Optional[int]:
+    """Helper to detect the installed Chrome major version on Windows/Linux to prevent driver mismatch."""
+    try:
+        import winreg
+        # Check user-level install / BLBeacon first
+        try:
+            with winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Google\Chrome\BLBeacon") as key:
+                version, _ = winreg.QueryValueEx(key, "version")
+                return int(version.split(".")[0])
+        except Exception:
+            pass
+        # Check system-level WOW64 install
+        try:
+            with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Wow6432Node\Microsoft\Windows\CurrentVersion\Uninstall\Google Chrome") as key:
+                version, _ = winreg.QueryValueEx(key, "DisplayVersion")
+                return int(version.split(".")[0])
+        except Exception:
+            pass
+        # Check system-level 64-bit install
+        try:
+            with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\Google Chrome") as key:
+                version, _ = winreg.QueryValueEx(key, "DisplayVersion")
+                return int(version.split(".")[0])
+        except Exception:
+            pass
+    except ImportError:
+        # Linux/macOS
+        import subprocess
+        import re
+        for cmd in ["google-chrome", "google-chrome-stable", "chromium", "chromium-browser"]:
+            try:
+                output = subprocess.check_output([cmd, "--version"], stderr=subprocess.DEVNULL)
+                version_str = output.decode("utf-8").strip()
+                match = re.search(r"(\d+)\.", version_str)
+                if match:
+                    return int(match.group(1))
+            except Exception:
+                continue
+    return None
+
+
+# ── User-Agent helpers ────────────────────────────────────────
+
+def _build_user_agent(chrome_version: Optional[int] = None) -> str:
+    """Build a realistic Chrome User-Agent string using the installed version."""
+    v = chrome_version or _get_chrome_major_version() or 131
+    return (
+        f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        f"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36"
+    )
+
+
+def _build_ua_pool() -> list[str]:
+    """Build a pool of realistic User-Agent strings for rotation."""
+    v = _get_chrome_major_version() or 131
+    return [
+        # Windows Chrome (primary — matches our Selenium fingerprint)
+        f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36",
+        # Windows Chrome (slightly older minor)
+        f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v-1}.0.0.0 Safari/537.36",
+        # macOS Chrome
+        f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36",
+        # Linux Chrome
+        f"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36",
+        # Windows Edge (Chromium-based, same engine)
+        f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36 Edg/{v}.0.0.0",
+    ]
+
+
+_UA_POOL: list[str] = []  # populated at startup
+
+
+# ── Response validation helpers ───────────────────────────────
+
+def _is_empty_stub(html: str) -> bool:
+    """Detect LinkedIn's known empty/blocked response patterns.
+
+    Returns True if the response is the 26-byte empty stub, a login wall,
+    a CAPTCHA page, or any other pattern that indicates zero real content.
+    """
+    stripped = html.strip()
+    # 26-byte empty stub: '<!DOCTYPE html>  <!---->  '
+    if len(stripped) < 60 and "<!--" in stripped:
+        return True
+    # Login/signup redirect pages
+    lower = stripped[:2000].lower()
+    if any(sig in lower for sig in [
+        "login", "sign in", "signup", "sign up",
+        "authwall", "auth_wall", "checkpoint",
+    ]):
+        return True
+    return False
+
+
+def _create_selenium_driver():
+    """Create an undetected Chrome instance (runs in thread pool)."""
+    options = uc.ChromeOptions()
+    options.add_argument("--headless=new")
+    options.add_argument("--no-sandbox")
+    options.add_argument("--disable-dev-shm-usage")
+    options.add_argument("--disable-gpu")
+    options.add_argument("--disable-blink-features=AutomationControlled")
+    options.add_argument("--window-size=1920,1080")
+    options.add_argument(f"user-agent={_build_user_agent()}")
+    version_main = _get_chrome_major_version()
+    if version_main:
+        driver = uc.Chrome(options=options, version_main=version_main)
+    else:
+        driver = uc.Chrome(options=options)
+    driver.set_page_load_timeout(30)
+    return driver
+
+_AUTO_SCRAPE_SCHOOLS = ["socse", "sob", "sodi", "soepp", "solaw", "sofmca", "solas", "soahp"]
+
+async def _auto_scrape_all_schools():
+    global _auto_scrape_state, _auto_scrape_history
+    logger.info("⏰ Auto-scrape: starting school sweep")
+
+    _auto_scrape_state = {
+        "is_running": True,
+        "current_school": None,
+        "completed": [],
+        "failed": [],
+        "started_at": time.time(),
+        "finished_at": None,
+        "cancel_requested": False,
+    }
+
+    # ── Create a history entry for this sweep run ──────────────
+    sweep_entry = {
+        "sweep_id": len(_auto_scrape_history) + 1,
+        "is_running": True,
+        "current_school": None,
+        "completed": [],
+        "failed": [],
+        "started_at": time.time(),
+        "finished_at": None,
+        "cancelled": False,
+    }
+    _auto_scrape_history.append(sweep_entry)
+
+    all_scraped_companies = set()
+
+    for school in _AUTO_SCRAPE_SCHOOLS:
+        if _auto_scrape_state.get("cancel_requested"):
+            logger.info("⏰ Auto-scrape: cancelled by user")
+            sweep_entry["cancelled"] = True
+            break
+
+        _auto_scrape_state["current_school"] = school
+        sweep_entry["current_school"] = school
+        try:
+            logger.info(f"⏰ Auto-scrape: triggering [{school}]")
+
+            params = {
+                "keywords": school,
+                "freshness": "r86400",
+                "work_types": ["onsite", "remote", "hybrid"],
+                "job_types": ["internship"],
+            }
+
+            async with httpx.AsyncClient(
+                base_url=_SELF_BASE_URL,
+                timeout=None,
+            ) as client:
+                async with client.stream(
+                    "POST",
+                    "/scrape-internships",
+                    params=params,
+                ) as resp:
+                    total = 0
+                    async for line in resp.aiter_lines():
+                        if not line.strip():
+                            continue
+                        try:
+                            event = json.loads(line)
+                            etype = event.get("type")
+                            if etype == "info":
+                                logger.info(f"⏰ [{school}] {event.get('message')}")
+                            elif etype == "jobs":
+                                jobs_data = event.get("data", [])
+                                total += len(jobs_data)
+                                for j in jobs_data:
+                                    if j.get("company"):
+                                        all_scraped_companies.add(j["company"])
+                                logger.info(f"⏰ [{school}] +{len(jobs_data)} jobs streamed")
+                            elif etype == "done":
+                                total = event.get("total", total)
+                                logger.info(
+                                    f"⏰ [{school}] done — total={total} "
+                                    f"engine={event.get('engine')} "
+                                    f"filtered_out={event.get('filtered_out')}"
+                                )
+                            elif etype == "error":
+                                logger.error(f"⏰ [{school}] error: {event.get('message')}")
+                        except Exception:
+                            pass
+
+            _auto_scrape_state["completed"].append({"school": school, "jobs": total})
+            sweep_entry["completed"].append({"school": school, "jobs": total})
+            await asyncio.sleep(random.uniform(10.0, 20.0))
+
+        except Exception as e:
+            logger.error(f"⏰ Auto-scrape [{school}]: failed — {e}")
+            _auto_scrape_state["failed"].append({"school": school, "error": str(e)})
+            sweep_entry["failed"].append({"school": school, "error": str(e)})
+            continue
+
+    _auto_scrape_state["is_running"] = False
+    _auto_scrape_state["current_school"] = None
+    _auto_scrape_state["finished_at"] = time.time()
+
+    # ── Finalize the history entry ─────────────────────────────
+    sweep_entry["is_running"] = False
+    sweep_entry["current_school"] = None
+    sweep_entry["finished_at"] = time.time()
+    logger.info("⏰ Auto-scrape: sweep complete")
+
+    if all_scraped_companies and not _auto_scrape_state.get("cancel_requested"):
+        logger.info(f"⏰ Auto-scrape: rating {len(all_scraped_companies)} collected companies...")
+        try:
+            ratings = await process_company_ratings(list(all_scraped_companies))
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(executor, _persist_ratings, ratings)
+        except Exception as e:
+            logger.error(f"⏰ Auto-scrape: failed to rate companies: {e}")
+
+
+@app.on_event("startup")
+async def startup_event():
+    global se_driver, pw_browser, pw_stealth_ctx, _UA_POOL
+    loop = asyncio.get_event_loop()
+
+    # 0. Build the UA rotation pool
+    _UA_POOL = _build_ua_pool()
+    logger.info(f"✓ UA pool built ({len(_UA_POOL)} variants, Chrome v{_get_chrome_major_version() or '?'})")
+
+    # 1. Primary: Selenium (best anti-detection for LinkedIn)
+    try:
+        se_driver = await loop.run_in_executor(executor, _create_selenium_driver)
+        logger.info("✓ Selenium undetected-chromedriver ready")
+    except Exception as e:
+        logger.warning(f"✗ Selenium init failed (Chrome installed?): {e}")
+
+    # 2. Fallback: Playwright + Stealth
+    try:
+        stealth = Stealth()
+        pw_stealth_ctx = stealth.use_async(async_playwright())
+        pw = await pw_stealth_ctx.__aenter__()
+        pw_browser = await pw.chromium.launch(
+            headless=True,
+            args=["--disable-blink-features=AutomationControlled"],
+        )
+        logger.info("✓ Playwright stealth browser ready")
+    except Exception as e:
+        logger.warning(f"✗ Playwright init failed: {e}")
+    if not scheduler.running:
+        scheduler.start()
+        logger.info("✓ Scheduler started (awaiting manual trigger)")
+    else:
+        logger.info("⚠ Scheduler already running, skipping start")
+    
+
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    global se_driver, pw_browser, pw_stealth_ctx
+    scheduler.shutdown(wait=False)
+    loop = asyncio.get_event_loop()
+
+    if se_driver:
+        await loop.run_in_executor(executor, se_driver.quit)
+    if pw_browser:
+        await pw_browser.close()
+    if pw_stealth_ctx:
+        await pw_stealth_ctx.__aexit__(None, None, None)
+
+
+# ── Cookie warm-up ────────────────────────────────────────────
+
+def _selenium_get_cookies(driver, url: str) -> dict:
+    """Navigate to a URL and extract cookies (runs in executor thread)."""
+    driver.get(url)
+    time.sleep(random.uniform(2.0, 4.0))  # let LinkedIn set session cookies
+    try:
+        return {c["name"]: c["value"] for c in driver.get_cookies()}
+    except Exception:
+        return {}
+
+
+async def _warm_up_cookies(force: bool = False) -> dict:
+    """Obtain fresh LinkedIn session cookies via a browser visit.
+
+    Uses Selenium (preferred) or Playwright to visit a simple LinkedIn
+    guest page, allowing LinkedIn to set its session/tracking cookies
+    (JSESSIONID, bcookie, li_gc, etc.). These cookies are then forwarded
+    to the httpx HTTP client for API requests.
+
+    Results are cached for _COOKIE_TTL_SECONDS (10 min) to avoid
+    redundant browser visits on consecutive scrapes.
+    """
+    global _warm_cookies, _warm_cookies_ts, se_driver
+
+    # Return cached cookies if still fresh
+    if not force and _warm_cookies and (time.time() - _warm_cookies_ts) < _COOKIE_TTL_SECONDS:
+        logger.info("cookie warm-up: using cached cookies (still fresh)")
+        return _warm_cookies
+
+    warm_url = "https://www.linkedin.com/jobs/search/?keywords=intern&f_TPR=r86400"
+    cookies: dict = {}
+
+    # Try Selenium first
+    if se_driver:
+        try:
+            loop = asyncio.get_event_loop()
+            cookies = await loop.run_in_executor(
+                executor, _selenium_get_cookies, se_driver, warm_url
+            )
+            if cookies:
+                logger.info(f"cookie warm-up: got {len(cookies)} cookies via Selenium")
+        except Exception as e:
+            logger.warning(f"cookie warm-up: Selenium failed: {e}")
+            err_str = str(e).lower()
+            if "window" in err_str or "closed" in err_str or "view" in err_str:
+                logger.info("cookie warm-up: Recreating crashed Selenium driver...")
+                try:
+                    loop = asyncio.get_event_loop()
+                    try:
+                        await loop.run_in_executor(executor, se_driver.quit)
+                    except Exception:
+                        pass
+                    se_driver = await loop.run_in_executor(executor, _create_selenium_driver)
+                    cookies = await loop.run_in_executor(
+                        executor, _selenium_get_cookies, se_driver, warm_url
+                    )
+                    if cookies:
+                        logger.info(f"cookie warm-up: got {len(cookies)} cookies via recreated Selenium")
+                except Exception as e2:
+                    logger.warning(f"cookie warm-up: Selenium recreation failed: {e2}")
+
+    # Fall back to Playwright
+    if not cookies and pw_browser:
+        try:
+            ctx = await pw_browser.new_context(user_agent=_build_user_agent())
+            page = await ctx.new_page()
+            await page.goto(warm_url, wait_until="domcontentloaded", timeout=20000)
+            await asyncio.sleep(random.uniform(2.0, 4.0))
+            cookies = {c["name"]: c["value"] for c in await ctx.cookies()}
+            await page.close()
+            await ctx.close()
+            if cookies:
+                logger.info(f"cookie warm-up: got {len(cookies)} cookies via Playwright")
+        except Exception as e:
+            logger.warning(f"cookie warm-up: Playwright failed: {e}")
+
+    if cookies:
+        _warm_cookies = cookies
+        _warm_cookies_ts = time.time()
+    else:
+        logger.warning("cookie warm-up: no cookies obtained from any engine")
+
+    return cookies
+
+
+# ── URL builder ───────────────────────────────────────────────
+#
+# Reference: https://www.linkedin.com/jobs/search/ guest-search parameter spec
+#   sortBy : R  = relevance (fixed)
+#   f_E    : 1  = Internship experience level (fixed)
+#   geoId  : LinkedIn region id — narrows location (fixed, Bengaluru region)
+#   f_PP   : comma-joined Primary-Place ids — UI-only narrowing (fixed)
+#   f_TPR  : r<seconds> freshness — variable (dropdown)
+#   f_WT   : comma-joined work-type codes — variable (3-checkbox group)
+#   f_JT   : comma-joined job-type codes — variable (2-checkbox group)
+#
+# Per project spec, location is fixed (no city picker) and we only expose
+# three user-visible filters: work-type, job-type, freshness.
+#
+# IMPORTANT — geoId vs f_PP split:
+#   • The full UI page at /jobs/search/ respects BOTH geoId AND f_PP and uses
+#     f_PP to narrow within a broader geoId. We keep both on the UI URL so the
+#     "Open in LinkedIn" link matches the user's reference URL.
+#   • The guest /jobs-guest/jobs/api/seeMoreJobPostings/search endpoint
+#     SILENTLY returns an empty 26-byte stub when f_PP is present (it does
+#     not support the f_PP filter), so the HTTP-pagination path strips f_PP
+#     and relies on geoId only. See _PAGINATION_PARAM_KEYS below.
+
+# Always-on params. geoId 90009633 = "Bengaluru" (covers Bengaluru metro on
+# the guest API). f_PP narrows further on the UI URL for the "Open in
+# LinkedIn" link; the API silently ignores it so we strip it for pagination.
+_FIXED_PARAMS: dict[str, str] = {
+    "sortBy": "R",
+    "f_E": "1",
+    "geoId": "90009633",
+}
+
+# Work-type checkbox options (label → LinkedIn f_WT code).
+WORK_TYPES: dict[str, str] = {
+    "onsite": "1",
+    "remote": "2",
+    "hybrid": "3",
+}
+
+# Job-type checkbox options (label → LinkedIn f_JT code) — spec restricts to
+# these two only.
+JOB_TYPES: dict[str, str] = {
+    "internship": "I",
+    "full_time":  "F",
+}
+
+# Freshness dropdown presets (label → LinkedIn f_TPR code).
+FRESHNESS_PRESETS: dict[str, str] = {
+    "hour":  "r3600",
+    "day":   "r86400",
+    "week":  "r604800",
+    "month": "r2592000",
+}
+
+# ── School-based search term generation ───────────────────────
+#
+# Convenient shortcut keywords that resolve to a specific school.
+# The user specifically requested that searches for a school (e.g. "SOB")
+# should search EXACTLY these keywords on LinkedIn.
+KEYWORD_TO_SCHOOL: dict[str, str] = {
+    # [[SOCSE]] — Computer Science & Engineering
+    "software":         "SOCSE",
+    "cs":               "SOCSE",
+    "computer science": "SOCSE",
+    "programming":      "SOCSE",
+    "coding":           "SOCSE",
+    "data":             "SOCSE",
+    "data science":     "SOCSE",
+    "machine learning": "SOCSE",
+    "ai":               "SOCSE",
+    "cyber":            "SOCSE",
+    "cybersecurity":    "SOCSE",
+    "qa":               "SOCSE",
+    "devops":           "SOCSE",
+    "frontend":         "SOCSE",
+    "backend":          "SOCSE",
+    "fullstack":        "SOCSE",
+    "cloud":            "SOCSE",
+    "developer":        "SOCSE",
+    "engineer":         "SOCSE",
+    "sde":                "SOCSE",
+    "swe":                "SOCSE",
+    "it":                 "SOCSE",
+    "web development":    "SOCSE",
+    "app development":    "SOCSE",
+    "mobile development": "SOCSE",
+    "android":            "SOCSE",
+    "ios":                "SOCSE",
+    "react":              "SOCSE",
+    "node":               "SOCSE",
+    "python":             "SOCSE",
+    "java":               "SOCSE",
+    "c++":                "SOCSE",
+    "golang":             "SOCSE",
+    "flutter":            "SOCSE",
+    "aws":                "SOCSE",
+    "azure":              "SOCSE",
+    "gcp":                "SOCSE",
+    "data analyst":       "SOCSE",
+    "data engineer":      "SOCSE",
+    "nlp":                "SOCSE",
+    "computer vision":    "SOCSE",
+    "quality assurance":  "SOCSE",
+    "testing":            "SOCSE",
+    "blockchain":         "SOCSE",
+    "web3":               "SOCSE",
+    "systems engineer":   "SOCSE",
+    "network engineer":   "SOCSE",
+    "site reliability":   "SOCSE",
+    "sre":                "SOCSE",
+    "hardware":           "SOCSE",
+    "embedded":           "SOCSE",
+    # SODI — Design & Innovation
+    "designing":        "SOCSE,SODI",
+    "ui ux":            "SOCSE,SODI",
+    "graphic designer": "SOCSE,SODI",
+    "ui designer":      "SOCSE,SODI",
+    "ux designer":      "SOCSE,SODI",
+    "product designer": "SOCSE,SODI",
+    "product design": "SODI",
+    
+    "industrial design":"SODI",
+    "interaction design":"SODI",
+    "visual design":    "SODI",
+    # SOB — Business
+    "business":         "SOB",
+    "marketing":        "SOB",
+    "finance":          "SOB",
+    "hr":               "SOB",
+    "accounting":       "SOB",
+    "commerce":         "SOB",
+    "mba":              "SOB",
+    "sales":            "SOB",
+    "strategy":         "SOB",
+    "operations":       "SOB",
+    "management":       "SOB",
+    "consulting":       "SOB",
+    "analyst":          "SOB",
+    "business analyst": "SOB",
+    "business development": "SOB",
+    "bda":              "SOB",
+    "bde":              "SOB",
+    "sales intern":     "SOB",
+    "marketing intern": "SOB",
+    "hr intern":        "SOB",
+    "finance intern":   "SOB",
+    "investment banking": "SOB",
+    "venture capital":  "SOB",
+    "private equity":   "SOB",
+    "digital marketing": "SOB",
+    "seo":              "SOB",
+    "content writer":   "SOB",
+    "social media":     "SOB",
+    "growth":           "SOB",
+    "product manager":  "SOB",
+    "project manager":  "SOB",
+    # SoEPP — Economics & Public Policy
+    "economics":        "SoEPP",
+    "policy":           "SoEPP",
+    "governance":       "SoEPP",
+    "public policy":    "SoEPP",
+    "research":         "SoEPP",
+    "development studies":"SoEPP",
+    "economist":        "SoEPP",
+    "policy analyst":   "SoEPP",
+    "public relations": "SoEPP",
+    "government":       "SoEPP",
+    # SOLaw — Law
+    "law":              "SOLaw",
+    "legal":            "SOLaw",
+    "paralegal":        "SOLaw",
+    "attorney":         "SOLaw",
+    "counsel":          "SOLaw",
+    "cyber law":        "SOLaw",
+    "law intern":       "SOLaw",
+    "legal intern":     "SOLaw",
+    "litigation":       "SOLaw",
+    "corporate law":    "SOLaw",
+    "ipr":              "SOLaw",
+    # SOFMCA — Film, Media & Creative Arts
+    "media":            "SOFMCA",
+    "film":             "SOFMCA",
+    "journalism":       "SOFMCA",
+    "animation":        "SOFMCA",
+    "vfx":              "SOFMCA",
+    "gaming":           "SOFMCA",
+    "content creation": "SOFMCA",
+    "acting":           "SOFMCA",
+    "editor":           "SOFMCA",
+    "video editor":     "SOFMCA",
+    "cinematographer":  "SOFMCA",
+    "copywriter":       "SOFMCA",
+    "media intern":     "SOFMCA",
+    "pr intern":        "SOFMCA",
+    "producer":         "SOFMCA",
+    # SOLAS — Liberal Arts & Sciences
+    "psychology":       "SOLAS",
+    "environment":      "SOLAS",
+    "liberal arts":     "SOLAS",
+    "sociology":        "SOLAS",
+    "history":          "SOLAS",
+    "behavioral science":"SOLAS",
+    "psychology intern": "SOLAS",
+    "counseling":       "SOLAS",
+    "sociologist":      "SOLAS",
+    "research assistant":"SOLAS",
+    # SOAHP — Allied & Healthcare
+    "healthcare":       "SOAHP",
+    "medical":          "SOAHP",
+    "clinical":         "SOAHP",
+    "laboratory":       "SOAHP",
+    "nursing":          "SOAHP",
+    "public health":    "SOAHP",
+    "clinical research": "SOAHP",
+    "public health intern": "SOAHP",
+    "hospital administration": "SOAHP",
+    "pharma":           "SOAHP",
+}
+
+# Build SCHOOL_SEARCH_TERMS directly from KEYWORD_TO_SCHOOL.
+# The user explicitly wants us to search *exactly* these keywords.
+SCHOOL_SEARCH_TERMS: dict[str, list[str]] = {}
+for kw, school_str in KEYWORD_TO_SCHOOL.items():
+    for school in school_str.split(","):
+        school_lower = school.strip().lower()
+        if school_lower not in SCHOOL_SEARCH_TERMS:
+            SCHOOL_SEARCH_TERMS[school_lower] = []
+        SCHOOL_SEARCH_TERMS[school_lower].append(kw)
+
+
+def build_search_params(
+    keywords: str,
+    freshness: str = "r86400",
+    work_types: Optional[list[str]] = None,
+    job_types: Optional[list[str]] = None,
+) -> dict[str, str]:
+    """Build the LinkedIn guest search query-param dict.
+
+    Fixed params (sortBy, f_E, f_PP) are always present.
+    Variable params reflect user-selected checkboxes / dropdown:
+      - f_TPR : freshness (raw "r<seconds>" code)
+      - f_WT  : comma-joined work-type codes (only when any box ticked)
+      - f_JT  : comma-joined job-type codes  (only when any box ticked)
+    """
+    params: dict[str, str] = {
+        "keywords": keywords,
+        **_FIXED_PARAMS,
+        "f_TPR": freshness,
+    }
+
+    if work_types:
+        codes = [WORK_TYPES[k] for k in work_types if k in WORK_TYPES]
+        if codes:
+            params["f_WT"] = ",".join(codes)
+
+    if job_types:
+        codes = [JOB_TYPES[k] for k in job_types if k in JOB_TYPES]
+        if codes:
+            params["f_JT"] = ",".join(codes)
+
+    return params
+
+
+def build_search_url(
+    keywords: str,
+    freshness: str = "r86400",
+    work_types: Optional[list[str]] = None,
+    job_types: Optional[list[str]] = None,
+) -> str:
+    """Builds a fully-parameterised LinkedIn guest search URL.
+
+    Shape matches LinkedIn's UI-canonical form, e.g.:
+      https://www.linkedin.com/jobs/search/?keywords=intern&sortBy=R&f_E=1
+        &f_PP=105214831,113968072,112565523,119634689&f_TPR=r86400&f_WT=1,2&f_JT=I
+    """
+    params = build_search_params(
+        keywords=keywords,
+        freshness=freshness,
+        work_types=work_types,
+        job_types=job_types,
+    )
+    # Trailing slash on /jobs/search/ matches LinkedIn's UI-canonical URL.
+    return f"https://www.linkedin.com/jobs/search/?{urllib.parse.urlencode(params)}"
+
+
+# ── Shared HTML parser ──���─────────────────────────────────────
+
+def _parse_jobs_from_html(html: str) -> list[dict]:
+    """Extract job listings from raw HTML using BeautifulSoup.
+    Cascading selectors handle multiple LinkedIn page variants.
+
+    NOTE: contact_details (hiring team links) are NOT extracted here.
+    They live on each job's *detail* page, not the search results list.
+    Use _fetch_hiring_links_for_jobs() after collection to enrich jobs.
+    """
+    soup = BeautifulSoup(html, "html.parser")
+    jobs = []
+
+    # Primary selector: public guest search results
+    cards = soup.select("ul.jobs-search__results-list > li")
+
+    # Fallback selectors for alternate page states + seeMoreJobPostings fragments
+    # (the paginated endpoint returns bare <li> elements with no wrapping <ul>).
+    if not cards:
+        cards = soup.select("li.job-search-card, li.result-card")
+    if not cards:
+        cards = soup.select("div.base-card.base-search-card")
+    if not cards:
+        cards = soup.select("[data-entity-urn*='jobPosting']")
+
+    for card in cards:
+        try:
+            # Title
+            title_el = (
+                card.select_one(".base-search-card__title")
+                or card.select_one("h3")
+                or card.select_one("[class*='title']")
+            )
+            # Company
+            company_el = (
+                card.select_one(".base-search-card__subtitle a")
+                or card.select_one(".base-search-card__subtitle")
+                or card.select_one("h4 a")
+            )
+            # Location
+            loc_el = (
+                card.select_one(".job-search-card__location")
+                or card.select_one("[class*='location']")
+            )
+            # Link
+            link_el = (
+                card.select_one("a.base-card__full-link")
+                or card.select_one("a[href*='/jobs/view/']")
+                or card.select_one("a[href*='linkedin.com/jobs']")
+            )
+            # Posted date
+            date_el = (
+                card.select_one("time.job-search-card__listdate")
+                or card.select_one("time.job-search-card__listdate--new")
+                or card.select_one("time")
+            )
+
+            title = title_el.get_text(strip=True) if title_el else None
+            company = company_el.get_text(strip=True) if company_el else None
+            location = loc_el.get_text(strip=True) if loc_el else None
+            raw_link = link_el.get("href", "") if link_el else ""
+            link = raw_link.split("?")[0] if raw_link else None
+            posted = date_el.get_text(strip=True) if date_el else None
+            posted_dt = date_el.get("datetime") if date_el else None
+
+            # At least title or company must exist for a valid card
+            if title or company:
+                classification = classify_job(title or "", company or "")
+                jobs.append({
+                    "title": title,
+                    "company": company,
+                    "location": location,
+                    "link": link,
+                    "posted": posted,
+                    "posted_datetime": posted_dt,
+                    "programs": classification["programs"],
+                    "schools": classification["schools"],
+                    "contact_details": [],  # enriched later by _fetch_hiring_links_for_jobs
+                })
+        except Exception:
+            continue
+
+    return jobs
+
+
+
+# ── Serper-based hiring manager search ───────────────────────
+
+_SERPER_API_KEY = os.environ.get("SERPER_API_KEY", "0820e51e9080c289b3849e30189e023774fdad87")
+_SERPER_URL = "https://google.serper.dev/search"
+
+# Semaphore: at most 3 concurrent Serper requests
+_SERPER_SEM = asyncio.Semaphore(3)
+
+_LINKEDIN_IN_RE = re.compile(r'https?://(?:[\w-]+\.)?linkedin\.com/in/([\w%-]+)', re.IGNORECASE)
+
+
+def _extract_linkedin_profile_from_serper(data: dict) -> dict | None:
+    """Parse a Serper JSON response and return the first LinkedIn /in/ profile found.
+
+    Priority order:
+      1. answerBox.link         (Google AI / Featured Snippet)
+      2. answerBox.snippet      (text contains a linkedin.com/in/ URL)
+      3. organic[0].link        (first organic result)
+      4. organic[].link         (first organic result that is a /in/ URL)
+
+    Returns:
+        {"name": "<person_slug_or_extracted_name>", "url": "<linkedin_profile_url>"}
+        or None if nothing found.
+    """
+    # 1. answerBox direct link
+    ab = data.get("answerBox", {})
+    ab_link = ab.get("link", "")
+    if ab_link and "linkedin.com/in/" in ab_link:
+        slug = ab_link.rstrip("/").split("/in/")[-1].split("?")[0]
+        name = ab.get("title") or slug.replace("-", " ").title()
+        return {"name": name, "url": ab_link.split("?")[0]}
+
+    # 2. answerBox snippet contains a URL
+    ab_snippet = ab.get("snippet", "") or ab.get("answer", "")
+    m = _LINKEDIN_IN_RE.search(ab_snippet)
+    if m:
+        url = m.group(0).split("?")[0]
+        slug = m.group(1)
+        name = ab.get("title") or slug.replace("-", " ").title()
+        return {"name": name, "url": url}
+
+    # 3 & 4. Organic results — prefer linkedin.com/in/ links
+    for result in data.get("organic", []):
+        link = result.get("link", "")
+        if "linkedin.com/in/" in link:
+            slug = link.rstrip("/").split("/in/")[-1].split("?")[0]
+            name = result.get("title", slug).split(" - ")[0].split(" | ")[0].strip()
+            return {"name": name, "url": link.split("?")[0]}
+
+    return None
+
+
+async def _fetch_hiring_contacts_via_serper(jobs: list[dict]) -> None:
+    """Enrich each job dict in-place with contact_details via Serper Google search.
+
+    For each job, searches:
+        site:linkedin.com/in "{company}" "hiring" OR "recruiter" OR "HR"
+
+    contact_details becomes a list with at most 1 entry:
+        [{"name": "Person Name", "url": "https://linkedin.com/in/slug"}]
+
+    Jobs where no profile is found keep contact_details as [].
+    Requires SERPER_API_KEY in environment.
+    """
+    if not _SERPER_API_KEY:
+        logger.warning("serper: SERPER_API_KEY not set — skipping hiring contact search")
+        return
+
+    async def _search_one(job: dict) -> None:
+        company = job.get("company") or ""
+        if not company:
+            return
+
+        # Build a targeted query: site operator + company + hiring signals
+        query = f'site:linkedin.com/in "{company}" hiring OR recruiter OR "HR"'
+
+        try:
+            async with _SERPER_SEM:
+                await asyncio.sleep(random.uniform(0.2, 0.6))
+                async with httpx.AsyncClient(timeout=10.0) as client:
+                    resp = await client.post(
+                        _SERPER_URL,
+                        json={"q": query, "num": 5, "gl": "in", "hl": "en"},
+                        headers={
+                            "X-API-KEY": _SERPER_API_KEY,
+                            "Content-Type": "application/json",
+                        },
+                    )
+
+            if resp.status_code != 200:
+                logger.debug(
+                    f"serper: HTTP {resp.status_code} for company={company!r}"
+                )
+                return
+
+            data = resp.json()
+            profile = _extract_linkedin_profile_from_serper(data)
+            if profile:
+                job["contact_details"] = [profile]
+                logger.info(
+                    f"serper: ✓ {job.get('title','?')} @ {company} → "
+                    f"{profile['name']} ({profile['url']})"
+                )
+            else:
+                logger.debug(f"serper: no profile found for company={company!r}")
+
+        except Exception as e:
+            logger.debug(f"serper: error searching for {company!r}: {e}")
+
+    await asyncio.gather(*[_search_one(j) for j in jobs])
+
+
+
+
+# ── Paginated HTTP fetcher (seeMoreJobPostings) ───────────────
+
+_SEE_MORE_URL = (
+    "https://www.linkedin.com/jobs-guest/jobs/api/seeMoreJobPostings/search"
+)
+# Keys forwarded to the seeMoreJobPostings endpoint.
+# NOTE: f_PP is intentionally EXCLUDED here — the guest API returns an empty
+# 26-byte stub whenever f_PP is present (verified 2026-05-20). geoId provides
+# equivalent narrowing. f_PP stays on the UI URL only.
+_PAGINATION_PARAM_KEYS = {
+    "keywords", "sortBy", "f_E", "geoId", "f_TPR", "f_WT", "f_JT",
+}
+
+
+async def _fetch_more_pages(
+    base_params: dict,
+    cookies: dict,
+    headers: dict,
+    max_pages: int = 20,
+    start_offset: int = 25,
+) -> list[dict]:
+    """Page through LinkedIn's guest seeMoreJobPostings endpoint.
+
+    The endpoint returns ~25 job cards as a raw HTML fragment per call.
+    Stops on: 4xx/5xx (rate limit, with retry on 429), empty body,
+    empty stub detection, or empty parsed result. Returns whatever was
+    collected before stopping — never raises.
+
+    Retries on transient network errors (DNS, connection timeouts) with
+    exponential backoff to handle temporary infrastructure issues.
+    """
+    params_clean = {
+        k: v for k, v in base_params.items()
+        if k in _PAGINATION_PARAM_KEYS and v not in (None, "")
+    }
+
+    collected: list[dict] = []
+    rate_limit_retries_left = 2  # total 429s tolerated across the whole loop
+    first_response_dumped = False
+    dump_path = os.path.join(tempfile.gettempdir(), "linkedin_pagination_first_response.html")
+    try:
+        async with httpx.AsyncClient(
+            timeout=20.0,
+            cookies=cookies,
+            headers=headers,
+            follow_redirects=True,
+            http2=False,
+        ) as client:
+            # Brief settle before first request — reduced from 3s for speed.
+            await asyncio.sleep(random.uniform(1.0, 2.0))
+
+            start = start_offset
+            for page_idx in range(max_pages):
+                url = f"{_SEE_MORE_URL}?{urllib.parse.urlencode({**params_clean, 'start': start})}"
+                logger.info(f"pagination: GET {url}")
+
+                # Retry loop for transient network errors
+                max_network_retries = 3
+                for attempt in range(1, max_network_retries + 1):
+                    try:
+                        resp = await client.get(url)
+                        break  # Success, exit retry loop
+                    except (httpx.TimeoutException, httpx.NetworkError, OSError) as e:
+                        # Transient network errors: DNS (getaddrinfo), timeout, connection reset
+                        if attempt < max_network_retries:
+                            backoff = min(2 ** attempt, 10.0) * random.uniform(1.0, 1.5)
+                            logger.warning(
+                                f"pagination: transient network error at start={start} "
+                                f"(attempt {attempt}/{max_network_retries}): {type(e).__name__}. "
+                                f"Retrying in {backoff:.1f}s..."
+                            )
+                            await asyncio.sleep(backoff)
+                        else:
+                            logger.warning(
+                                f"pagination: network error at start={start} after "
+                                f"{max_network_retries} attempts: {e}; stopping with {len(collected)} jobs"
+                            )
+                            return collected
+                    except httpx.HTTPError as e:
+                        logger.warning(
+                            f"pagination: HTTP error at start={start}: {e}; "
+                            f"stopping with {len(collected)} extra jobs"
+                        )
+                        return collected
+
+                # Retry on 429 with longer backoff — LinkedIn's rate limits
+                # are usually short-lived and clear after ~10s.
+                if resp.status_code == 429 and rate_limit_retries_left > 0:
+                    rate_limit_retries_left -= 1
+                    backoff = random.uniform(15.0, 25.0)
+                    logger.warning(
+                        f"pagination: 429 at start={start}; backing off {backoff:.1f}s "
+                        f"and retrying ({rate_limit_retries_left} retries left)"
+                    )
+                    await asyncio.sleep(backoff)
+                    try:
+                        resp = await client.get(url)
+                    except (httpx.TimeoutException, httpx.NetworkError, OSError, httpx.HTTPError) as e:
+                        logger.warning(f"pagination: retry HTTP error: {e}; stopping")
+                        return collected
+
+                html = resp.text or ""
+                body_len = len(resp.content) if resp.content is not None else 0
+                li_count = html.lower().count("<li")
+
+                logger.info(
+                    f"pagination: start={start} status={resp.status_code} "
+                    f"bytes={body_len} <li>={li_count}"
+                )
+
+                if not first_response_dumped:
+                    try:
+                        with open(dump_path, "w", encoding="utf-8") as f:
+                            f.write(html)
+                        logger.info(f"pagination: dumped first response → {dump_path}")
+                    except Exception as e:
+                        logger.warning(f"pagination: could not dump response: {e}")
+                    first_response_dumped = True
+
+                if resp.status_code >= 400:
+                    logger.warning(
+                        f"pagination: status {resp.status_code} at start={start} "
+                        f"(likely rate-limit); stopping with {len(collected)} extra jobs"
+                    )
+                    break
+
+                if not html.strip():
+                    logger.info(
+                        f"pagination: empty body at start={start}; end of results "
+                        f"(+{len(collected)} extra jobs)"
+                    )
+                    break
+
+                # ── Early stub detection ──────────────────────────
+                # Detect the 26-byte empty stub and login walls before
+                # wasting time on HTML parsing. On the very first page
+                # this signals "HTTP path is blocked" so we can bail fast.
+                if _is_empty_stub(html):
+                    logger.warning(
+                        f"pagination: empty stub/block detected at start={start} "
+                        f"(bytes={body_len}); stopping with {len(collected)} extra jobs"
+                    )
+                    break
+
+                page_jobs = _parse_jobs_from_html(html)
+                logger.info(
+                    f"pagination: start={start} parser returned {len(page_jobs)} jobs"
+                )
+
+                if not page_jobs:
+                    snippet = html[:500].replace("\n", " ")
+                    logger.warning(
+                        f"pagination: parser returned 0 jobs from a non-empty body "
+                        f"(bytes={body_len}, <li>={li_count}). First 500 chars: {snippet!r}"
+                    )
+                    logger.info(
+                        f"pagination: no jobs parsed at start={start}; end of results "
+                        f"(+{len(collected)} extra jobs)"
+                    )
+                    break
+
+                collected.extend(page_jobs)
+                logger.info(
+                    f"pagination: start={start} → +{len(page_jobs)} jobs "
+                    f"(running paginated total {len(collected)})"
+                )
+
+                # Advance by the actual page size LinkedIn returned.
+                start += len(page_jobs)
+
+                # Human-like inter-page jitter — varied enough to avoid
+                # pattern detection.
+                await asyncio.sleep(random.uniform(3.5, 6.0))
+    except Exception as e:
+        logger.warning(
+            f"pagination: unexpected error: {e}; returning {len(collected)} extra jobs"
+        )
+
+    return collected
+
+
+# ── Scraping engines ──────────────────────────────────────────
+
+_JOB_CARD_CSS = (
+    "ul.jobs-search__results-list > li, "
+    "div.base-card.base-search-card, "
+    "[data-entity-urn*='jobPosting']"
+)
+
+
+def _selenium_scrape(driver, url: str) -> tuple[list[dict], dict]:
+    """Synchronous Selenium scrape — executed inside executor thread.
+    Returns (jobs, cookies) so the async caller can paginate over HTTP."""
+    driver.get(url)
+
+    # Wait up to 15 s for at least one job card — if nothing loads, bail so
+    # Playwright fallback can try instead.
+    try:
+        WebDriverWait(driver, 15).until(
+            EC.presence_of_element_located((By.CSS_SELECTOR, _JOB_CARD_CSS))
+        )
+    except Exception:
+        logger.warning("Selenium: no job cards appeared within 15 s")
+        return [], {}
+
+    # Scroll to bottom repeatedly; break after 3 consecutive unchanged heights.
+    no_change = 0
+    for _ in range(15):
+        prev_height = driver.execute_script("return document.body.scrollHeight")
+        driver.execute_script("window.scrollTo(0, document.body.scrollHeight)")
+        time.sleep(1.5)
+
+        # Click "See more jobs" / "Show more results" if present
+        try:
+            btn = driver.find_element(
+                By.XPATH,
+                "//button[contains(translate(., 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'see more')"
+                " or contains(translate(., 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'show more')]"
+            )
+            btn.click()
+            time.sleep(1.0)
+        except Exception:
+            pass
+
+        new_height = driver.execute_script("return document.body.scrollHeight")
+        if new_height == prev_height:
+            no_change += 1
+            if no_change >= 3:
+                break
+        else:
+            no_change = 0
+
+    jobs = _parse_jobs_from_html(driver.page_source)
+    try:
+        cookies = {c["name"]: c["value"] for c in driver.get_cookies()}
+    except Exception as e:
+        logger.warning(f"Selenium: could not extract cookies for pagination: {e}")
+        cookies = {}
+    return jobs, cookies
+
+
+async def _playwright_scrape(browser, url: str) -> tuple[list[dict], dict]:
+    """Async Playwright scrape — fallback engine.
+    Returns (jobs, cookies) so the caller can paginate over HTTP."""
+    context = await browser.new_context(
+        user_agent=_build_user_agent()
+    )
+    page = await context.new_page()
+
+    try:
+        await page.goto(url, wait_until="domcontentloaded", timeout=30000)
+
+        # Wait up to 15 s for at least one job card
+        try:
+            await page.wait_for_selector(_JOB_CARD_CSS, timeout=15000)
+        except Exception:
+            logger.warning("Playwright: no job cards appeared within 15 s")
+            return [], {}
+
+        # Scroll to bottom repeatedly; break after 3 consecutive unchanged heights.
+        no_change = 0
+        for _ in range(15):
+            prev_height = await page.evaluate("document.body.scrollHeight")
+            await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
+            await asyncio.sleep(1.5)
+
+            # Click "See more jobs" button if present
+            try:
+                await page.locator("button:has-text('See more')").click(timeout=1000)
+                await asyncio.sleep(1.0)
+            except Exception:
+                pass
+
+            new_height = await page.evaluate("document.body.scrollHeight")
+            if new_height == prev_height:
+                no_change += 1
+                if no_change >= 3:
+                    break
+            else:
+                no_change = 0
+
+        html = await page.content()
+        jobs = _parse_jobs_from_html(html)
+        try:
+            cookies = {c["name"]: c["value"] for c in await context.cookies()}
+        except Exception as e:
+            logger.warning(f"Playwright: could not extract cookies for pagination: {e}")
+            cookies = {}
+        return jobs, cookies
+    finally:
+        await page.close()
+        await context.close()
+
+
+# ── Single-URL scrape helper (Selenium → Playwright) ─────────
+
+def _build_pagination_headers(referer_url: str) -> dict:
+    """Browser-realistic headers for the seeMoreJobPostings endpoint.
+    Uses a random UA from the rotation pool and matches the installed
+    Chrome version for Sec-Ch-Ua consistency."""
+    v = _get_chrome_major_version() or 131
+    ua = random.choice(_UA_POOL) if _UA_POOL else _build_user_agent()
+    return {
+        "User-Agent": ua,
+        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+        "Accept-Language": "en-US,en;q=0.9",
+        "Accept-Encoding": "gzip, deflate, br",
+        "Referer": referer_url,
+        "Sec-Fetch-Dest": "empty",
+        "Sec-Fetch-Mode": "cors",
+        "Sec-Fetch-Site": "same-origin",
+        "Sec-Ch-Ua": f'"Chromium";v="{v}", "Not_A Brand";v="24"',
+        "Sec-Ch-Ua-Mobile": "?0",
+        "Sec-Ch-Ua-Platform": '"Windows"',
+        "Connection": "keep-alive",
+    }
+
+
+async def _scrape_one_url(
+    url: str,
+    params: Optional[dict] = None,
+    paginate: bool = False,
+) -> tuple[list[dict], str]:
+    """Scrape one URL with cookie warm-up, exponential-backoff retry,
+    and smarter browser fallback.
+
+    Pipeline:
+      1. Warm up cookies via browser visit (cached for 10 min).
+      2. Try HTTP pagination with warm cookies (up to 3 attempts
+         with exponential backoff on failure).
+      3. If HTTP fails after all retries, fall back to Selenium
+         then Playwright to scrape the UI page directly.
+      4. If a browser engine succeeds, try one more HTTP pagination
+         pass using the browser's fresh cookies to extend the results.
+
+    Returns (jobs, engine_used) where engine_used is one of
+    'http' / 'selenium' / 'playwright' / 'none'.
+    """
+    global se_driver
+    jobs: list[dict] = []
+    engine_used = "none"
+    browser_cookies: dict = {}
+
+    # ── Phase 1 — HTTP with warm cookies + retry loop ─────────────
+    if paginate and params:
+        max_http_attempts = 3
+        for attempt in range(1, max_http_attempts + 1):
+            # Warm up cookies (uses cache if fresh; force re-warm on retries)
+            cookies = await _warm_up_cookies(force=(attempt > 1))
+            logger.info(
+                f"HTTP attempt {attempt}/{max_http_attempts} "
+                f"(cookies={'warm' if cookies else 'cold'})"
+            )
+
+            http_jobs = await _fetch_more_pages(
+                base_params=params,
+                cookies=cookies,
+                headers=_build_pagination_headers(referer_url=url),
+                max_pages=40,
+                start_offset=0,
+            )
+
+            if http_jobs:
+                jobs = http_jobs
+                engine_used = "http"
+                logger.info(
+                    f"HTTP scraped {len(jobs)} jobs on attempt {attempt}"
+                )
+                break
+
+            # Exponential backoff before next attempt
+            if attempt < max_http_attempts:
+                backoff = (2 ** attempt) * random.uniform(2.0, 4.0)
+                logger.warning(
+                    f"HTTP attempt {attempt} returned 0 jobs; "
+                    f"backing off {backoff:.1f}s before retry"
+                )
+                await asyncio.sleep(backoff)
+
+        if not jobs:
+            logger.warning(
+                f"HTTP pagination failed after {max_http_attempts} attempts "
+                "— falling back to browser engines"
+            )
+
+    # ── Phase 2 — Browser fallback (Selenium → Playwright) ────────
+    if not jobs and se_driver:
+        try:
+            async with _scrape_lock:
+                loop = asyncio.get_event_loop()
+                jobs, browser_cookies = await loop.run_in_executor(
+                    executor, _selenium_scrape, se_driver, url
+                )
+                if jobs:
+                    engine_used = "selenium"
+                    logger.info(f"Selenium scraped {len(jobs)} jobs (fallback)")
+        except Exception as e:
+            logger.warning(f"Selenium fallback failed: {e}")
+            err_str = str(e).lower()
+            if "window" in err_str or "closed" in err_str or "view" in err_str:
+                logger.info("Selenium fallback: Recreating crashed driver...")
+                try:
+                    async with _scrape_lock:
+                        loop = asyncio.get_event_loop()
+                        try:
+                            await loop.run_in_executor(executor, se_driver.quit)
+                        except Exception:
+                            pass
+                        se_driver = await loop.run_in_executor(executor, _create_selenium_driver)
+                        jobs, browser_cookies = await loop.run_in_executor(
+                            executor, _selenium_scrape, se_driver, url
+                        )
+                        if jobs:
+                            engine_used = "selenium"
+                            logger.info(f"Selenium scraped {len(jobs)} jobs via recreated driver")
+                except Exception as e2:
+                    logger.warning(f"Selenium fallback recreation failed: {e2}")
+
+    if not jobs and pw_browser:
+        try:
+            jobs, browser_cookies = await _playwright_scrape(pw_browser, url)
+            if jobs:
+                engine_used = "playwright"
+                logger.info(f"Playwright scraped {len(jobs)} jobs (fallback)")
+        except Exception as e:
+            logger.warning(f"Playwright fallback failed: {e}")
+
+    # ── Phase 3 — Post-browser HTTP extension ─────────────────────
+    # If a browser engine got results, try HTTP pagination with its
+    # fresh cookies to collect additional pages beyond what the browser
+    # initially loaded (the browser typically only sees page 1).
+    if jobs and browser_cookies and paginate and params:
+        logger.info(
+            f"Extending {len(jobs)} browser results via HTTP with "
+            f"{len(browser_cookies)} fresh browser cookies"
+        )
+        extra_jobs = await _fetch_more_pages(
+            base_params=params,
+            cookies=browser_cookies,
+            headers=_build_pagination_headers(referer_url=url),
+            max_pages=20,
+            start_offset=len(jobs),  # continue from where the browser left off
+        )
+        if extra_jobs:
+            jobs.extend(extra_jobs)
+            logger.info(
+                f"HTTP extension added {len(extra_jobs)} jobs "
+                f"(total now {len(jobs)})"
+            )
+
+    return jobs, engine_used
+
+
+# ── API endpoints ─────────────────────────────────────────────
+
+@app.get("/schools")
+async def get_schools():
+    """Return the full school registry used for job classification."""
+    return SCHOOLS
+
+
+@app.get("/locations")
+async def get_locations():
+    """Return the city → geoId mapping used for location checkboxes."""
+    return INDIAN_METROS
+
+
+# ── Parallel multi-keyword HTTP scraper ───────────────────────
+
+# Concurrency limiter — at most 1 parallel HTTP scrape to avoid
+# triggering LinkedIn's rate limiter. Sequential is safer and still fast.
+_PARALLEL_SEMAPHORE = asyncio.Semaphore(1)
+
+
+async def _fetch_all_pages_for_keyword(
+    keyword: str,
+    base_params: dict,
+    cookies: dict,
+    max_pages: int = 10,
+) -> list[dict]:
+    """Fast HTTP-only scraper for a single keyword.
+
+    Designed for parallel fan-out: lightweight, no browser, shorter
+    inter-page delays. Uses the shared semaphore to throttle concurrency.
+
+    Retries on transient network errors with exponential backoff to handle
+    temporary infrastructure issues.
+    """
+    params = {
+        k: v for k, v in base_params.items()
+        if k in _PAGINATION_PARAM_KEYS and v not in (None, "")
+    }
+    params["keywords"] = keyword  # override with this specific keyword
+
+    referer_url = f"https://www.linkedin.com/jobs/search/?{urllib.parse.urlencode(params)}"
+    headers = _build_pagination_headers(referer_url=referer_url)
+
+    collected: list[dict] = []
+
+    async with _PARALLEL_SEMAPHORE:
+        try:
+            async with httpx.AsyncClient(
+                timeout=20.0,
+                cookies=cookies,
+                headers=headers,
+                follow_redirects=True,
+                http2=False,
+            ) as client:
+                # Brief initial settle
+                await asyncio.sleep(random.uniform(2.0, 4.0))
+
+                start = 0
+                consecutive_empty = 0
+                for page_idx in range(max_pages):
+                    url = f"{_SEE_MORE_URL}?{urllib.parse.urlencode({**params, 'start': start})}"
+                    logger.info(f"parallel[{keyword[:30]}]: GET start={start}")
+
+                    # Retry loop for transient network errors
+                    max_network_retries = 3
+                    resp = None
+                    for attempt in range(1, max_network_retries + 1):
+                        try:
+                            resp = await client.get(url)
+                            break  # Success, exit retry loop
+                        except (httpx.TimeoutException, httpx.NetworkError, OSError) as e:
+                            # Transient network errors
+                            if attempt < max_network_retries:
+                                backoff = min(2 ** attempt, 10.0) * random.uniform(1.0, 1.5)
+                                logger.warning(
+                                    f"parallel[{keyword[:30]}]: transient network error "
+                                    f"(attempt {attempt}/{max_network_retries}): {type(e).__name__}. "
+                                    f"Retrying in {backoff:.1f}s..."
+                                )
+                                await asyncio.sleep(backoff)
+                            else:
+                                logger.warning(
+                                    f"parallel[{keyword[:30]}]: network error at start={start} "
+                                    f"after {max_network_retries} attempts: {e}; "
+                                    f"stopping with {len(collected)} jobs"
+                                )
+                                break
+                        except httpx.HTTPError as e:
+                            logger.warning(
+                                f"parallel[{keyword[:30]}]: HTTP error at start={start}: {e}; stopping"
+                            )
+                            break
+
+                    if resp is None:
+                        break  # All retries failed, stop pagination
+
+                    # Handle rate limits
+                    if resp.status_code == 429:
+                        backoff = random.uniform(15.0, 25.0)
+                        logger.warning(
+                            f"parallel[{keyword[:30]}]: 429 at start={start}; "
+                            f"backing off {backoff:.1f}s"
+                        )
+                        await asyncio.sleep(backoff)
+                        try:
+                            resp = await client.get(url)
+                        except (httpx.TimeoutException, httpx.NetworkError, OSError, httpx.HTTPError):
+                            break
+
+                    if resp.status_code >= 400:
+                        logger.warning(
+                            f"parallel[{keyword[:30]}]: status {resp.status_code}; stopping"
+                        )
+                        break
+
+                    html = resp.text or ""
+                    if not html.strip() or _is_empty_stub(html):
+                        consecutive_empty += 1
+                        if consecutive_empty >= 2:
+                            break
+                        # One empty might be transient — try next offset
+                        start += 25
+                        await asyncio.sleep(random.uniform(0.5, 1.0))
+                        continue
+
+                    consecutive_empty = 0
+                    page_jobs = _parse_jobs_from_html(html)
+                    logger.info(
+                        f"parallel[{keyword[:30]}]: start={start} → {len(page_jobs)} jobs"
+                    )
+
+                    if not page_jobs:
+                        break
+
+                    collected.extend(page_jobs)
+                    start += len(page_jobs)
+
+                    # Human-like delays between pages
+                    await asyncio.sleep(random.uniform(4.0, 7.0))
+
+        except Exception as e:
+            logger.warning(
+                f"parallel[{keyword[:30]}]: unexpected error: {e}; "
+                f"returning {len(collected)} jobs"
+            )
+
+        # Cool-down before releasing semaphore to next keyword
+        await asyncio.sleep(random.uniform(4.0, 8.0))
+
+    logger.info(f"parallel[{keyword[:30]}]: done — {len(collected)} jobs total")
+    return collected
+
+
+def _resolve_search(user_keyword: str) -> tuple[list[str], Optional[str]]:
+    """Resolve user input into expanded search keywords + optional school filter.
+
+    Resolution order:
+      1. Exact school code (case-insensitive): "SOCSE" → all SOCSE terms
+      2. Shortcut keyword: "software" → SOCSE terms
+      3. Direct passthrough: "Data Analyst" → ["Data Analyst"], no filter
+
+    Returns:
+        (search_keywords, school_code_to_filter_by_or_None)
+    """
+    key = user_keyword.strip().lower()
+
+    # 1. Exact school code match (SOCSE, solas, SOB, etc.)
+    school_lower_map = {code.lower(): code for code in SCHOOLS}
+    if key in school_lower_map:
+        code = school_lower_map[key]
+        terms = SCHOOL_SEARCH_TERMS.get(key, [])
+        logger.info(
+            f"resolve_search: school code '{code}' → {len(terms)} search terms"
+        )
+        return terms if terms else [user_keyword], code
+
+    # 2. Shortcut keyword → school (e.g. "software" → SOCSE)
+    if key in KEYWORD_TO_SCHOOL:
+        school_code = KEYWORD_TO_SCHOOL[key]
+        terms = []
+        for school in school_code.split(","):
+            for t in SCHOOL_SEARCH_TERMS.get(school.strip().lower(), []).copy():
+                if t not in terms:
+                    terms.append(t)
+        
+        # Guarantee the user's explicit shortcut keyword is also searched!
+        # Sometimes shortcuts are exact job titles (e.g. "marketing", "data").
+        if key not in [t.lower() for t in terms]:
+            # Insert at the beginning so it's prioritized
+            terms.insert(0, user_keyword.strip())
+            
+        logger.info(
+            f"resolve_search: shortcut '{key}' → school {school_code} "
+            f"({len(terms)} search terms including shortcut)"
+        )
+        return terms if terms else [user_keyword], school_code
+
+    # 3. Direct keyword — no expansion, no school filter
+    logger.info(f"resolve_search: direct keyword '{user_keyword}' (no expansion)")
+    return [user_keyword], None
+
+
+def _deduplicate_jobs(jobs: list[dict]) -> tuple[list[dict], int]:
+    """Deduplicate jobs by canonical link. Returns (deduped_list, num_removed)."""
+    seen: dict[str, dict] = {}
+    untagged: list[dict] = []
+    for job in jobs:
+        key = job.get("link")
+        if key:
+            if key not in seen:
+                seen[key] = job
+        else:
+            untagged.append(job)
+    deduped = list(seen.values()) + untagged
+    return deduped, len(jobs) - len(deduped)
+
+
+
+@app.post("/scrape-internships")
+async def scrape_internships(
+    keywords: str = Query("Software Engineer", description="Job search keywords"),
+    freshness: str = Query("r86400", description="Posting age: r3600=1h r86400=24h r604800=week r2592000=month"),
+    work_types: Optional[list[str]] = Query(
+        default=None,
+        description="Work-type checkboxes — repeat per selection: onsite | remote | hybrid",
+    ),
+    job_types: Optional[list[str]] = Query(
+        default=None,
+        description="Job-type checkboxes — repeat per selection: internship | full_time",
+    ),
+):
+    # ── Resolve keywords + optional school filter ──────────────
+    all_keywords, school_filter = _resolve_search(keywords)
+
+    if job_types and any(jt.lower() == "internship" for jt in job_types):
+        all_keywords = [f"{kw} intern" for kw in all_keywords]
+        keywords = f"{keywords} intern"
+
+    is_multi = len(all_keywords) > 1
+
+    # Build the "primary" URL (used for the "Open in LinkedIn" link)
+    primary_url = build_search_url(
+        keywords=keywords,
+        freshness=freshness,
+        work_types=work_types,
+        job_types=job_types,
+    )
+    primary_params = build_search_params(
+        keywords=keywords,
+        freshness=freshness,
+        work_types=work_types,
+        job_types=job_types,
+    )
+
+    async def _stream_generator():
+        """NDJSON streaming generator.
+
+        Emits lines of JSON, one per event:
+          {"type": "start", ...}
+          {"type": "info", "message": ...}
+          {"type": "jobs", "data": [...], ...}
+          {"type": "done", ...}
+          {"type": "error", "message": ...}
+        """
+        all_jobs: list[dict] = []
+        engine_used = "http"
+        total_dupes_removed = 0
+        total_filtered_out = 0
+
+        try:
+            # ── Emit start event ──────────────────────────────
+            yield json.dumps({
+                "type": "start",
+                "engine": "http",
+                "total_searches": len(all_keywords) if is_multi else 1,
+            }) + "\n"
+
+            if is_multi:
+                school_label = f" (school: {school_filter})" if school_filter else ""
+                yield json.dumps({
+                    "type": "info",
+                    "message": f"Expanding '{keywords}' → {len(all_keywords)} parallel searches{school_label}",
+                }) + "\n"
+
+            # ── Warm up cookies once ──────────────────────────
+            cookies = await _warm_up_cookies(force=False)
+            if not cookies:
+                yield json.dumps({
+                    "type": "info",
+                    "message": "Cookie warm-up: re-warming...",
+                }) + "\n"
+                cookies = await _warm_up_cookies(force=True)
+
+            if is_multi:
+                # ══ PARALLEL MULTI-KEYWORD PATH ══════════════
+                # Fan out all keywords as concurrent tasks.
+                # As each completes, stream its results immediately.
+
+                seen_links: set = set()
+
+                async def _scrape_and_collect(kw: str, index: int):
+                    """Scrape one keyword with a stagger delay to avoid 429s."""
+                    # Stagger: each task waits before starting so requests are spread out.
+                    stagger = index * random.uniform(3.0, 6.0)
+                    if stagger > 0:
+                        await asyncio.sleep(stagger)
+                    jobs = await _fetch_all_pages_for_keyword(
+                        keyword=kw,
+                        base_params=primary_params,
+                        cookies=cookies,
+                        max_pages=10,
+                    )
+                    return kw, jobs
+
+                # Create all tasks
+                tasks = [
+                    asyncio.create_task(_scrape_and_collect(kw, i))
+                    for i, kw in enumerate(all_keywords)
+                ]
+
+                completed = 0
+                for coro in asyncio.as_completed(tasks):
+                    try:
+                        kw, kw_jobs = await coro
+                        completed += 1
+
+                        # Deduplicate against already-seen links
+                        new_jobs = []
+                        for job in kw_jobs:
+                            link = job.get("link")
+                            if link:
+                                if link not in seen_links:
+                                    seen_links.add(link)
+                                    new_jobs.append(job)
+                                else:
+                                    total_dupes_removed += 1
+                            else:
+                                new_jobs.append(job)
+
+                        if school_filter and new_jobs:
+                            pre = len(new_jobs)
+                            filter_schools = [s.strip() for s in school_filter.split(",")]
+                            new_jobs = [
+                                j for j in new_jobs
+                                if any(sf in j.get("schools", []) for sf in filter_schools)
+                            ]
+                            total_filtered_out += pre - len(new_jobs)
+
+                        # Strict internship title filter if only internship is selected
+                        if job_types and "internship" in job_types and "full_time" not in job_types:
+                            pre = len(new_jobs)
+                            intern_kw = ["intern", "trainee", "student", "co-op", "apprentice", "fellow"]
+                            new_jobs = [
+                                j for j in new_jobs
+                                if any(ik in j.get("title", "").lower() for ik in intern_kw)
+                            ]
+                            total_filtered_out += pre - len(new_jobs)
+
+
+                        all_jobs.extend(new_jobs)
+
+                        logger.info(
+                            f"multi-search [{completed}/{len(all_keywords)}]: "
+                            f"'{kw}' → {len(kw_jobs)} raw, {len(new_jobs)} new "
+                            f"(total {len(all_jobs)})"
+                        )
+
+                        # Stream the batch
+                        if new_jobs:
+                            yield json.dumps({
+                                "type": "jobs",
+                                "data": new_jobs,
+                                "keyword": kw,
+                                "engine": "http",
+                                "deduplicated": total_dupes_removed,
+                            }) + "\n"
+
+                        yield json.dumps({
+                            "type": "info",
+                            "message": (
+                                f"[{completed}/{len(all_keywords)}] "
+                                f"'{kw}' → {len(new_jobs)} new jobs "
+                                f"(total: {len(all_jobs)})"
+                            ),
+                        }) + "\n"
+
+                    except Exception as e:
+                        completed += 1
+                        logger.warning(f"multi-search: task error: {e}")
+                        yield json.dumps({
+                            "type": "info",
+                            "message": f"[{completed}/{len(all_keywords)}] Search failed: {e}",
+                        }) + "\n"
+
+                engine_used = "http"
+
+
+            else:
+                # ══ SINGLE-KEYWORD PATH ══════════════════════
+                # Original flow: HTTP → Selenium → Playwright
+                jobs, engine_used = await _scrape_one_url(
+                    primary_url, params=primary_params, paginate=True
+                )
+
+                if jobs:
+                    jobs, dupes = _deduplicate_jobs(jobs)
+                    total_dupes_removed = dupes
+
+                    if school_filter:
+                        pre = len(jobs)
+                        filter_schools = [s.strip() for s in school_filter.split(",")]
+                        jobs = [
+                            j for j in jobs
+                            if any(sf in j.get("schools", []) for sf in filter_schools)
+                        ]
+                        total_filtered_out += pre - len(jobs)
+
+                    # Strict internship title filter if only internship is selected
+                    if job_types and "internship" in job_types and "full_time" not in job_types:
+                        pre = len(jobs)
+                        intern_kw = ["intern", "trainee", "student", "co-op", "apprentice", "fellow"]
+                        jobs = [
+                            j for j in jobs
+                            if any(ik in j.get("title", "").lower() for ik in intern_kw)
+                        ]
+                        total_filtered_out += pre - len(jobs)
+
+
+                    all_jobs = jobs
+
+                    yield json.dumps({
+                        "type": "jobs",
+                        "data": all_jobs,
+                        "engine": engine_used,
+                        "deduplicated": total_dupes_removed,
+                    }) + "\n"
+
+            # ── Emit done event ───────────────────────────────
+            if not all_jobs:
+                yield json.dumps({
+                    "type": "error",
+                    "message": "All scraping engines failed — 0 results",
+                }) + "\n"
+            else:
+                try:
+                    logger.info(
+                        f"serper: Searching hiring contacts for {len(all_jobs)} companies..."
+                    )
+                    yield json.dumps({
+                        "type": "info",
+                        "message": f"Searching for hiring contacts via Google for {len(all_jobs)} companies...",
+                    }) + "\n"
+                    await _fetch_hiring_contacts_via_serper(all_jobs)
+                    found_count = sum(
+                        1 for j in all_jobs if j.get("contact_details")
+                    )
+                    logger.info(
+                        f"serper: Hiring contact search done — "
+                        f"{found_count}/{len(all_jobs)} jobs have a contact"
+                    )
+                    # ── VERIFICATION PRINT (visible in terminal logs) ──────
+                    print(f"\n{'='*60}")
+                    print(f"[SERPER HIRING CONTACTS] {found_count}/{len(all_jobs)} jobs enriched")
+                    for j in all_jobs:
+                        contacts = j.get("contact_details") or []
+                        if contacts:
+                            for c in contacts:
+                                if isinstance(c, dict):
+                                    print(
+                                        f"  ✓ {j.get('title','?')} @ {j.get('company','?')}: "
+                                        f"{c.get('name','?')} → {c.get('url','?')}"
+                                    )
+                                else:
+                                    print(f"  ✓ {j.get('title','?')} @ {j.get('company','?')}: {c}")
+                    print(f"{'='*60}\n")
+
+                    # ── Stream enriched jobs back to frontend ──────────────
+                    enriched_jobs = [j for j in all_jobs if j.get("contact_details")]
+                    if enriched_jobs:
+                        yield json.dumps({
+                            "type": "jobs",
+                            "data": enriched_jobs,
+                            "engine": engine_used,
+                            "deduplicated": 0,
+                            "is_contact_update": True,
+                        }) + "\n"
+                        logger.info(
+                            f"serper: Streamed {len(enriched_jobs)} enriched jobs back to frontend"
+                        )
+                except Exception as e:
+                    logger.warning(f"serper: Hiring contact search failed: {e}")
+
+                # Save jobs to database for student dashboard
+                search_params = {
+                    'keywords': keywords,
+                    'freshness': freshness,
+                    'work_types': work_types or [],
+                    'job_types': job_types or []
+                }
+                try:
+                    saved_count = save_scraped_jobs(all_jobs, search_params)
+                    logger.info(f"Saved {saved_count} jobs to database")
+                except Exception as e:
+                    logger.warning(f"Failed to save jobs to database: {e}")
+                
+                yield json.dumps({
+                    "type": "done",
+                    "total": len(all_jobs),
+                    "engine": engine_used,
+                    "url": primary_url,
+                    "deduplicated": total_dupes_removed,
+                    "filtered_out": total_filtered_out,
+                    "school_filter": school_filter,
+                    "searches_completed": len(all_keywords),
+                }) + "\n"
+
+        except Exception as e:
+            logger.exception(f"scrape-internships stream error: {e}")
+            # If we have collected any jobs, emit them first with partial results notification
+            if all_jobs:
+                yield json.dumps({
+                    "type": "info",
+                    "message": f"Stream interrupted after collecting {len(all_jobs)} jobs",
+                }) + "\n"
+                yield json.dumps({
+                    "type": "done",
+                    "total": len(all_jobs),
+                    "engine": engine_used,
+                    "partial": True,
+                    "error": str(e),
+                    "url": primary_url,
+                    "deduplicated": total_dupes_removed,
+                    "filtered_out": total_filtered_out,
+                    "school_filter": school_filter,
+                    "searches_completed": len(all_keywords),
+                }) + "\n"
+            else:
+                # Only emit error if we have no jobs
+                yield json.dumps({
+                    "type": "error",
+                    "message": str(e),
+                }) + "\n"
+
+    global global_scrape
+    if global_scrape.is_active:
+        global_scrape.cancel()
+    
+    session = ScrapeSession()
+    session.is_active = True
+    global_scrape = session
+
+    async def _pump():
+        search_params_for_db = {
+            'keywords': keywords,
+            'freshness': freshness,
+            'work_types': work_types or [],
+            'job_types': job_types or []
+        }
+        try:
+            async for item in _stream_generator():
+                # Intercept and incrementally save jobs to DB so they appear in student portal mid-scrape
+                try:
+                    data = json.loads(item.strip())
+                    if data.get("type") == "jobs" and data.get("data"):
+                        save_scraped_jobs(data["data"], search_params_for_db)
+                except Exception as e:
+                    logger.warning(f"Incremental save failed: {e}")
+                
+                # Save to history and broadcast to active queues
+                session.history.append(item)
+                for q in session.queues:
+                    await q.put(item)
+        except Exception as e:
+            logger.error(f"Background scraper error: {e}")
+        finally:
+            session.is_active = False
+            for q in session.queues:
+                await q.put(None)
+            session.queues.clear()
+
+    # Start the background task so it continues even if the HTTP connection drops
+    session.task = asyncio.create_task(_pump())
+
+    async def _consumer():
+        q = asyncio.Queue()
+        session.queues.append(q)
+        try:
+            while True:
+                item = await q.get()
+                if item is None:
+                    break
+                yield item
+        except asyncio.CancelledError:
+            logger.info("Client disconnected, background scraper will continue.")
+            raise
+        finally:
+            if q in session.queues:
+                session.queues.remove(q)
+
+    return StreamingResponse(
+        _consumer(),
+        media_type="application/x-ndjson",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+@app.get("/scrape-internships/status")
+async def scrape_status():
+    """Return whether a scrape session is currently active or has history."""
+    return {
+        "is_active": global_scrape.is_active,
+        "has_history": len(global_scrape.history) > 0
+    }
+
+@app.get("/scrape-internships/stream")
+async def stream_scrape():
+    """Reconnect to the active scrape session or replay the last session's history."""
+    async def _reconnect_consumer():
+        # Yield all past events instantly
+        for item in global_scrape.history:
+            yield item
+        
+        # If the session is still active, listen for new events
+        if global_scrape.is_active:
+            q = asyncio.Queue()
+            global_scrape.queues.append(q)
+            try:
+                while True:
+                    item = await q.get()
+                    if item is None:
+                        break
+                    yield item
+            except asyncio.CancelledError:
+                raise
+            finally:
+                if q in global_scrape.queues:
+                    global_scrape.queues.remove(q)
+
+    return StreamingResponse(
+        _reconnect_consumer(),
+        media_type="application/x-ndjson",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+# ── Student Dashboard Endpoints ───────────────────────────────
+
+@app.get("/student/jobs/recent")
+def get_recent_jobs(hours: int = Query(1, description="Hours ago (1 or 24)")):
+    """Get jobs scraped within the last N hours for student dashboard"""
+    try:
+        if hours == 1:
+            # Jobs from the last 1 hour
+            jobs = get_jobs_in_timeframe(1, 0)
+        elif hours == 24:
+            # Jobs from 1-24 hours ago (excluding the last hour to avoid overlap)
+            jobs = get_jobs_in_timeframe(24, 1)
+        else:
+            raise HTTPException(status_code=400, detail="Only 1 or 24 hours supported")
+        
+        return {
+            "jobs": jobs,
+            "count": len(jobs),
+            "timeframe": f"{hours} hour{'s' if hours > 1 else ''} ago"
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/student/jobs/all-timeframes")
+def get_all_timeframes(background_tasks: BackgroundTasks):
+    """Get jobs for all timeframes for student dashboard and auto-cleanup old jobs"""
+    try:
+        # Auto-delete jobs older than 30 days in the background
+        background_tasks.add_task(cleanup_old_jobs, 30)
+
+        # Fetch all jobs in one query and bin them to save DB roundtrips
+        return get_binned_jobs()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/admin/cleanup")
+def cleanup_old_data(days: int = Query(7, description="Remove jobs older than N days")):
+    """Admin endpoint to cleanup old job data"""
+    try:
+        deleted_count = cleanup_old_jobs(days)
+        return {
+            "message": f"Cleaned up {deleted_count} jobs older than {days} days",
+            "deleted_count": deleted_count
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/admin/trigger-scrape")
+async def trigger_auto_scrape():
+    """
+    Manually kick off the full school sweep. Returns immediately;
+    progress can be polled at GET /admin/scrape-status.
+    If a sweep is already running, returns 409.
+    """
+    global _auto_scrape_task
+    if _auto_scrape_state.get("is_running"):
+        return {
+            "status": "already_running",
+            "message": "Auto-scrape is already in progress.",
+            "state": _auto_scrape_state,
+        }
+        
+    scheduler.add_job(
+        _auto_scrape_all_schools,
+        trigger=IntervalTrigger(hours=1),
+        id="hourly_school_scrape",
+        replace_existing=True,
+        next_run_time=datetime.now(timezone.utc),
+        misfire_grace_time=300,
+    )
+    
+    return {
+        "status": "started",
+        "message": f"Auto-scrape started for {len(_AUTO_SCRAPE_SCHOOLS)} schools (running hourly).",
+        "schools": _AUTO_SCRAPE_SCHOOLS,
+    }
+
+
+@app.post("/admin/stop-scrape")
+async def stop_auto_scrape():
+    """
+    Stop the currently running auto-scrape sweep and remove the hourly scheduled job.
+    """
+    if scheduler.get_job("hourly_school_scrape"):
+        scheduler.remove_job("hourly_school_scrape")
+        
+    _auto_scrape_state["cancel_requested"] = True
+    
+    global global_scrape
+    if global_scrape.is_active:
+        global_scrape.cancel()
+
+    return {"status": "stopped", "message": "Auto-scrape has been stopped and scheduled jobs removed."}
+
+
+@app.get("/admin/scrape-status")
+async def get_scrape_status():
+    """Poll the live progress of the currently-running (or last) auto-scrape sweep."""
+    state = dict(_auto_scrape_state)
+    if state.get("started_at"):
+        elapsed = (state.get("finished_at") or time.time()) - state["started_at"]
+        state["elapsed_seconds"] = round(elapsed)
+    total_done = len(state.get("completed", [])) + len(state.get("failed", []))
+    state["progress"] = f"{total_done}/{len(_AUTO_SCRAPE_SCHOOLS)}"
+    return state
+
+
+@app.get("/admin/scrape-history")
+async def get_scrape_history():
+    """Return the full history of all auto-scrape sweep runs (newest first).
+
+    Each entry has: sweep_id, is_running, current_school, completed, failed,
+    started_at, finished_at, cancelled, elapsed_seconds, progress.
+    """
+    history = []
+    for entry in reversed(_auto_scrape_history):
+        item = dict(entry)
+        # Compute elapsed seconds
+        if item.get("started_at"):
+            elapsed = (item.get("finished_at") or time.time()) - item["started_at"]
+            item["elapsed_seconds"] = round(elapsed)
+        # Compute progress string
+        total_done = len(item.get("completed", [])) + len(item.get("failed", []))
+        item["progress"] = f"{total_done}/{len(_AUTO_SCRAPE_SCHOOLS)}"
+        item["total_schools"] = len(_AUTO_SCRAPE_SCHOOLS)
+        item["schools"] = _AUTO_SCRAPE_SCHOOLS
+        history.append(item)
+    return {"history": history, "total_runs": len(history)}
+
+
+async def process_company_ratings(companies: list[str]) -> dict:
+    if not companies:
+        return {}
+
+    SERPER_API_KEY = os.environ.get("SERPER_API_KEY", "0820e51e9080c289b3849e30189e023774fdad87")
+    ratings = {}
+    
+    async def _fetch_company_rating(client: httpx.AsyncClient, company: str, semaphore: asyncio.Semaphore) -> tuple[str, float]:
+        async with semaphore:
+            await asyncio.sleep(random.uniform(0.5, 1.5))
+            query = f'"{company}" reviews Bangalore site:glassdoor.co.in OR site:ambitionbox.com'
+            try:
+                resp = await client.post(
+                    "https://google.serper.dev/search",
+                    json={"q": query},
+                    headers={"X-API-KEY": SERPER_API_KEY, "Content-Type": "application/json"}
+                )
+                data = resp.json()
+                for org in data.get("organic", []):
+                    # Try Serper's built-in rating first
+                    if "rating" in org and isinstance(org["rating"], (int, float)):
+                        return company, float(org["rating"])
+                    # Fallback to regex on snippet
+                    snippet = org.get("snippet", "")
+                    match = re.search(r"Rating:\s*([\d\.]+)", snippet, re.IGNORECASE)
+                    if match:
+                        return company, float(match.group(1))
+            except Exception as e:
+                logger.warning(f"rate-companies: Serper search failed for {company}: {repr(e)}")
+            return company, 2.5
+
+    try:
+        # We limit concurrency to 3 to avoid overwhelming the network or API limits
+        semaphore = asyncio.Semaphore(3)
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            tasks = [_fetch_company_rating(client, c, semaphore) for c in companies]
+            results = await asyncio.gather(*tasks)
+
+        for company, rating in results:
+            # Clamp to [1, 5] just in case
+            ratings[company] = round(max(1.0, min(5.0, rating)), 1)
+
+        # Fill in any companies missed with a neutral score
+        for company in companies:
+            if company not in ratings:
+                ratings[company] = 2.5
+
+        return ratings
+    except Exception as e:
+        logger.error(f"rate-companies: process_company_ratings failed: {repr(e)}")
+        raise e
+
+
+class RateCompaniesRequest(BaseModel):
+    companies: list[str]
+
+
+@app.post("/rate-companies")
+async def rate_companies(req: RateCompaniesRequest, background_tasks: BackgroundTasks):
+    """
+    Use Serper.dev (Google Search API) to estimate a reputation/internship
+    quality rating (1.0–5.0) for each submitted company name (Bangalore context).
+
+    A concurrent search is made for each company, extracting the rating from 
+    Glassdoor/Ambitionbox search snippets.
+    """
+    try:
+        ratings = await process_company_ratings(req.companies)
+        # Persist ratings to DB in the background (non-blocking)
+        background_tasks.add_task(_persist_ratings, ratings)
+
+        logger.info(f"rate-companies: successfully rated {len(ratings)} companies via Search API")
+        return {"ratings": ratings}
+
+    except Exception as e:
+        logger.error(f"rate-companies: Search API call failed: {repr(e)}")
+        raise HTTPException(status_code=500, detail=f"Search rating failed: {str(e)}")
+
+
+def _persist_ratings(ratings: dict):
+    """Write company ratings back to the DB (called as a background task)."""
+    try:
+        updated = update_company_ratings(ratings)
+        logger.info(f"rate-companies: persisted ratings for {updated} job rows in DB")
+    except Exception as e:
+        logger.error(f"rate-companies: DB persist failed: {e}")
+
+
+# ── Manual Rating Management ─────────────────────────────────────
+
+class SetRatingRequest(BaseModel):
+    company: str
+    rating: float
+
+@app.post("/set-company-rating")
+async def set_company_rating(req: SetRatingRequest):
+    """Manually set a rating for a specific company (for testing/admin purposes)."""
+    try:
+        # Validate rating range
+        rating = max(1.0, min(5.0, req.rating))
+        
+        # Update the database
+        updated = update_company_ratings({req.company: rating})
+        
+        return {
+            "message": f"Updated rating for '{req.company}' to {rating}",
+            "company": req.company,
+            "rating": rating,
+            "rows_updated": updated
+        }
+    except Exception as e:
+        logger.error(f"set-company-rating: failed: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/get-company-ratings")
+async def get_company_ratings():
+    """Get all companies and their current ratings from the database."""
+    try:
+        conn = _connect()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    """
+                    SELECT DISTINCT company, company_rating 
+                    FROM scraped_jobs 
+                    WHERE company IS NOT NULL AND company_rating IS NOT NULL
+                    ORDER BY company
+                    """
+                )
+                rows = cur.fetchall()
+        finally:
+            conn.close()
+        
+        ratings = {row[0]: row[1] for row in rows}
+        return {"ratings": ratings}
+    except Exception as e:
+        logger.error(f"get-company-ratings: failed: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+