149 lines
3.9 KiB
Python
149 lines
3.9 KiB
Python
# ik ua/tls doesn't match headers/header order but targeted antibots don't seem to care might fix later
|
|
import asyncio
|
|
import random
|
|
import json
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
from typing import AsyncGenerator
|
|
|
|
import aiohttp
|
|
import requests
|
|
from curl_cffi.requests import AsyncSession
|
|
from curl_cffi.requests.errors import RequestsError
|
|
from curl_cffi.requests.exceptions import Timeout as CurlTimeout
|
|
from fake_useragent import FakeUserAgent
|
|
from tenacity import (
|
|
after_log,
|
|
before_log,
|
|
retry,
|
|
retry_if_exception_type,
|
|
stop_after_attempt,
|
|
wait_fixed,
|
|
)
|
|
|
|
from monitor_utils.config import get_proxies
|
|
from monitor_utils.constants import (
|
|
CLIENT_TIMEOUT,
|
|
TIMEOUT_SECONDS,
|
|
CLIENT_IDENTIFIERS,
|
|
RETRY_LIMIT,
|
|
RETRY_WAIT,
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
UA = FakeUserAgent()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def get_base_session(
|
|
headers: dict | None = None,
|
|
cookies: dict | None = None,
|
|
connector: aiohttp.TCPConnector | None = None,
|
|
timeout: aiohttp.ClientTimeout = CLIENT_TIMEOUT,
|
|
) -> AsyncGenerator[aiohttp.ClientSession, None]:
|
|
headers = headers if headers else {}
|
|
headers.setdefault("User-Agent", UA.random)
|
|
session = aiohttp.ClientSession(
|
|
headers=headers, cookies=cookies, connector=connector, timeout=timeout
|
|
)
|
|
try:
|
|
yield session
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def get_tls_session(
|
|
headers: dict | None = None, timeout: int = TIMEOUT_SECONDS
|
|
) -> AsyncGenerator[AsyncSession, None]:
|
|
headers = headers if headers else {}
|
|
headers.setdefault("User-Agent", UA.chrome)
|
|
async with AsyncSession(
|
|
impersonate=random.choice(CLIENT_IDENTIFIERS),
|
|
timeout=timeout,
|
|
headers=headers,
|
|
) as session:
|
|
yield session
|
|
|
|
|
|
def build_session_details(
|
|
session: aiohttp.ClientSession | AsyncSession, headers: dict = None
|
|
):
|
|
# ik headers doesn't match tls i'm lazy and the antibots used don't care
|
|
headers = headers or {}
|
|
headers.setdefault("User-Agent", FakeUserAgent().random)
|
|
session.headers.update(headers)
|
|
|
|
|
|
@retry(
|
|
stop=stop_after_attempt(RETRY_LIMIT),
|
|
wait=wait_fixed(RETRY_WAIT),
|
|
retry=retry_if_exception_type(
|
|
(
|
|
aiohttp.ClientError,
|
|
asyncio.TimeoutError,
|
|
RequestsError,
|
|
requests.Timeout,
|
|
CurlTimeout,
|
|
)
|
|
),
|
|
before=before_log(LOGGER, logging.DEBUG),
|
|
after=after_log(LOGGER, logging.DEBUG),
|
|
reraise=True,
|
|
)
|
|
async def send_request(
|
|
session: aiohttp.ClientSession | AsyncSession,
|
|
url: str,
|
|
method: str,
|
|
params: dict | None = None,
|
|
json_data: dict | None = None,
|
|
data: dict | None = None,
|
|
reraise_json_error: bool = False,
|
|
development_proxy: str | None = None,
|
|
) -> dict | str:
|
|
verify = True
|
|
if development_proxy:
|
|
proxy = development_proxy
|
|
verify = False
|
|
else:
|
|
proxy = random.choice(get_proxies())
|
|
|
|
if isinstance(session, aiohttp.ClientSession):
|
|
async with session.request(
|
|
method=method,
|
|
url=url,
|
|
params=params,
|
|
json=json_data,
|
|
data=data,
|
|
proxy=proxy,
|
|
ssl=verify,
|
|
) as response:
|
|
response.raise_for_status()
|
|
text = await response.text()
|
|
else:
|
|
if (
|
|
url
|
|
== "https://www.hp.com/wcs/resources/store/10151/component/configure/addtocartconfig"
|
|
):
|
|
pass
|
|
|
|
response = await session.request(
|
|
method=method.upper(),
|
|
url=url,
|
|
params=params,
|
|
json=json_data,
|
|
data=data,
|
|
proxy=proxy,
|
|
verify=verify,
|
|
)
|
|
response.raise_for_status()
|
|
text = response.text
|
|
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
if reraise_json_error:
|
|
raise
|
|
|
|
return text
|