cores/search/http.py

"""Shared HTTP session — requests with connection pooling + retries."""

from __future__ import annotations

from typing import Optional

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


_session: Optional[requests.Session] = None


def shared_session() -> requests.Session:
    """Return a process-wide shared requests.Session."""
    global _session
    if _session is None:
        _session = _make_session()
    return _session


def _make_session() -> requests.Session:
    session = requests.Session()
    retry = Retry(
        total=2,
        backoff_factor=0.3,
        status_forcelist=[502, 503, 504],
        allowed_methods=["GET", "POST", "HEAD"],
    )
    adapter = HTTPAdapter(pool_connections=10, pool_maxsize=10, max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


def fetch_html(url: str, timeout: int = 15, headers: Optional[dict] = None) -> str:
    """GET a URL and return text."""
    from cores.search.user_agent import random_user_agent
    h = {"User-Agent": random_user_agent()}
    if headers:
        h.update(headers)
    resp = shared_session().get(url, headers=h, timeout=timeout)
    resp.raise_for_status()
    return resp.text


def fetch_bytes(url: str, timeout: int = 15, headers: Optional[dict] = None) -> bytes:
    """GET a URL and return raw bytes."""
    from cores.search.user_agent import random_user_agent
    h = {"User-Agent": random_user_agent()}
    if headers:
        h.update(headers)
    resp = shared_session().get(url, headers=h, timeout=timeout, stream=True)
    resp.raise_for_status()
    return resp.content


def fetch_json(url: str, timeout: int = 15, headers: Optional[dict] = None,
               params: Optional[dict] = None) -> dict:
    """GET a URL and return parsed JSON."""
    from cores.search.user_agent import random_user_agent
    h = {"User-Agent": random_user_agent(), "Accept": "application/json"}
    if headers:
        h.update(headers)
    resp = shared_session().get(url, headers=h, timeout=timeout, params=params)
    resp.raise_for_status()
    return resp.json()