mirror of
https://github.com/bohd4nx/FragmentAPI.git
synced 2026-07-25 06:14:29 +00:00
bohd4nx
8163ed1a9d
- Implemented `search_gifts` method for searching gifts in the Fragment marketplace. - Implemented `search_numbers` method for searching anonymous Telegram numbers. - Added new example scripts for searching gifts, numbers, and usernames. - Updated HTML parsing logic to handle gift items and number listings. - Added unit tests for the new search methods and updated existing tests for usernames and numbers. - Removed obsolete test file for auction searches.
68 lines
2.6 KiB
Python
68 lines
2.6 KiB
Python
from typing import TYPE_CHECKING, Any
|
|
|
|
import httpx
|
|
|
|
from pyfragment.types import FragmentAPIError, FragmentError, UnexpectedError
|
|
from pyfragment.types.constants import FRAGMENT_BASE_URL
|
|
from pyfragment.types.results import UsernamesResult
|
|
from pyfragment.utils import fragment_request, get_fragment_hash, make_headers, parse_auction_rows
|
|
|
|
if TYPE_CHECKING:
|
|
from pyfragment.client import FragmentClient
|
|
|
|
HEADERS: dict[str, str] = make_headers(FRAGMENT_BASE_URL)
|
|
|
|
|
|
async def search_usernames(
|
|
client: "FragmentClient",
|
|
query: str = "",
|
|
sort: str | None = None,
|
|
filter: str | None = None,
|
|
offset_id: str | None = None,
|
|
) -> UsernamesResult:
|
|
"""Search the Fragment marketplace for Telegram usernames.
|
|
|
|
Args:
|
|
client: Authenticated :class:`FragmentClient` instance.
|
|
query: Search text (e.g. ``"durov"``). Omit or pass ``""`` to browse all.
|
|
sort: Sort order — ``"price_desc"``, ``"price_asc"``, ``"listed"``, or
|
|
``"ending"``. Omit to use Fragment's default ordering.
|
|
filter: Filter results — ``"auction"``, ``"sale"``, ``"sold"``, or ``""``
|
|
(available items). Omit to return all.
|
|
offset_id: Pagination cursor from a previous :class:`UsernamesResult`.
|
|
Pass ``next_offset_id`` to fetch the next page.
|
|
|
|
Returns:
|
|
:class:`UsernamesResult` with ``items`` (parsed list of item dicts) and
|
|
``next_offset_id`` (``None`` when there are no more pages).
|
|
|
|
Raises:
|
|
FragmentAPIError: If the Fragment API returns an error.
|
|
UnexpectedError: For any other unexpected failure.
|
|
"""
|
|
data: dict[str, Any] = {"method": "searchAuctions", "type": "usernames", "query": query}
|
|
if sort is not None:
|
|
data["sort"] = sort
|
|
if filter is not None:
|
|
data["filter"] = filter
|
|
if offset_id is not None:
|
|
data["offset_id"] = offset_id
|
|
|
|
try:
|
|
fragment_hash = await get_fragment_hash(client.cookies, HEADERS, FRAGMENT_BASE_URL, client.timeout)
|
|
async with httpx.AsyncClient(cookies=client.cookies, timeout=client.timeout) as session:
|
|
result = await fragment_request(session, fragment_hash, HEADERS, data)
|
|
|
|
if result.get("error"):
|
|
raise FragmentAPIError(result["error"])
|
|
|
|
items = parse_auction_rows(result.get("html") or "")
|
|
raw_noi = result.get("next_offset_id")
|
|
next_offset_id = str(raw_noi) if raw_noi else None
|
|
return UsernamesResult(items=items, next_offset_id=next_offset_id)
|
|
|
|
except FragmentError:
|
|
raise
|
|
except Exception as exc:
|
|
raise UnexpectedError(UnexpectedError.UNEXPECTED.format(exc=exc)) from exc
|