rates/backend/services/proxy.py

"""
Shared residential-proxy configuration for every scraper.

Single source of truth for:
  - where proxy config lives (system_config `booking_proxy_*` keys, with
    BOOKING_PROXY_* env fallback),
  - the DataImpulse sticky-session username syntax (LOGIN__cr.<cc>;sessid.<id>),
  - building proxy settings for both Playwright (launch dict) and httpx (URL).

The Booking.com browser scraper (Playwright) and the direct booking-engine
scraper (httpx) both consume this, so proxy logic is never duplicated. Direct
scraping is opt-in via the `direct_scraper_use_proxy` config key (default off —
booking-engine APIs generally don't need it), but the plumbing is ready.
"""
import os
import random
from typing import Optional


def new_session_id() -> str:
    """A fresh sticky-session id. DataImpulse pins one IP per id, so a new id
    yields a new residential IP."""
    return f"hnf{random.randint(100000, 999999)}"


def config_from_env() -> dict:
    """Proxy config from BOOKING_PROXY_* env vars ({} when unset)."""
    host = os.getenv("BOOKING_PROXY_HOST", "").strip()
    if not host:
        return {}
    return {
        'host': host,
        'port': os.getenv("BOOKING_PROXY_PORT", "823").strip(),
        'username': os.getenv("BOOKING_PROXY_USERNAME", "").strip(),
        'password': os.getenv("BOOKING_PROXY_PASSWORD", "").strip(),
        'country': os.getenv("BOOKING_PROXY_COUNTRY", "gb").strip(),
    }


def normalize(raw: dict) -> dict:
    """Turn raw `booking_proxy_*` system_config values into a proxy config dict.
    DB is authoritative when `booking_proxy_enabled` is present; otherwise fall
    back to env. Returns {} when disabled or unconfigured."""
    if 'booking_proxy_enabled' in raw:
        if raw.get('booking_proxy_enabled') != 'true':
            return {}
        return {
            'host': (raw.get('booking_proxy_host') or '').strip(),
            'port': (raw.get('booking_proxy_port') or '823').strip(),
            'username': (raw.get('booking_proxy_username') or '').strip(),
            'password': (raw.get('booking_proxy_password') or '').strip(),
            'country': (raw.get('booking_proxy_country') or 'gb').strip(),
        }
    return config_from_env()


def load_config(db) -> dict:
    """Resolve proxy config using a sync SQLAlchemy session. Returns {} when
    disabled/unconfigured. (Async callers should fetch the rows themselves and
    pass them through normalize().)"""
    from sqlalchemy import text
    rows = db.execute(
        text("SELECT config_key, config_value FROM system_config WHERE config_key LIKE 'booking_proxy_%'")
    ).fetchall()
    return normalize({r.config_key: r.config_value for r in rows})


def is_enabled(cfg: dict) -> bool:
    return bool(cfg.get('host') and cfg.get('username'))


def _provider(cfg: dict) -> str:
    """Detect proxy provider from hostname."""
    return 'iproyal' if 'iproyal' in cfg.get('host', '').lower() else 'dataimpulse'


def _build_auth(cfg: dict, session_id: Optional[str] = None):
    """Return (proxy_user, proxy_password) with session/country options applied.

    DataImpulse: options appended to username  → LOGIN__cr.gb;sessid.ID : PASS
    IPRoyal:     options appended to password  → LOGIN : PASS_country-gb_session-ID_lifetime-30m
    """
    cc = cfg.get('country', 'gb')
    base_user = cfg.get('username', '')
    base_pass = cfg.get('password', '')
    if _provider(cfg) == 'iproyal':
        pwd = f"{base_pass}_country-{cc}"
        if session_id:
            pwd += f"_session-{session_id}_lifetime-30m"
        return base_user, pwd
    else:
        user = f"{base_user}__cr.{cc}"
        if session_id:
            user += f";sessid.{session_id}"
        return user, base_pass


def playwright_proxy(cfg: dict, session_id: Optional[str] = None) -> Optional[dict]:
    """Proxy dict for new_context(proxy=...). None when disabled.

    DataImpulse: credentials embedded in the server URL — separate fields cause
    a ~14s 407 round-trip before DataImpulse issues a challenge.

    IPRoyal: separate username/password fields — URL-embedded credentials are
    not parsed correctly by Chromium for this provider (auth silently dropped).
    IPRoyal responds to the 407 challenge quickly so there is no latency penalty.
    """
    if not is_enabled(cfg):
        return None
    user, pwd = _build_auth(cfg, session_id)
    server = f"http://{cfg['host']}:{cfg['port']}"
    if _provider(cfg) == 'iproyal':
        return {'server': server, 'username': user, 'password': pwd}
    from urllib.parse import quote
    return {
        'server': f"http://{quote(user, safe='')}:{quote(pwd, safe='')}@{cfg['host']}:{cfg['port']}",
    }


def httpx_proxy(cfg: dict, session_id: Optional[str] = None):
    """httpx.Proxy object for AsyncClient(proxy=...). None when disabled."""
    if not is_enabled(cfg):
        return None
    import httpx
    return httpx.Proxy(httpx_proxy_url(cfg, session_id))


def httpx_proxy_url(cfg: dict, session_id: Optional[str] = None) -> Optional[str]:
    """Proxy URL for httpx.AsyncClient(proxy=...). None when disabled."""
    if not is_enabled(cfg):
        return None
    from urllib.parse import quote
    user, pwd = _build_auth(cfg, session_id)
    return f"http://{quote(user, safe='')}:{quote(pwd, safe='')}@{cfg['host']}:{cfg['port']}"


def direct_httpx_proxy(db):
    """httpx.Proxy for the direct booking-engine scraper — only when the proxy is
    configured AND `direct_scraper_use_proxy` is explicitly enabled (default off).
    Uses a fresh session id per call so direct runs spread across IPs."""
    from sqlalchemy import text
    flag = db.execute(
        text("SELECT config_value FROM system_config WHERE config_key = 'direct_scraper_use_proxy'")
    ).fetchone()
    if not flag or flag.config_value != 'true':
        return None
    return httpx_proxy(load_config(db), new_session_id())