diff --git a/worker/src/scraper.py b/worker/src/scraper.py index d24d585..f1d96ff 100644 --- a/worker/src/scraper.py +++ b/worker/src/scraper.py @@ -1,13 +1,195 @@ import asyncio import logging -from datetime import datetime, timezone +import os +from datetime import datetime from typing import Any -from urllib.parse import quote_plus import httpx +from db import get_pool +from settings import get_effective_proxy_url, get_proxy_url_from_env, proxy_available + logger = logging.getLogger(__name__) +_client: httpx.AsyncClient | None = None +_client_proxy_url: str | None = None +_proxy_ip_logged = False +_system_public_ip: str | None = None +_proxy_public_ip: str | None = None +_last_used_public_ip: str | None = None +_proxy_enabled_effective: bool | None = None + + +def _get_proxy_url() -> str | None: + return get_proxy_url_from_env() + + +async def _get_effective_proxy_url() -> str | None: + global _proxy_enabled_effective + + pool = await get_pool() + proxy_url = await get_effective_proxy_url(pool) + _proxy_enabled_effective = proxy_url is not None + return proxy_url + + +def _redact_proxy_url(proxy_url: str) -> str: + try: + parsed = httpx.URL(proxy_url) + host = parsed.host or "unknown-host" + port = parsed.port or 80 + user = parsed.username or "unknown-user" + return f"{host}:{port} (user={user}, credentials=set)" + except Exception: + return "" + + +async def _fetch_public_ip(client: httpx.AsyncClient) -> str | None: + try: + resp = await client.get("https://api.ipify.org?format=json") + resp.raise_for_status() + data = resp.json() + ip = data.get("ip") + return str(ip) if ip else None + except Exception as exc: + logger.warning("Could not resolve public IP: %s", exc) + return None + + +async def _log_proxy_ip_comparison(proxy_url: str) -> None: + global _proxy_ip_logged, _system_public_ip, _proxy_public_ip, _last_used_public_ip + + if _proxy_ip_logged: + return + + _proxy_ip_logged = True + + direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False) + proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url) + + try: + direct_ip = await _fetch_public_ip(direct_client) + proxy_ip = await _fetch_public_ip(proxy_client) + _system_public_ip = direct_ip + _proxy_public_ip = proxy_ip + _last_used_public_ip = proxy_ip or direct_ip + logger.info( + "HTTPS proxy enabled: %s | public_ip_without_proxy=%s | public_ip_with_proxy=%s", + _redact_proxy_url(proxy_url), + direct_ip or "unknown", + proxy_ip or "unknown", + ) + finally: + await direct_client.aclose() + await proxy_client.aclose() + + +async def get_client() -> httpx.AsyncClient: + """Return a shared AsyncClient with keepalive connection pool.""" + global _client, _client_proxy_url, _system_public_ip, _last_used_public_ip + + proxy_url = await _get_effective_proxy_url() + + if _client is not None and not _client.is_closed and _client_proxy_url != proxy_url: + await _client.aclose() + logger.info( + "Recreated httpx client because proxy changed: %s -> %s", + "enabled" if _client_proxy_url else "disabled", + "enabled" if proxy_url else "disabled", + ) + _client = None + + if _client is None or _client.is_closed: + max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10")) + max_keepalive = int(os.getenv("HTTP_KEEPALIVE_CONNECTIONS", "5")) + + if proxy_url: + await _log_proxy_ip_comparison(proxy_url) + elif _system_public_ip is None: + direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False) + try: + _system_public_ip = await _fetch_public_ip(direct_client) + _last_used_public_ip = _system_public_ip + finally: + await direct_client.aclose() + + _client = httpx.AsyncClient( + timeout=float(os.getenv("HTTP_TIMEOUT_S", "30.0")), + limits=httpx.Limits( + max_connections=max_conns, + max_keepalive_connections=max_keepalive, + keepalive_expiry=60, + ), + proxy=proxy_url, + trust_env=False, + ) + _client_proxy_url = proxy_url + logger.info( + "Created httpx client: max_conns=%d, keepalive=%d, proxy=%s", + max_conns, + max_keepalive, + "enabled" if proxy_url else "disabled", + ) + + return _client + + +async def refresh_network_status() -> dict[str, str | bool | None]: + """Ensure network status has best-effort values even before first scrape.""" + global _system_public_ip, _proxy_public_ip, _last_used_public_ip, _proxy_enabled_effective + + proxy_url = await _get_effective_proxy_url() + _proxy_enabled_effective = proxy_url is not None + + if _system_public_ip is None: + direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False) + try: + _system_public_ip = await _fetch_public_ip(direct_client) + finally: + await direct_client.aclose() + + if proxy_url and _proxy_public_ip is None: + proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url) + try: + _proxy_public_ip = await _fetch_public_ip(proxy_client) + finally: + await proxy_client.aclose() + + if proxy_url: + _last_used_public_ip = _proxy_public_ip or _system_public_ip + else: + _last_used_public_ip = _system_public_ip + + return get_network_status() + + +def get_network_status() -> dict[str, str | bool | None]: + """Return best-effort network status for UI display.""" + available = proxy_available() + proxy_enabled = bool(_proxy_enabled_effective) if _proxy_enabled_effective is not None else available + last_used = _last_used_public_ip + if last_used is None: + last_used = _proxy_public_ip if proxy_enabled else _system_public_ip + + return { + "proxy_available": available, + "proxy_enabled": proxy_enabled, + "system_public_ip": _system_public_ip, + "proxy_public_ip": _proxy_public_ip, + "last_used_public_ip": last_used, + } + + +async def close_client() -> None: + """Close the shared AsyncClient. Call during shutdown.""" + global _client, _client_proxy_url + if _client and not _client.is_closed: + await _client.aclose() + logger.info("Closed httpx client") + _client = None + _client_proxy_url = None + + _API_URL = ( "https://www.willhaben.at/webapi/ad-search/search/atz/seo/" "kaufen-und-verkaufen/marktplatz" @@ -21,25 +203,27 @@ _HEADERS = { async def fetch_ads(keyword: str) -> tuple[list[dict[str, Any]], int]: + """Fetch the latest 30 ads for a keyword (single page, newest first).""" params = { "keyword": keyword, "rows": 30, "sort": 1, } - async with httpx.AsyncClient(timeout=30.0) as client: - for attempt in range(1, 4): - try: - resp = await client.get(_API_URL, headers=_HEADERS, params=params) - resp.raise_for_status() - data = resp.json() - break - except Exception as exc: - logger.warning("fetch_ads attempt %d failed: %s", attempt, exc) - if attempt < 3: - await asyncio.sleep(2 ** attempt) - continue - raise + client = await get_client() + + for attempt in range(1, 4): + try: + resp = await client.get(_API_URL, headers=_HEADERS, params=params) + resp.raise_for_status() + data = resp.json() + break + except Exception as exc: + logger.warning("fetch_ads attempt %d failed for '%s': %s", attempt, keyword, exc) + if attempt < 3: + await asyncio.sleep(2 ** attempt) + continue + raise ads_raw = data.get("advertSummaryList", {}).get("advertSummary", []) total_hits = int(data.get("rowsFound", 0)) @@ -111,4 +295,4 @@ def extract_ad_fields(ad_dict: dict[str, Any]) -> dict[str, Any]: "main_image_url": main_image_url, "postcode": postcode, "modified_at": modified_at, - } + } \ No newline at end of file