mirror of
https://github.com/bohd4nx/FragmentAPI.git
synced 2026-07-25 06:14:29 +00:00
bohd4nx
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from curl_cffi.requests import AsyncSession
|
|
|
|
from pyfragment.core.constants import BASE_HEADERS
|
|
from pyfragment.core.transport import fragment_request, get_fragment_hash
|
|
|
|
if TYPE_CHECKING:
|
|
from pyfragment.client import FragmentClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def raw_api_call(
|
|
cookies: dict[str, Any],
|
|
timeout: float,
|
|
method: str,
|
|
data: dict[str, Any] | None,
|
|
page_url: str,
|
|
headers: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
base = headers if headers is not None else BASE_HEADERS
|
|
payload = {"method": method, **(data or {})}
|
|
call_headers = {**base, "referer": page_url, "x-aj-referer": page_url}
|
|
logger.debug("Starting Fragment API call '%s' on %s", method, page_url)
|
|
try:
|
|
async with AsyncSession(cookies=cookies, timeout=timeout, impersonate="chrome") as session:
|
|
fragment_hash = await get_fragment_hash(session, call_headers, page_url)
|
|
response = await fragment_request(session, fragment_hash, call_headers, payload)
|
|
logger.debug("Completed Fragment API call '%s' with response keys: %s", method, sorted(response.keys()))
|
|
return response
|
|
except Exception:
|
|
logger.exception("Failed to call Fragment API method '%s' on %s", method, page_url)
|
|
raise
|
|
|
|
|
|
class BaseService:
|
|
def __init__(self, client: FragmentClient) -> None:
|
|
self._client = client
|