"""Shared HTTP session helpers for all monitors. A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's request_with_retries, adding per-request proxy rotation and old-style JSON/text return values so monitor call sites change as little as possible. """ import asyncio import logging import random from aioproxies import AioProxies from aioweb import Response from aioweb_tls import CurlCffi, TLSSession from monitor_utils.constants import ( CLIENT_IDENTIFIERS, MAX_CURL_HANDLES, RETRY_LIMIT, SESSION_CLOSE_TIMEOUT_SECONDS, TIMEOUT_SECONDS, ) LOGGER = logging.getLogger(__name__) class BoundedCloseSession(TLSSession): """A TLSSession whose close cannot outlive SESSION_CLOSE_TIMEOUT_SECONDS. curl_cffi's close awaits the multi handle with no timeout of its own, so a wedged transfer parks whatever is closing the session. A caller-side budget cannot cover this reliably: asyncio.timeout cancels its task once, so a cycle budget that has already fired leaves the close running as a fresh, uncancelled await, and the process parks in the `async with` exit that was meant to reclaim it. Bounding it here keeps that guarantee attached to the session itself, so every `async with get_default_session(...)` gets it without the call site having to arrange anything. Abandoning a session leaks its handle until the process restarts, which is strictly better than a wedged monitor. Known inconsistency, deliberate: an abandoned session reports itself closed while its curl handle is still open. TLSSession.close sets _closed before awaiting the backend, so _is_closed() returns True on this path, and since aioweb's __del__ only warns when _is_closed() is False, its leak warning is suppressed for exactly the sessions that leaked. The "session close exceeded budget" error logged below is therefore the only signal that it happened — alert on it rather than on the library's warning. Correcting the flag would mean overriding _is_closed too, and aioweb_tls is a third-party library: this class deliberately overrides one documented lifecycle method and nothing private, so an upstream release has one narrow way to break it. A stale flag on a session discarded the same instant is the cheaper side of that trade. """ async def close(self) -> None: try: async with asyncio.timeout(SESSION_CLOSE_TIMEOUT_SECONDS): await super().close() except TimeoutError: LOGGER.error( "session close exceeded budget, abandoning session", extra={"timeout": SESSION_CLOSE_TIMEOUT_SECONDS}, ) def get_default_session( proxies: AioProxies, headers: dict | None = None, timeout: int = TIMEOUT_SECONDS, max_clients: int = MAX_CURL_HANDLES, ) -> TLSSession: """Return a TLSSession impersonating a random client fingerprint. Fingerprint rotation lives here so every monitor rotates the same way; the caller owns the session lifetime via ``async with``. Cookies are not set here: aioweb has no constructor cookie path, and monitors usually obtain them at runtime, so call session.set_cookie(name, value, domain=...) where they are fetched (e.g. a pre_task_func) instead. Args: proxies: Pool attached to the session as proxy_pool, which send_request rotates through per request. Required so a session can never send un-proxied. headers: Default headers merged into every request by aioweb. Use for monitor-wide static headers; per-request headers still go through send_request. The curl_cffi impersonation already supplies a matching User-Agent, so one is not defaulted here. timeout: Total request timeout in seconds, defaulting to TIMEOUT_SECONDS. aioweb's own default is only 10s, so this restores the wider budget the old sessions used and slow retailers depend on. max_clients: Size of the curl handle pool, defaulting to MAX_CURL_HANDLES. This is a hard concurrency ceiling: a request finding the pool empty waits in pop_curl's untimed pool.get(), and `timeout` does not cover that wait, only the transfer after checkout. Keep it at or above the monitor's real fan-out (the sum of its pagination semaphores plus any unsemaphored per-cycle requests) so the wait is never reachable. Returns: A BoundedCloseSession (a TLSSession whose close is bounded) over a curl_cffi backend with a random impersonation, with the proxy pool attached as proxy_pool. """ impersonate = random.choice(CLIENT_IDENTIFIERS) session = BoundedCloseSession( backend=CurlCffi(impersonate=impersonate), headers=headers, timeout=timeout, max_clients=max_clients, ) session.proxy_pool = proxies return session async def send_request( session: TLSSession, url: str, method: str, *, params: dict | None = None, json_data: dict | None = None, data: dict | None = None, headers: dict | None = None, ) -> Response: """Send a request through a TLSSession with retries and proxy rotation. Wraps aioweb's request_with_retries, which retries transport errors, timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with exponential backoff. The next proxy is drawn from the pool per call and applied to both schemes. A non-2xx response or a fully failed request raises AiowebError, an expected task error, so a failed fetch surfaces the same way the old HTTPError path did. On success the aioweb Response is returned for the caller to read via response.json(), response.text(), etc. Args: session: The TLSSession to send through. url: Request URL. method: HTTP method. params: Query parameters. json_data: JSON body. A dict here is sent as JSON with the application/json content-type set by aioweb. data: Raw request body, passed through unchanged (e.g. a url-encoded form string). The caller owns the matching content-type via headers. A dict is JSON-encoded by aioweb, so use json_data for JSON bodies. headers: Per-request headers merged over the session defaults. Returns: The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with response.json() (dict/list, or None if not valid JSON) or response.text(). Raises: AiowebError: On a non-2xx response or after all retries fail. """ proxy_url = session.proxy_pool.next().url() proxy = {"http": proxy_url, "https": proxy_url} body = json_data if json_data is not None else data response = await session.request_with_retries( method.upper(), url, params=params, data=body, headers=headers, proxies=proxy, attempts=RETRY_LIMIT, ) response.raise_for_status() return response