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

Bump version to 2026.2.3 and refactor wallet utilities

Browse Source 
- Updated version in pyproject.toml to 2026.2.3.
- Refactored wallet utilities:
  - Moved `clean_decode` and `process_transaction` to `transaction.py`.
  - Created `balance.py` for balance-related functions.
  - Created `info.py` for wallet information retrieval.
  - Created `transfer.py` for sending TON and USDT transfers.
- Updated tests to reflect new module structure and imports.
- Added new API utility functions for handling Fragment API requests.
This commit is contained in:
bohd4nx
2026-05-12 02:01:15 +03:00
parent b67415bcc2
commit 363e0719ab
25 changed files with 502 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyfragment/client.py
+20 -16
View File
@@ -1,20 +1,24 @@
from __future__ import annotations
import json
from typing import Any, cast
from typing import Any, cast, get_args
import httpx
from pyfragment.methods.anonymous_number import get_login_code, terminate_sessions, toggle_login_codes
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.recharge_ads import recharge_ads
from pyfragment.methods.search_gifts import search_gifts
from pyfragment.methods.search_numbers import search_numbers
from pyfragment.methods.search_usernames import search_usernames
from pyfragment.methods.topup_ton import topup_ton
from pyfragment.methods import (
get_login_code,
giveaway_premium,
giveaway_stars,
purchase_premium,
purchase_stars,
recharge_ads,
search_gifts,
search_numbers,
search_usernames,
terminate_sessions,
toggle_login_codes,
topup_ton,
)
from pyfragment.types import (
AdsRechargeResult,
AdsTopupResult,
@@ -32,14 +36,14 @@ from pyfragment.types import (
WalletInfo,
)
from pyfragment.types.constants import (
BASE_HEADERS,
DEFAULT_TIMEOUT,
FRAGMENT_BASE_URL,
REQUIRED_COOKIE_KEYS,
SUPPORTED_WALLET_VERSIONS,
PaymentMethod,
WalletVersion,
)
from pyfragment.utils.http import fragment_request, get_fragment_hash, make_headers
from pyfragment.utils.api import fragment_request, get_fragment_hash
from pyfragment.utils.wallet import get_wallet_info
@@ -104,10 +108,10 @@ class FragmentClient:
raise CookieError(CookieError.MISSING_KEYS.format(keys=", ".join(missing_keys)))
version = wallet_version.strip().upper()
if version not in SUPPORTED_WALLET_VERSIONS:
if version not in get_args(WalletVersion):
raise ConfigurationError(
ConfigurationError.UNSUPPORTED_VERSION.format(
version=version, supported=", ".join(sorted(SUPPORTED_WALLET_VERSIONS))
version=version, supported=", ".join(sorted(get_args(WalletVersion)))
)
)
@@ -386,7 +390,7 @@ class FragmentClient:
page_url="https://fragment.com/premium/gift",
)
"""
headers = make_headers(page_url)
headers = {**BASE_HEADERS, "referer": page_url, "x-aj-referer": page_url}
async with httpx.AsyncClient(cookies=self.cookies, timeout=self.timeout) as session:
fragment_hash = await get_fragment_hash(self.cookies, headers, page_url, self.timeout)
return await fragment_request(session, fragment_hash, headers, {"method": method, **(data or {})})
pyfragment/methods/giveaway_premium.py
+4 -4
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, get_args
from pyfragment.types import (
ConfigurationError,
@@ -12,7 +12,7 @@ from pyfragment.types import (
UserNotFoundError,
VerificationError,
)
from pyfragment.types.constants import DEVICE, PREMIUM_GIVEAWAY_PAGE, SUPPORTED_PAYMENT_METHODS, PaymentMethod
from pyfragment.types.constants import DEVICE, PREMIUM_GIVEAWAY_PAGE, PaymentMethod
from pyfragment.utils import get_account_info, parse_required_payment_amount, process_transaction
if TYPE_CHECKING:
@@ -49,11 +49,11 @@ async def giveaway_premium(
raise ConfigurationError(ConfigurationError.INVALID_WINNERS_PREMIUM)
if months not in (3, 6, 12):
raise ConfigurationError(ConfigurationError.INVALID_MONTHS)
if payment_method not in SUPPORTED_PAYMENT_METHODS:
if payment_method not in get_args(PaymentMethod):
raise ConfigurationError(
ConfigurationError.INVALID_PAYMENT_METHOD.format(
method=payment_method,
supported=", ".join(sorted(SUPPORTED_PAYMENT_METHODS)),
supported=", ".join(sorted(get_args(PaymentMethod))),
)
)
pyfragment/methods/giveaway_stars.py
+4 -4
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, get_args
from pyfragment.types import (
ConfigurationError,
@@ -12,7 +12,7 @@ from pyfragment.types import (
UserNotFoundError,
VerificationError,
)
from pyfragment.types.constants import DEVICE, STARS_GIVEAWAY_PAGE, SUPPORTED_PAYMENT_METHODS, PaymentMethod
from pyfragment.types.constants import DEVICE, STARS_GIVEAWAY_PAGE, PaymentMethod
from pyfragment.utils import get_account_info, parse_required_payment_amount, process_transaction
if TYPE_CHECKING:
@@ -49,11 +49,11 @@ async def giveaway_stars(
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)
if payment_method not in SUPPORTED_PAYMENT_METHODS:
if payment_method not in get_args(PaymentMethod):
raise ConfigurationError(
ConfigurationError.INVALID_PAYMENT_METHOD.format(
method=payment_method,
supported=", ".join(sorted(SUPPORTED_PAYMENT_METHODS)),
supported=", ".join(sorted(get_args(PaymentMethod))),
)
)
pyfragment/methods/purchase_premium.py
+4 -4
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import json
import time
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, get_args
from pyfragment.types import (
ConfigurationError,
@@ -13,7 +13,7 @@ from pyfragment.types import (
UserNotFoundError,
VerificationError,
)
from pyfragment.types.constants import DEVICE, PREMIUM_PAGE, SUPPORTED_PAYMENT_METHODS, PaymentMethod
from pyfragment.types.constants import DEVICE, PREMIUM_PAGE, PaymentMethod
from pyfragment.utils import get_account_info, parse_required_payment_amount, process_transaction
if TYPE_CHECKING:
@@ -47,11 +47,11 @@ async def purchase_premium(
"""
if months not in (3, 6, 12):
raise ConfigurationError(ConfigurationError.INVALID_MONTHS)
if payment_method not in SUPPORTED_PAYMENT_METHODS:
if payment_method not in get_args(PaymentMethod):
raise ConfigurationError(
ConfigurationError.INVALID_PAYMENT_METHOD.format(
method=payment_method,
supported=", ".join(sorted(SUPPORTED_PAYMENT_METHODS)),
supported=", ".join(sorted(get_args(PaymentMethod))),
)
)
pyfragment/methods/purchase_stars.py
+4 -4
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import json
import time
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, get_args
from pyfragment.types import (
ConfigurationError,
@@ -13,7 +13,7 @@ from pyfragment.types import (
UserNotFoundError,
VerificationError,
)
from pyfragment.types.constants import DEVICE, STARS_PAGE, SUPPORTED_PAYMENT_METHODS, PaymentMethod
from pyfragment.types.constants import DEVICE, STARS_PAGE, PaymentMethod
from pyfragment.utils import get_account_info, parse_required_payment_amount, process_transaction
if TYPE_CHECKING:
@@ -43,11 +43,11 @@ async def purchase_stars(
"""
if not isinstance(amount, int) or not (50 <= amount <= 1_000_000):
raise ConfigurationError(ConfigurationError.INVALID_STARS_AMOUNT)
if payment_method not in SUPPORTED_PAYMENT_METHODS:
if payment_method not in get_args(PaymentMethod):
raise ConfigurationError(
ConfigurationError.INVALID_PAYMENT_METHOD.format(
method=payment_method,
supported=", ".join(sorted(SUPPORTED_PAYMENT_METHODS)),
supported=", ".join(sorted(get_args(PaymentMethod))),
)
)
pyfragment/types/__init__.py
+4
View File
@@ -27,6 +27,8 @@ from pyfragment.types.results import (
StarsGiveawayResult,
StarsResult,
TerminateSessionsResult,
TonTransferResult,
UsdtTransferResult,
UsernamesResult,
WalletInfo,
)
@@ -60,6 +62,8 @@ __all__ = [
"StarsGiveawayResult",
"StarsResult",
"TerminateSessionsResult",
"TonTransferResult",
"UsdtTransferResult",
"UsernamesResult",
"WalletInfo",
# literal types
pyfragment/types/constants.py
+1 -3
View File
@@ -1,17 +1,15 @@
from __future__ import annotations
import json
from typing import Any, Literal, get_args
from typing import Any, Literal
from tonutils.contracts.wallet import WalletV4R2, WalletV5R1
# Payment methods
PaymentMethod = Literal["ton", "usdt_ton"]
SUPPORTED_PAYMENT_METHODS: frozenset[str] = frozenset(get_args(PaymentMethod))
# Single source of truth for supported wallet versions
WalletVersion = Literal["V4R2", "V5R1"]
SUPPORTED_WALLET_VERSIONS: frozenset[str] = frozenset(get_args(WalletVersion))
# Wallet class map — used to resolve the correct contract from WALLET_VERSION
WALLET_CLASSES: dict[str, Any] = {"V4R2": WalletV4R2, "V5R1": WalletV5R1}
pyfragment/types/results.py
+26
View File
@@ -206,6 +206,30 @@ class GiftsResult:
return f"GiftsResult(items={len(self.items)}, next_offset={self.next_offset!r})"
@dataclass
class TonTransferResult:
"""Result of a direct TON transfer via :meth:`FragmentClient.send_ton`."""
transaction_id: str
destination: str
amount: int
def __repr__(self) -> str:
return f"TonTransferResult(destination='{self.destination}', amount={self.amount} TON, tx='{self.transaction_id}')"
@dataclass
class UsdtTransferResult:
"""Result of a direct USDT transfer via :meth:`FragmentClient.send_usdt`."""
transaction_id: str
destination: str
amount: int
def __repr__(self) -> str:
return f"UsdtTransferResult(destination='{self.destination}', amount={self.amount} USDT, tx='{self.transaction_id}')"
__all__ = [
"AdsRechargeResult",
"AdsTopupResult",
@@ -217,6 +241,8 @@ __all__ = [
"StarsGiveawayResult",
"StarsResult",
"TerminateSessionsResult",
"TonTransferResult",
"UsdtTransferResult",
"UsernamesResult",
"WalletInfo",
]
pyfragment/utils/__init__.py
+6 -7
View File
@@ -1,14 +1,12 @@
from pyfragment.utils.cookies import CookieResult, get_cookies_from_browser
from pyfragment.utils.decoder import clean_decode
from pyfragment.utils.html import parse_auction_rows, parse_gift_items, parse_login_code, parse_required_payment_amount
from pyfragment.utils.http import (
from pyfragment.utils.api import (
execute_transaction_request,
fragment_request,
get_fragment_hash,
make_headers,
parse_json_response,
)
from pyfragment.utils.wallet import get_account_info, process_transaction
from pyfragment.utils.cookies import CookieResult, get_cookies_from_browser
from pyfragment.utils.parser import parse_auction_rows, parse_gift_items, parse_login_code, parse_required_payment_amount
from pyfragment.utils.wallet import clean_decode, get_account_info, process_transaction, send_ton_transfer, send_usdt_transfer
__all__ = [
"clean_decode",
@@ -22,7 +20,8 @@ __all__ = [
"fragment_request",
"get_account_info",
"get_fragment_hash",
"make_headers",
"parse_json_response",
"process_transaction",
"send_ton_transfer",
"send_usdt_transfer",
]
pyfragment/utils/http.py → pyfragment/utils/api.py
+1 -5
View File
@@ -8,11 +8,7 @@ from typing import Any, cast
import httpx
from pyfragment.types import FragmentPageError, ParseError, VerificationError
from pyfragment.types.constants import BASE_HEADERS, DEFAULT_TIMEOUT, FRAGMENT_BASE_URL
def make_headers(page_url: str = FRAGMENT_BASE_URL) -> dict[str, str]:
return {**BASE_HEADERS, "referer": page_url, "x-aj-referer": page_url}
from pyfragment.types.constants import DEFAULT_TIMEOUT, FRAGMENT_BASE_URL
async def get_fragment_hash(
pyfragment/utils/decoder.py
-43
View File
@@ -1,43 +0,0 @@
from __future__ import annotations
import base64
from ton_core import Cell
from pyfragment.types import ParseError
def clean_decode(payload: str) -> str | Cell:
"""Decode a base64-encoded BOC payload to a plain-text comment string.
Fragment transaction payloads are BOC-serialised TVM cells. This function
base64-decodes the payload, parses the cell, skips the 32-bit op-code
prefix, and reads the snake-encoded UTF-8 comment.
Args:
payload: Base64url-encoded BOC string (padding is added automatically).
Returns:
Decoded comment string, ``""`` for an empty payload, or raw ``Cell``
when payload is a non-UTF8 binary body.
Raises:
ParseError: If the payload cannot be decoded or parsed.
"""
s = payload.strip()
if not s:
return ""
s += "=" * (-len(s) % 4)
try:
# Fragment may return URL-safe base64 ("-"/"_") in transaction payloads.
boc = base64.b64decode(s, altchars=b"-_", validate=True)
cell = Cell.one_from_boc(boc)
sl = cell.begin_parse()
sl.load_uint(32) # op code
try:
return sl.load_snake_string().strip()
except UnicodeDecodeError:
# Some Fragment payloads are binary TVM cells rather than text comments.
return cell
except Exception as exc:
raise ParseError(ParseError.UNPARSEABLE.format(context="payload decode", exc=exc)) from exc
pyfragment/utils/html.py → pyfragment/utils/parser.py
View File
pyfragment/utils/wallet.py
-241
View File
@@ -1,241 +0,0 @@
from __future__ import annotations
import asyncio
import base64
import random
import ssl
from typing import TYPE_CHECKING, Any
from ton_core import NetworkGlobalID
from tonutils.clients import TonapiClient
from tonutils.contracts.jetton import get_wallet_address_get_method, get_wallet_data_get_method
from tonutils.exceptions import ProviderResponseError
from pyfragment.types import TransactionError, WalletError, WalletInfo
from pyfragment.types.constants import (
MIN_TON_BALANCE,
MIN_USDT_BALANCE,
USDT_TON_MASTER_ADDRESS,
WALLET_CLASSES,
PaymentMethod,
)
from pyfragment.utils.decoder import clean_decode
if TYPE_CHECKING:
from pyfragment.client import FragmentClient
async def _get_usdt_balance(ton: Any, wallet_address: str) -> float:
"""Return wallet USDT balance via tonutils jetton get-methods."""
try:
jetton_wallet_address = await get_wallet_address_get_method(
client=ton,
address=USDT_TON_MASTER_ADDRESS,
owner_address=wallet_address,
)
wallet_data = await get_wallet_data_get_method(client=ton, address=jetton_wallet_address)
raw_balance = int(wallet_data[0]) if wallet_data else 0
return float(raw_balance) / 1_000_000.0
except ProviderResponseError as exc:
# No jetton wallet deployed yet -> effectively zero USDT balance.
if exc.code == 404:
return 0.0
raise WalletError(WalletError.USDT_BALANCE_CHECK_FAILED.format(exc=exc)) from exc
except Exception as exc:
raise WalletError(WalletError.USDT_BALANCE_CHECK_FAILED.format(exc=exc)) from exc
async def _check_ton_payment_balance(
balance_ton: float,
amount_ton: float,
required_payment_amount: float | None,
) -> None:
"""Validate balance requirements for TON payment method."""
tx_price_ton = amount_ton
if required_payment_amount is not None and required_payment_amount > 0:
tx_price_ton = max(tx_price_ton, required_payment_amount)
required_ton = max(tx_price_ton, MIN_TON_BALANCE)
if balance_ton < required_ton:
raise WalletError(
WalletError.LOW_TON_BALANCE.format(
balance=balance_ton,
required=required_ton,
)
)
async def _check_usdt_payment_balance(
balance_ton: float,
required_payment_amount: float | None,
ton: Any,
wallet_address: str,
) -> None:
"""Validate balance requirements for USDT payment method."""
# USDT payment still needs TON for network fees.
if balance_ton < MIN_TON_BALANCE:
raise WalletError(
WalletError.LOW_TON_BALANCE.format(
balance=balance_ton,
required=MIN_TON_BALANCE,
)
)
usdt_balance = await _get_usdt_balance(ton, wallet_address)
required_usdt = required_payment_amount if required_payment_amount is not None else MIN_USDT_BALANCE
if usdt_balance < required_usdt:
raise WalletError(WalletError.LOW_USDT_BALANCE.format(balance=usdt_balance, required=required_usdt))
async def process_transaction(
client: FragmentClient,
transaction_data: dict[str, Any],
payment_method: PaymentMethod = "ton",
required_payment_amount: float | None = None,
) -> str:
"""Sign and broadcast a Fragment transaction to the TON network.
Validates the payload structure, checks the wallet balance, decodes the
on-chain comment, and calls ``wallet.transfer``.
Args:
client: Authenticated :class:`FragmentClient` instance.
transaction_data: Raw transaction dict from ``execute_transaction_request``.
payment_method: Payment currency — ``"ton"`` or ``"usdt_ton"``.
required_payment_amount: Optional price from init*Request response.
Returns:
Normalised transaction hash string.
Raises:
TransactionError: If the payload is malformed or the broadcast fails.
WalletError: If the wallet balance is too low or cannot be fetched.
"""
if "transaction" not in transaction_data or not transaction_data["transaction"].get("messages"):
raise TransactionError(TransactionError.INVALID_PAYLOAD)
message = transaction_data["transaction"]["messages"][0]
amount_ton = int(message["amount"]) / 1_000_000_000
async with TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key) as ton:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
# Check balance covers selected payment flow requirements.
try:
await wallet.refresh()
balance_ton = wallet.balance / 1_000_000_000
wallet_address = wallet.address.to_str(False, False)
if payment_method == "ton":
await _check_ton_payment_balance(
balance_ton,
amount_ton,
required_payment_amount,
)
else:
await _check_usdt_payment_balance(
balance_ton,
required_payment_amount,
ton,
wallet_address,
)
except WalletError:
raise
except Exception as exc:
raise WalletError(WalletError.TON_BALANCE_CHECK_FAILED.format(exc=exc)) from exc
try:
raw_payload = str(message.get("payload", ""))
payload = clean_decode(raw_payload)
for attempt in range(3):
try:
result = await wallet.transfer(
destination=message["address"],
amount=int(message["amount"]), # nanotons, not TON
body=payload,
)
return str(result.normalized_hash)
except ProviderResponseError as exc:
if exc.code == 429 and attempt == 0:
await asyncio.sleep(1 + random.uniform(0, 0.5))
continue
if exc.code == 406 and "seqno" in str(exc).lower():
# Previous tx seqno not yet confirmed — wallet will re-fetch seqno on retry
if attempt < 2:
await asyncio.sleep(2 + random.uniform(0, 1))
continue
raise TransactionError(TransactionError.DUPLICATE_SEQNO) from exc
raise
except (WalletError, TransactionError):
raise
except Exception as exc:
cause: BaseException | None = exc
while cause is not None:
if isinstance(cause, ssl.SSLError):
raise TransactionError(TransactionError.BROADCAST_FAILED_SSL.format(exc=exc)) from exc
cause = cause.__cause__ or cause.__context__
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc=exc)) from exc
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc="transfer loop exited without result"))
async def get_account_info(client: FragmentClient) -> dict[str, Any]:
"""Fetch wallet address, public key, and state-init for the Fragment API.
Fragment requires account info to build each transaction payload. The
returned dict is JSON-serialised and passed as the ``account`` field in
``getBuy*Link`` / ``get*Link`` requests.
Args:
client: Authenticated :class:`FragmentClient` instance.
Returns:
Dict with ``address``, ``publicKey``, ``chain``, ``walletStateInit``.
Raises:
WalletError: If account info cannot be retrieved.
"""
async with TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key) as ton:
try:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, pub_key, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
boc = wallet.state_init.serialize().to_boc()
return {
"address": wallet.address.to_str(False, False),
"publicKey": pub_key.as_hex,
"chain": "-239",
"walletStateInit": base64.b64encode(boc).decode(),
}
except Exception as exc:
raise WalletError(WalletError.ACCOUNT_INFO_FAILED.format(exc=exc)) from exc
async def get_wallet_info(client: FragmentClient) -> WalletInfo:
"""Return the address, state and balance of the TON wallet.
Args:
client: Authenticated :class:`FragmentClient` instance.
Returns:
:class:`WalletInfo` with ``address``, ``state``, ``balance`` in TON,
and ``usdt_balance`` in USDT.
Raises:
WalletError: If the wallet state cannot be fetched.
"""
async with TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key) as ton:
try:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
await wallet.refresh()
wallet_address = wallet.address.to_str(False, False)
usdt_balance = await _get_usdt_balance(ton, wallet_address)
return WalletInfo(
address=wallet.address.to_str(is_user_friendly=True, is_bounceable=False),
state=wallet.state.value,
ton_balance=round(wallet.balance / 1_000_000_000, 4),
usdt_balance=round(usdt_balance, 4),
)
except Exception as exc:
raise WalletError(WalletError.WALLET_INFO_FAILED.format(exc=exc)) from exc
pyfragment/utils/wallet/__init__.py
+14
View File
@@ -0,0 +1,14 @@
from pyfragment.utils.wallet.balance import get_usdt_balance
from pyfragment.utils.wallet.info import get_account_info, get_wallet_info
from pyfragment.utils.wallet.transaction import clean_decode, process_transaction
from pyfragment.utils.wallet.transfer import send_ton_transfer, send_usdt_transfer
__all__ = [
"get_account_info",
"get_usdt_balance",
"get_wallet_info",
"process_transaction",
"clean_decode",
"send_ton_transfer",
"send_usdt_transfer",
]
pyfragment/utils/wallet/balance.py
+71
View File
@@ -0,0 +1,71 @@
from __future__ import annotations
from typing import Any
from tonutils.contracts.jetton import get_wallet_address_get_method, get_wallet_data_get_method
from tonutils.exceptions import ProviderResponseError
from pyfragment.types import WalletError
from pyfragment.types.constants import MIN_TON_BALANCE, MIN_USDT_BALANCE, USDT_TON_MASTER_ADDRESS
async def get_usdt_balance(ton: Any, wallet_address: str) -> float:
"""Return wallet USDT balance via tonutils jetton get-methods."""
try:
jetton_wallet_address = await get_wallet_address_get_method(
client=ton,
address=USDT_TON_MASTER_ADDRESS,
owner_address=wallet_address,
)
wallet_data = await get_wallet_data_get_method(client=ton, address=jetton_wallet_address)
raw_balance = int(wallet_data[0]) if wallet_data else 0
return float(raw_balance) / 1_000_000.0
except ProviderResponseError as exc:
# No jetton wallet deployed yet -> effectively zero USDT balance.
if exc.code == 404:
return 0.0
raise WalletError(WalletError.USDT_BALANCE_CHECK_FAILED.format(exc=exc)) from exc
except Exception as exc:
raise WalletError(WalletError.USDT_BALANCE_CHECK_FAILED.format(exc=exc)) from exc
async def check_ton_payment_balance(
balance_ton: float,
amount_ton: float,
required_payment_amount: float | None,
) -> None:
"""Validate balance requirements for TON payment method."""
tx_price_ton = amount_ton
if required_payment_amount is not None and required_payment_amount > 0:
tx_price_ton = max(tx_price_ton, required_payment_amount)
required_ton = max(tx_price_ton, MIN_TON_BALANCE)
if balance_ton < required_ton:
raise WalletError(
WalletError.LOW_TON_BALANCE.format(
balance=balance_ton,
required=required_ton,
)
)
async def check_usdt_payment_balance(
balance_ton: float,
required_payment_amount: float | None,
ton: Any,
wallet_address: str,
) -> None:
"""Validate balance requirements for USDT payment method."""
# USDT payment still needs TON for network fees.
if balance_ton < MIN_TON_BALANCE:
raise WalletError(
WalletError.LOW_TON_BALANCE.format(
balance=balance_ton,
required=MIN_TON_BALANCE,
)
)
usdt_balance = await get_usdt_balance(ton, wallet_address)
required_usdt = required_payment_amount if required_payment_amount is not None else MIN_USDT_BALANCE
if usdt_balance < required_usdt:
raise WalletError(WalletError.LOW_USDT_BALANCE.format(balance=usdt_balance, required=required_usdt))
pyfragment/utils/wallet/info.py
+75
View File
@@ -0,0 +1,75 @@
from __future__ import annotations
import base64
from typing import TYPE_CHECKING, Any
from ton_core import NetworkGlobalID
from tonutils.clients import TonapiClient
from pyfragment.types import WalletError, WalletInfo
from pyfragment.types.constants import WALLET_CLASSES
from pyfragment.utils.wallet.balance import get_usdt_balance
if TYPE_CHECKING:
from pyfragment.client import FragmentClient
async def get_account_info(client: FragmentClient) -> dict[str, Any]:
"""Fetch wallet address, public key, and state-init for the Fragment API.
Fragment requires account info to build each transaction payload. The
returned dict is JSON-serialised and passed as the ``account`` field in
``getBuy*Link`` / ``get*Link`` requests.
Args:
client: Authenticated :class:`FragmentClient` instance.
Returns:
Dict with ``address``, ``publicKey``, ``chain``, ``walletStateInit``.
Raises:
WalletError: If account info cannot be retrieved.
"""
async with TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key) as ton:
try:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, pub_key, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
boc = wallet.state_init.serialize().to_boc()
return {
"address": wallet.address.to_str(False, False),
"publicKey": pub_key.as_hex,
"chain": "-239",
"walletStateInit": base64.b64encode(boc).decode(),
}
except Exception as exc:
raise WalletError(WalletError.ACCOUNT_INFO_FAILED.format(exc=exc)) from exc
async def get_wallet_info(client: FragmentClient) -> WalletInfo:
"""Return the address, state and balance of the TON wallet.
Args:
client: Authenticated :class:`FragmentClient` instance.
Returns:
:class:`WalletInfo` with ``address``, ``state``, ``balance`` in TON,
and ``usdt_balance`` in USDT.
Raises:
WalletError: If the wallet state cannot be fetched.
"""
async with TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key) as ton:
try:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
await wallet.refresh()
wallet_address = wallet.address.to_str(False, False)
usdt_balance = await get_usdt_balance(ton, wallet_address)
return WalletInfo(
address=wallet.address.to_str(is_user_friendly=True, is_bounceable=False),
state=wallet.state.value,
ton_balance=round(wallet.balance / 1_000_000_000, 4),
usdt_balance=round(usdt_balance, 4),
)
except Exception as exc:
raise WalletError(WalletError.WALLET_INFO_FAILED.format(exc=exc)) from exc
pyfragment/utils/wallet/transaction.py
+128
View File
@@ -0,0 +1,128 @@
from __future__ import annotations
import asyncio
import base64
import random
import ssl
from typing import TYPE_CHECKING, Any
from ton_core import Cell, NetworkGlobalID
from tonutils.clients import TonapiClient
from tonutils.exceptions import ProviderResponseError
from pyfragment.types import ParseError, TransactionError, WalletError
from pyfragment.types.constants import WALLET_CLASSES, PaymentMethod
from pyfragment.utils.wallet.balance import check_ton_payment_balance, check_usdt_payment_balance
if TYPE_CHECKING:
from pyfragment.client import FragmentClient
def clean_decode(payload: str) -> str | Cell:
"""Decode a base64-encoded BOC payload to a plain-text comment string."""
s = payload.strip()
if not s:
return ""
s += "=" * (-len(s) % 4)
try:
boc = base64.b64decode(s, altchars=b"-_", validate=True)
cell = Cell.one_from_boc(boc)
sl = cell.begin_parse()
op = sl.load_uint(32)
if op != 0:
# Non-zero op code means this is a structured message (e.g. jetton transfer),
# not a plain text comment — return the full cell as-is.
return cell
try:
return sl.load_snake_string().strip()
except UnicodeDecodeError:
return cell
except Exception as exc:
raise ParseError(ParseError.UNPARSEABLE.format(context="payload decode", exc=exc)) from exc
async def process_transaction(
client: FragmentClient,
transaction_data: dict[str, Any],
payment_method: PaymentMethod = "ton",
required_payment_amount: float | None = None,
) -> str:
"""Sign and broadcast a Fragment transaction to the TON network.
Validates the payload structure, checks the wallet balance, decodes the
on-chain comment, and calls ``wallet.transfer``.
Args:
client: Authenticated :class:`FragmentClient` instance.
transaction_data: Raw transaction dict from ``execute_transaction_request``.
payment_method: Payment currency — ``"ton"`` or ``"usdt_ton"``.
required_payment_amount: Optional price from init*Request response.
Returns:
Normalised transaction hash string.
Raises:
TransactionError: If the payload is malformed or the broadcast fails.
WalletError: If the wallet balance is too low or cannot be fetched.
"""
if "transaction" not in transaction_data or not transaction_data["transaction"].get("messages"):
raise TransactionError(TransactionError.INVALID_PAYLOAD)
message = transaction_data["transaction"]["messages"][0]
amount_ton = int(message["amount"]) / 1_000_000_000
async with TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key) as ton:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
# Check balance covers selected payment flow requirements.
try:
await wallet.refresh()
balance_ton = wallet.balance / 1_000_000_000
if payment_method == "ton":
wallet.address.to_str(False, False)
await check_ton_payment_balance(balance_ton, amount_ton, required_payment_amount)
else:
# USDT is withdrawn from the Fragment-linked wallet (transaction["from"]),
# not from the signing seed wallet. Seed wallet only pays TON gas.
fragment_wallet_address = transaction_data["transaction"].get("from", "")
await check_usdt_payment_balance(balance_ton, required_payment_amount, ton, fragment_wallet_address)
except WalletError:
raise
except Exception as exc:
raise WalletError(WalletError.TON_BALANCE_CHECK_FAILED.format(exc=exc)) from exc
try:
raw_payload = str(message.get("payload", ""))
payload = clean_decode(raw_payload)
for attempt in range(3):
try:
result = await wallet.transfer(
destination=message["address"],
amount=int(message["amount"]), # nanotons, not TON
body=payload,
)
return str(result.normalized_hash)
except ProviderResponseError as exc:
if exc.code == 429 and attempt == 0:
await asyncio.sleep(1 + random.uniform(0, 0.5))
continue
if exc.code == 406 and "seqno" in str(exc).lower():
# Previous tx seqno not yet confirmed — wallet will re-fetch seqno on retry
if attempt < 2:
await asyncio.sleep(2 + random.uniform(0, 1))
continue
raise TransactionError(TransactionError.DUPLICATE_SEQNO) from exc
raise
except (WalletError, TransactionError):
raise
except Exception as exc:
cause: BaseException | None = exc
while cause is not None:
if isinstance(cause, ssl.SSLError):
raise TransactionError(TransactionError.BROADCAST_FAILED_SSL.format(exc=exc)) from exc
cause = cause.__cause__ or cause.__context__
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc=exc)) from exc
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc="transfer loop exited without result"))
pyfragment/utils/wallet/transfer.py
+103
View File
@@ -0,0 +1,103 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from ton_core import Address, NetworkGlobalID
from tonutils.clients import ToncenterClient
from tonutils.contracts import JettonTransferBuilder, TONTransferBuilder
from pyfragment.types import TransactionError, WalletError
from pyfragment.types.constants import USDT_TON_MASTER_ADDRESS, WALLET_CLASSES
from pyfragment.types.results import TonTransferResult, UsdtTransferResult
if TYPE_CHECKING:
from pyfragment.client import FragmentClient
async def send_ton_transfer(
client: FragmentClient,
destination: str,
amount: int,
body: str | None = None,
) -> TonTransferResult:
"""Send a direct TON transfer on-chain using ToncenterClient.
Args:
client: Authenticated :class:`FragmentClient` instance (seed and wallet_version used).
destination: Recipient TON address (any format, e.g. ``"UQ..."``).
amount: Amount in nanotons (1 TON = 1 000 000 000 nanotons).
body: Optional on-chain comment attached to the transfer.
Returns:
:class:`TonTransferResult` with ``transaction_id``, ``destination``, and ``amount``.
Raises:
TransactionError: If the transaction fails to broadcast.
"""
try:
async with ToncenterClient(network=NetworkGlobalID.MAINNET) as ton:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
result = await wallet.transfer_message(
TONTransferBuilder(
destination=Address(destination),
amount=amount,
body=body,
)
)
return TonTransferResult(
transaction_id=str(result.normalized_hash),
destination=destination,
amount=amount,
)
except (TransactionError, WalletError):
raise
except Exception as exc:
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc=exc)) from exc
async def send_usdt_transfer(
client: FragmentClient,
destination: str,
usdt_amount: int,
forward_payload: str | None = None,
ton_for_gas: int = 50_000_000,
) -> UsdtTransferResult:
"""Send a direct USDT (TON jetton) transfer on-chain using ToncenterClient.
Args:
client: Authenticated :class:`FragmentClient` instance (seed and wallet_version used).
destination: Recipient TON address (any format, e.g. ``"UQ..."``).
usdt_amount: Amount in USDT base units (6 decimals; 1 USDT = 1 000 000).
forward_payload: Optional comment forwarded to the recipient with the transfer notification.
ton_for_gas: TON attached for gas in nanotons. Defaults to ``50_000_000`` (0.05 TON).
Returns:
:class:`UsdtTransferResult` with ``transaction_id``, ``destination``, and ``amount``.
Raises:
TransactionError: If the transaction fails to broadcast.
"""
try:
async with ToncenterClient(network=NetworkGlobalID.MAINNET) as ton:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
result = await wallet.transfer_message(
JettonTransferBuilder(
destination=Address(destination),
jetton_amount=usdt_amount,
jetton_master_address=Address(USDT_TON_MASTER_ADDRESS),
forward_payload=forward_payload,
forward_amount=1,
amount=ton_for_gas,
)
)
return UsdtTransferResult(
transaction_id=str(result.normalized_hash),
destination=destination,
amount=usdt_amount,
)
except (TransactionError, WalletError):
raise
except Exception as exc:
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc=exc)) from exc