"""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 curl_cffi.requests.impersonate import BrowserTypeLiteral 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, impersonate: BrowserTypeLiteral | None = None, ) -> 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. impersonate: A curl_cffi profile to pin instead of drawing one from CLIENT_IDENTIFIERS, for a retailer whose bot protection only tolerates some profiles from a given IP (B&H's Cloudflare refuses every chrome profile on a flagged IP but still accepts safari). 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 = impersonate or 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 def get_response_json( response: Response, *, logger: logging.Logger = LOGGER, expected: type | tuple[type, ...] = dict, **context, ) -> dict | list | None: """Parse a response body as JSON, or log it and return None. aioweb's response.json() yields None for a non-JSON body, so a single isinstance check covers both the empty/invalid case and a body of the wrong shape. Callers stop on None the same way they did with the inline check this replaces. Args: response: A 2xx aioweb Response from send_request. logger: Logger the warning is emitted on; pass the monitor's own so the record stays attributed to that monitor. expected: Type(s) the parsed body must be an instance of. **context: Extra fields for the warning record (category, page, ...). Returns: The parsed body when it matches `expected`, else None. """ data = response.json() if isinstance(data, expected): return data logger.warning( "unexpected response", extra={ **context, "body_type": type(data).__name__, "body": str(data)[:200], }, ) return None