FOSS/FragmentAPI
0
0
Fork 0
mirror of https://github.com/bohd4nx/FragmentAPI.git synced 2026-07-25 06:14:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor: Replace httpx with curl-cffi for HTTP requests and update dependencies

Browse Source
This commit is contained in:
bohd4nx
2026-07-05 16:13:16 +03:00
parent d78be012a3
commit fd31342d03
5 changed files with 22 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyfragment/core/constants.py
+2 -3
View File
@@ -24,15 +24,14 @@ BASE_HEADERS: dict[str, str] = {
"content-type": "application/x-www-form-urlencoded; charset=UTF-8",
"origin": FRAGMENT_BASE_URL,
"priority": "u=1, i",
"sec-ch-ua": '"Google Chrome";v="147", "Not.A/Brand";v="8", "Chromium";v="147"',
"sec-ch-ua": '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"',
"sec-ch-ua-mobile": "?1",
"sec-ch-ua-platform": '"Android"',
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"user-agent": (
"Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/147.0.0.0 Mobile Safari/537.36"
"Mozilla/5.0 (Linux; Android 15; Pixel 9) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Mobile Safari/537.36"
),
"x-requested-with": "XMLHttpRequest",
}
pyfragment/core/transport.py
+13 -23
View File
@@ -5,35 +5,25 @@ import random
import re
from typing import Any, cast
import httpx
from curl_cffi.requests import AsyncSession, Response
from pyfragment.core.constants import DEFAULT_TIMEOUT, FRAGMENT_BASE_URL
from pyfragment.core.constants import FRAGMENT_BASE_URL
from pyfragment.exceptions import FragmentPageError, ParseError
async def get_fragment_hash(
cookies: dict[str, Any],
session: AsyncSession[Any], # type: ignore[type-arg]
headers: dict[str, str],
page_url: str,
timeout: float = DEFAULT_TIMEOUT,
) -> str:
page_headers = {
k: v
for k, v in headers.items()
if k not in ("accept", "accept-encoding", "content-type", "x-requested-with", "x-aj-referer")
}
page_headers.update(
{
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"referer": f"{FRAGMENT_BASE_URL}/",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"upgrade-insecure-requests": "1",
}
)
# 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
async with httpx.AsyncClient(cookies=cookies, timeout=timeout) as session:
response = await session.get(page_url, headers=page_headers)
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))
@@ -45,15 +35,15 @@ async def get_fragment_hash(
return match.group(1)
def parse_json_response(response: httpx.Response, context: str) -> dict[str, Any]:
def parse_json_response(response: Response, context: str) -> dict[str, Any]:
try:
return cast(dict[str, Any], response.json())
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: httpx.AsyncClient,
session: AsyncSession[Any], # type: ignore[type-arg]
fragment_hash: str,
headers: dict[str, str],
data: dict[str, Any],