Refactor methods for Telegram Premium and Stars giveaways

- Added new methods for handling giveaways: `giveaway_premium` and `giveaway_stars`.
- Refactored existing purchase methods into separate files: `purchase_premium.py` and `purchase_stars.py`.
- Removed old premium and stars methods from the codebase.
- Updated `__init__.py` to include new giveaway methods in the public API.
- Introduced new result types for giveaways: `PremiumGiveawayResult` and `StarsGiveawayResult`.
- Updated constants for new giveaway pages.
- Enhanced error handling for configuration and user validation in giveaway methods.
- Updated tests to cover new functionality and refactored existing tests to match new method locations.
This commit is contained in:
bohd4nx
2026-03-21 16:05:38 +02:00
parent 0af5ce5935
commit 902b692dcb
17 changed files with 483 additions and 53 deletions
+4
View File
@@ -16,7 +16,9 @@ from pyfragment.types import (
FragmentPageError,
OperationError,
ParseError,
PremiumGiveawayResult,
PremiumResult,
StarsGiveawayResult,
StarsResult,
TransactionError,
UnexpectedError,
@@ -32,7 +34,9 @@ __all__ = [
"__version__",
"FragmentClient",
"AdsTopupResult",
"PremiumGiveawayResult",
"PremiumResult",
"StarsGiveawayResult",
"StarsResult",
"WalletInfo",
"ClientError",
+47 -6
View File
@@ -1,9 +1,11 @@
import json
from typing import cast
from pyfragment.methods.premium import purchase_premium
from pyfragment.methods.stars import purchase_stars
from pyfragment.methods.ton import topup_ton
from pyfragment.methods.giveaway_premium import giveaway_premium
from pyfragment.methods.giveaway_stars import giveaway_stars
from pyfragment.methods.purchase_premium import purchase_premium
from pyfragment.methods.purchase_stars import purchase_stars
from pyfragment.methods.topup_ton import topup_ton
from pyfragment.types import (
AdsTopupResult,
ConfigurationError,
@@ -13,6 +15,7 @@ from pyfragment.types import (
WalletInfo,
)
from pyfragment.types.constants import DEFAULT_TIMEOUT, REQUIRED_COOKIE_KEYS, SUPPORTED_WALLET_VERSIONS, WalletVersion
from pyfragment.types.results import PremiumGiveawayResult, StarsGiveawayResult
from pyfragment.utils.wallet import get_wallet_info
@@ -108,7 +111,7 @@ class FragmentClient:
show_sender: Show your name as the sender. Defaults to ``True``.
Returns:
:class:`PremiumResult` with ``transaction_id``, ``username``, ``months``, ``timestamp``.
:class:`PremiumResult` with ``transaction_id``, ``username``, and ``months``.
"""
return await purchase_premium(self, username, months, show_sender)
@@ -121,7 +124,7 @@ class FragmentClient:
show_sender: Show your name as the gift sender. Defaults to ``True``.
Returns:
:class:`StarsResult` with ``transaction_id``, ``username``, ``stars``, ``timestamp``.
:class:`StarsResult` with ``transaction_id``, ``username``, and ``stars``.
"""
return await purchase_stars(self, username, amount, show_sender)
@@ -134,7 +137,7 @@ class FragmentClient:
show_sender: Show your name as the sender. Defaults to ``True``.
Returns:
:class:`AdsTopupResult` with ``transaction_id``, ``username``, ``amount``, ``timestamp``.
:class:`AdsTopupResult` with ``transaction_id``, ``username``, and ``amount``.
"""
return await topup_ton(self, username, amount, show_sender)
@@ -146,3 +149,41 @@ class FragmentClient:
(``"active"``, ``"uninit"``, or ``"frozen"``), and ``balance`` in TON.
"""
return await get_wallet_info(self)
async def giveaway_stars(
self,
channel: str,
winners: int,
amount: int,
) -> StarsGiveawayResult:
"""Run a Telegram Stars giveaway for a channel.
Args:
channel: Channel username (with or without ``@``).
winners: Number of winners — integer from ``1`` to ``5``.
amount: Stars each winner receives — integer from ``500`` to ``1 000 000``.
Returns:
:class:`StarsGiveawayResult` with ``transaction_id``, ``channel``,
``winners``, and ``amount``.
"""
return await giveaway_stars(self, channel, winners, amount)
async def giveaway_premium(
self,
channel: str,
winners: int,
months: int = 3,
) -> PremiumGiveawayResult:
"""Run a Telegram Premium giveaway for a channel.
Args:
channel: Channel username (with or without ``@``).
winners: Number of winners — positive integer.
months: Premium duration per winner — ``3``, ``6``, or ``12``. Defaults to ``3``.
Returns:
:class:`PremiumGiveawayResult` with ``transaction_id``, ``channel``,
``winners``, and ``amount``.
"""
return await giveaway_premium(self, channel, winners, months)
+6 -4
View File
@@ -1,5 +1,7 @@
from pyfragment.methods.premium import purchase_premium
from pyfragment.methods.stars import purchase_stars
from pyfragment.methods.ton import topup_ton
from pyfragment.methods.giveaway_premium import giveaway_premium
from pyfragment.methods.giveaway_stars import giveaway_stars
from pyfragment.methods.purchase_premium import purchase_premium
from pyfragment.methods.purchase_stars import purchase_stars
from pyfragment.methods.topup_ton import topup_ton
__all__ = ["purchase_premium", "purchase_stars", "topup_ton"]
__all__ = ["giveaway_premium", "giveaway_stars", "purchase_premium", "purchase_stars", "topup_ton"]
+139
View File
@@ -0,0 +1,139 @@
import json
from typing import TYPE_CHECKING
import httpx
from pyfragment.types import (
ConfigurationError,
FragmentAPIError,
FragmentError,
UnexpectedError,
UserNotFoundError,
)
from pyfragment.types.constants import BASE_HEADERS, DEVICE, PREMIUM_GIVEAWAY_PAGE
from pyfragment.types.results import PremiumGiveawayResult
from pyfragment.utils import (
execute_transaction_request,
fragment_post,
get_account_info,
get_fragment_hash,
process_transaction,
)
if TYPE_CHECKING:
from pyfragment.client import FragmentClient
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": PREMIUM_GIVEAWAY_PAGE,
"x-aj-referer": PREMIUM_GIVEAWAY_PAGE,
}
async def _search_recipient(
session: httpx.AsyncClient,
fragment_hash: str,
channel: str,
winners: int,
months: int,
) -> str:
result = await fragment_post(
session,
fragment_hash,
HEADERS,
{
"query": channel,
"quantity": winners,
"months": months,
"method": "searchPremiumGiveawayRecipient",
},
)
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(UserNotFoundError.NOT_FOUND.format(username=channel))
return recipient
async def _init_request(
session: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
winners: int,
months: int,
) -> str:
result = await fragment_post(
session,
fragment_hash,
HEADERS,
{
"recipient": recipient,
"quantity": str(winners),
"months": str(months),
"method": "initGiveawayPremiumRequest",
},
)
req_id = result.get("req_id")
if not req_id:
raise FragmentAPIError(FragmentAPIError.NO_REQUEST_ID.format(context="Premium giveaway"))
return req_id
async def giveaway_premium(
client: "FragmentClient",
channel: str,
winners: int,
months: int = 3,
) -> PremiumGiveawayResult:
"""Run a Telegram Premium giveaway for a channel.
Args:
client: Authenticated :class:`FragmentClient` instance.
channel: Channel username (with or without ``@``).
winners: Number of winners — integer from ``1`` to ``24 000``.
months: Premium duration per winner — ``3``, ``6``, or ``12``. Defaults to ``3``.
Returns:
:class:`PremiumGiveawayResult` with ``transaction_id``, ``channel``,
``winners``, and ``months``.
Raises:
ConfigurationError: If ``winners`` is not 124 000 or ``months`` is not 3, 6, or 12.
UserNotFoundError: If the channel is not found on Fragment.
FragmentAPIError: If the Fragment API returns an error.
UnexpectedError: For any other unexpected failure.
"""
if not isinstance(winners, int) or not (1 <= winners <= 24_000):
raise ConfigurationError(ConfigurationError.INVALID_WINNERS_PREMIUM)
if months not in (3, 6, 12):
raise ConfigurationError(ConfigurationError.INVALID_MONTHS)
try:
fragment_hash = await get_fragment_hash(client.cookies, HEADERS, PREMIUM_GIVEAWAY_PAGE, client.timeout)
account = await get_account_info(client)
async with httpx.AsyncClient(cookies=client.cookies, timeout=client.timeout) as session:
recipient = await _search_recipient(session, fragment_hash, channel, winners, months)
req_id = await _init_request(session, fragment_hash, recipient, winners, months)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"method": "getGiveawayPremiumLink",
}
transaction = await execute_transaction_request(session, HEADERS, tx_data, fragment_hash)
tx_hash = await process_transaction(client, transaction)
return PremiumGiveawayResult(
transaction_id=tx_hash,
channel=channel,
winners=winners,
amount=months,
)
except FragmentError:
raise
except Exception as exc:
raise UnexpectedError(UnexpectedError.UNEXPECTED.format(exc=exc)) from exc
+135
View File
@@ -0,0 +1,135 @@
import json
from typing import TYPE_CHECKING
import httpx
from pyfragment.types import (
ConfigurationError,
FragmentAPIError,
FragmentError,
UnexpectedError,
UserNotFoundError,
)
from pyfragment.types.constants import BASE_HEADERS, DEVICE, STARS_GIVEAWAY_PAGE
from pyfragment.types.results import StarsGiveawayResult
from pyfragment.utils import (
execute_transaction_request,
fragment_post,
get_account_info,
get_fragment_hash,
process_transaction,
)
if TYPE_CHECKING:
from pyfragment.client import FragmentClient
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": STARS_GIVEAWAY_PAGE,
"x-aj-referer": STARS_GIVEAWAY_PAGE,
}
async def _search_recipient(
session: httpx.AsyncClient,
fragment_hash: str,
channel: str,
) -> str:
result = await fragment_post(
session,
fragment_hash,
HEADERS,
{
"query": channel,
"method": "searchStarsGiveawayRecipient",
},
)
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(UserNotFoundError.NOT_FOUND.format(username=channel))
return recipient
async def _init_request(
session: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
winners: int,
amount: int,
) -> str:
result = await fragment_post(
session,
fragment_hash,
HEADERS,
{
"recipient": recipient,
"quantity": str(winners),
"stars": str(amount),
"method": "initGiveawayStarsRequest",
},
)
req_id = result.get("req_id")
if not req_id:
raise FragmentAPIError(FragmentAPIError.NO_REQUEST_ID.format(context="Stars giveaway"))
return req_id
async def giveaway_stars(
client: "FragmentClient",
channel: str,
winners: int,
amount: int,
) -> StarsGiveawayResult:
"""Run a Telegram Stars giveaway for a channel.
Args:
client: Authenticated :class:`FragmentClient` instance.
channel: Channel username (with or without ``@``).
winners: Number of winners — integer from ``1`` to ``5``.
amount: Stars each winner receives — integer from ``500`` to ``1 000 000``.
Returns:
:class:`StarsGiveawayResult` with ``transaction_id``, ``channel``,
``winners``, and ``amount``.
Raises:
ConfigurationError: If ``winners`` is not 15 or ``amount`` is not 5001 000 000.
UserNotFoundError: If the channel is not found on Fragment.
FragmentAPIError: If the Fragment API returns an error.
UnexpectedError: For any other unexpected failure.
"""
if not isinstance(winners, int) or not (1 <= winners <= 5):
raise ConfigurationError(ConfigurationError.INVALID_WINNERS_STARS)
if not isinstance(amount, int) or not (500 <= amount <= 1_000_000):
raise ConfigurationError(ConfigurationError.INVALID_STARS_PER_WINNER)
try:
fragment_hash = await get_fragment_hash(client.cookies, HEADERS, STARS_GIVEAWAY_PAGE, client.timeout)
account = await get_account_info(client)
async with httpx.AsyncClient(cookies=client.cookies, timeout=client.timeout) as session:
recipient = await _search_recipient(session, fragment_hash, channel)
req_id = await _init_request(session, fragment_hash, recipient, winners, amount)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"method": "getGiveawayStarsLink",
}
transaction = await execute_transaction_request(session, HEADERS, tx_data, fragment_hash)
tx_hash = await process_transaction(client, transaction)
return StarsGiveawayResult(
transaction_id=tx_hash,
channel=channel,
winners=winners,
amount=amount,
)
except FragmentError:
raise
except Exception as exc:
raise UnexpectedError(UnexpectedError.UNEXPECTED.format(exc=exc)) from exc
@@ -88,6 +88,23 @@ async def _init_request(
async def purchase_premium(client: "FragmentClient", username: str, months: int, show_sender: bool = True) -> PremiumResult:
"""Gift Telegram Premium to a user.
Args:
client: Authenticated :class:`FragmentClient` instance.
username: Recipient's Telegram username (with or without ``@``).
months: Premium duration ``3``, ``6``, or ``12``.
show_sender: Show your name as the gift sender. Defaults to ``True``.
Returns:
:class:`PremiumResult` with ``transaction_id``, ``username``, and ``months``.
Raises:
ConfigurationError: If ``months`` is not ``3``, ``6``, or ``12``.
UserNotFoundError: If the user is not found on Fragment.
FragmentAPIError: If the Fragment API returns an error.
UnexpectedError: For any other unexpected failure.
"""
if months not in (3, 6, 12):
raise ConfigurationError(ConfigurationError.INVALID_MONTHS)
@@ -110,7 +127,7 @@ async def purchase_premium(client: "FragmentClient", username: str, months: int,
transaction = await execute_transaction_request(session, HEADERS, tx_data, fragment_hash)
tx_hash = await process_transaction(client, transaction)
return PremiumResult(transaction_id=tx_hash, username=username, months=months)
return PremiumResult(transaction_id=tx_hash, username=username, amount=months)
except FragmentError:
raise
@@ -75,6 +75,23 @@ async def _init_request(
async def purchase_stars(client: "FragmentClient", username: str, amount: int, show_sender: bool = True) -> StarsResult:
"""Send Telegram Stars to a user.
Args:
client: Authenticated :class:`FragmentClient` instance.
username: Recipient's Telegram username (with or without ``@``).
amount: Number of Stars to send integer from ``50`` to ``1 000 000``.
show_sender: Show your name as the gift sender. Defaults to ``True``.
Returns:
:class:`StarsResult` with ``transaction_id``, ``username``, and ``stars``.
Raises:
ConfigurationError: If ``amount`` is not an integer between 50 and 1 000 000.
UserNotFoundError: If the user is not found on Fragment.
FragmentAPIError: If the Fragment API returns an error.
UnexpectedError: For any other unexpected failure.
"""
if not isinstance(amount, int) or not (50 <= amount <= 1_000_000):
raise ConfigurationError(ConfigurationError.INVALID_STARS_AMOUNT)
@@ -97,7 +114,7 @@ async def purchase_stars(client: "FragmentClient", username: str, amount: int, s
transaction = await execute_transaction_request(session, HEADERS, tx_data, fragment_hash)
tx_hash = await process_transaction(client, transaction)
return StarsResult(transaction_id=tx_hash, username=username, stars=amount)
return StarsResult(transaction_id=tx_hash, username=username, amount=amount)
except FragmentError:
raise
@@ -75,6 +75,23 @@ async def _init_request(
async def topup_ton(client: "FragmentClient", username: str, amount: int, show_sender: bool = True) -> AdsTopupResult:
"""Top up a Telegram Ads account balance with TON.
Args:
client: Authenticated :class:`FragmentClient` instance.
username: Ads account username (with or without ``@``).
amount: Amount in TON integer from ``1`` to ``1 000 000 000``.
show_sender: Show your name as the sender. Defaults to ``True``.
Returns:
:class:`AdsTopupResult` with ``transaction_id``, ``username``, and ``amount``.
Raises:
ConfigurationError: If ``amount`` is not an integer between 1 and 1 000 000 000.
UserNotFoundError: If the user is not found on Fragment.
FragmentAPIError: If the Fragment API returns an error.
UnexpectedError: For any other unexpected failure.
"""
if not isinstance(amount, int) or not (1 <= amount <= 1_000_000_000):
raise ConfigurationError(ConfigurationError.INVALID_TON_AMOUNT)
+10 -1
View File
@@ -13,7 +13,14 @@ from pyfragment.types.exceptions import (
VerificationError,
WalletError,
)
from pyfragment.types.results import AdsTopupResult, PremiumResult, StarsResult, WalletInfo
from pyfragment.types.results import (
AdsTopupResult,
PremiumGiveawayResult,
PremiumResult,
StarsGiveawayResult,
StarsResult,
WalletInfo,
)
__all__ = [
# client exceptions
@@ -33,7 +40,9 @@ __all__ = [
"WalletError",
# result types
"AdsTopupResult",
"PremiumGiveawayResult",
"PremiumResult",
"StarsGiveawayResult",
"StarsResult",
"WalletInfo",
]
+2
View File
@@ -21,7 +21,9 @@ REQUIRED_COOKIE_KEYS: tuple[str, ...] = ("stel_ssid", "stel_dt", "stel_token", "
# Fragment page URLs
STARS_PAGE: str = "https://fragment.com/stars/buy"
STARS_GIVEAWAY_PAGE: str = "https://fragment.com/stars/giveaway"
PREMIUM_PAGE: str = "https://fragment.com/premium/gift"
PREMIUM_GIVEAWAY_PAGE: str = "https://fragment.com/premium/giveaway"
TON_PAGE: str = "https://fragment.com/ads/topup"
# Tonkeeper device fingerprint — serialized once, reused in every tx_data payload.
+3
View File
@@ -19,6 +19,9 @@ class ConfigurationError(ClientError):
INVALID_USERNAME = (
"Invalid username '{username}'. Must be 532 characters: letters (AZ, az), digits (09), or underscores (_)."
)
INVALID_WINNERS_STARS = "winners must be an integer between 1 and 5."
INVALID_WINNERS_PREMIUM = "winners must be an integer between 1 and 24 000."
INVALID_STARS_PER_WINNER = "amount must be an integer between 500 and 1 000 000 stars."
class CookieError(ClientError):
+38 -10
View File
@@ -1,7 +1,6 @@
import time
from dataclasses import dataclass, field
from dataclasses import dataclass
__all__ = ["AdsTopupResult", "PremiumResult", "StarsResult", "WalletInfo"]
__all__ = ["AdsTopupResult", "PremiumGiveawayResult", "PremiumResult", "StarsGiveawayResult", "StarsResult", "WalletInfo"]
@dataclass
@@ -22,11 +21,10 @@ class PremiumResult:
transaction_id: str
username: str
months: int
timestamp: int = field(default_factory=lambda: int(time.time()))
amount: int
def __repr__(self) -> str:
return f"PremiumResult(username='{self.username}', months={self.months}, tx='{self.transaction_id}')"
return f"PremiumResult(username='{self.username}', amount={self.amount} months, tx='{self.transaction_id}')"
@dataclass
@@ -35,11 +33,10 @@ class StarsResult:
transaction_id: str
username: str
stars: int
timestamp: int = field(default_factory=lambda: int(time.time()))
amount: int
def __repr__(self) -> str:
return f"StarsResult(username='{self.username}', stars={self.stars}, tx='{self.transaction_id}')"
return f"StarsResult(username='{self.username}', amount={self.amount} stars, tx='{self.transaction_id}')"
@dataclass
@@ -49,7 +46,38 @@ class AdsTopupResult:
transaction_id: str
username: str
amount: int
timestamp: int = field(default_factory=lambda: int(time.time()))
def __repr__(self) -> str:
return f"AdsTopupResult(username='{self.username}', amount={self.amount} TON, tx='{self.transaction_id}')"
@dataclass
class StarsGiveawayResult:
"""Result of a successful Telegram Stars giveaway."""
transaction_id: str
channel: str
winners: int
amount: int
def __repr__(self) -> str:
return (
f"StarsGiveawayResult(channel='{self.channel}', winners={self.winners}, "
f"amount={self.amount} stars per winner, tx='{self.transaction_id}')"
)
@dataclass
class PremiumGiveawayResult:
"""Result of a successful Telegram Premium giveaway."""
transaction_id: str
channel: str
winners: int
amount: int
def __repr__(self) -> str:
return (
f"PremiumGiveawayResult(channel='{self.channel}', winners={self.winners}, "
f"amount={self.amount} months per winner, tx='{self.transaction_id}')"
)