mirror of
https://github.com/bohd4nx/FragmentAPI.git
synced 2026-07-25 06:14:29 +00:00
bohd4nx
66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import random
|
|
import re
|
|
from typing import Any, cast
|
|
|
|
from curl_cffi.requests import AsyncSession, Response
|
|
|
|
from pyfragment.core.constants import FRAGMENT_BASE_URL
|
|
from pyfragment.exceptions import FragmentPageError, ParseError
|
|
|
|
|
|
async def get_fragment_hash(
|
|
session: AsyncSession[Any],
|
|
headers: dict[str, str],
|
|
page_url: str,
|
|
) -> str:
|
|
# Derive the natural referer: strip the last path segment (e.g. /stars/buy → /stars)
|
|
parent_url = page_url.rsplit("/", 1)[0] or FRAGMENT_BASE_URL
|
|
|
|
page_headers = {k: v for k, v in headers.items() if k not in ("content-type", "origin")}
|
|
page_headers["referer"] = parent_url
|
|
page_headers["x-aj-referer"] = parent_url
|
|
|
|
response = await session.get(page_url, headers=page_headers)
|
|
|
|
if response.status_code != 200:
|
|
raise FragmentPageError(FragmentPageError.BAD_STATUS.format(status=response.status_code, url=page_url))
|
|
|
|
match = re.search(r"(?:https://fragment\.com)?\\?/api\?hash=([a-f0-9]+)", response.text)
|
|
if not match:
|
|
raise FragmentPageError(FragmentPageError.NOT_FOUND.format(url=page_url))
|
|
|
|
return match.group(1)
|
|
|
|
|
|
def parse_json_response(response: Response, context: str) -> dict[str, Any]:
|
|
try:
|
|
return cast(dict[str, Any], response.json()) # type: ignore[no-untyped-call]
|
|
except Exception as exc:
|
|
raise ParseError(ParseError.UNPARSEABLE.format(context=context, exc=exc)) from exc
|
|
|
|
|
|
async def fragment_request(
|
|
session: AsyncSession[Any],
|
|
fragment_hash: str,
|
|
headers: dict[str, str],
|
|
data: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
for attempt in range(3):
|
|
resp = await session.post(
|
|
f"{FRAGMENT_BASE_URL}/api?hash={fragment_hash}",
|
|
headers=headers,
|
|
data=data,
|
|
)
|
|
if resp.status_code == 429 and attempt < 2:
|
|
await asyncio.sleep(1 + attempt + random.uniform(0, 0.5))
|
|
continue
|
|
if resp.status_code != 200:
|
|
raise FragmentPageError(
|
|
FragmentPageError.BAD_STATUS.format(status=resp.status_code, url=f"{FRAGMENT_BASE_URL}/api")
|
|
)
|
|
return parse_json_response(resp, data.get("method", "request"))
|
|
raise FragmentPageError(FragmentPageError.BAD_STATUS.format(status=429, url=f"{FRAGMENT_BASE_URL}/api"))
|