mirror of
https://github.com/bohd4nx/FragmentAPI.git
synced 2026-07-25 06:14:29 +00:00
bohd4nx
41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import httpx
|
|
|
|
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,
|
|
) -> dict[str, Any]:
|
|
payload = {"method": method, **(data or {})}
|
|
headers = {**BASE_HEADERS, "referer": page_url, "x-aj-referer": page_url}
|
|
logger.debug("Starting Fragment API call '%s' on %s", method, page_url)
|
|
try:
|
|
async with httpx.AsyncClient(cookies=cookies, timeout=timeout) as session:
|
|
fragment_hash = await get_fragment_hash(cookies, headers, page_url, timeout)
|
|
response = await fragment_request(session, fragment_hash, 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
|