FOSS/pyfragment
1
0
Fork 0
mirror of https://github.com/vibe-existing/pyfragment.git synced 2026-07-25 06:54:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: restructure as installable PyPI package

Browse Source 
- Rename app/ → fragmentapi/ for proper package naming
- Add FragmentClient class with gift_premium, gift_stars, topup_ton methods
- Restructure core/ → types/ (exceptions, results, constants)
- Merge utils/hash.py into utils/client.py
- Replace _version.py with importlib.metadata
- Add input validation with min/max bounds
- Add unit tests: decode, client init/cookies
- Rename 002_test_hash → 003_test_hash
- Clean up pyproject.toml: pin deps, production classifiers
This commit is contained in:
bohd4nx
2026-03-15 21:03:03 +02:00
parent 4e017cb7a1
commit d2d046a2b9
38 changed files with 1051 additions and 959 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
-11
View File
@@ -1,11 +0,0 @@
# Fragment.com cookies - copy from browser after login (Header String format)
# Hash is now fetched dynamically
# TON wallet seed phrase - 12 or 24 words separated by spaces
SEED = "your_ton_wallet_seed_phrase_here"
# TON API key - get from https://tonconsole.com
API_KEY = "your_ton_api_key_here"
# TON wallet contract version: V4R2 or V5R1 (default: V5R1)
WALLET_VERSION = "V5R1"
.hypothesis/unicode_data/15.0.0/charmap.json.gz
BIN
View File
Binary file not shown.
app/core/__init__.py
-44
View File
@@ -1,44 +0,0 @@
from app.core.config import config
from app.core.constants import (
ADS_PAGE,
BASE_HEADERS,
DEVICE,
PREMIUM_PAGE,
STARS_PAGE,
WALLET_CLASSES,
WalletVersion,
)
from app.core.cookies import load_cookies
from app.core.exceptions import (
ConfigError,
CookiesError,
FragmentError,
HashFetchError,
RequestError,
TransactionError,
UserNotFoundError,
WalletError,
)
from app.core.logging import logger, setup_logging
__all__ = [
"ADS_PAGE",
"BASE_HEADERS",
"DEVICE",
"PREMIUM_PAGE",
"STARS_PAGE",
"WALLET_CLASSES",
"WalletVersion",
"ConfigError",
"CookiesError",
"FragmentError",
"HashFetchError",
"RequestError",
"TransactionError",
"UserNotFoundError",
"WalletError",
"config",
"load_cookies",
"logger",
"setup_logging",
]
app/core/config.py
-46
View File
@@ -1,46 +0,0 @@
import logging
import os
from pathlib import Path
from dotenv import load_dotenv
from app.core.constants import SUPPORTED_WALLET_VERSIONS, WalletVersion
from app.core.exceptions import ConfigError
logger = logging.getLogger(__name__)
class Config:
SEED: str
API_KEY: str
WALLET_VERSION: WalletVersion
def __init__(self) -> None:
# Load .env if present; env vars already in the process take precedence
env_path = Path(__file__).resolve().parents[2] / ".env"
if env_path.exists():
load_dotenv(env_path)
missing = [k for k in ("SEED", "API_KEY") if not os.getenv(k, "").strip()]
if missing:
raise ConfigError(
f"Missing required environment variables: {', '.join(missing)}. "
"Copy .env.example to .env and fill in SEED and API_KEY."
)
self.SEED = os.getenv("SEED", "").strip()
self.API_KEY = os.getenv("API_KEY", "").strip()
version = os.getenv("WALLET_VERSION", "V5R1").strip().upper()
if version not in SUPPORTED_WALLET_VERSIONS:
raise ConfigError(
f"Unsupported WALLET_VERSION '{version}'. " f"Must be one of: {', '.join(sorted(SUPPORTED_WALLET_VERSIONS))}."
)
self.WALLET_VERSION: WalletVersion = version # type: ignore[assignment]
config: Config | None = None
try:
config = Config()
except ConfigError as e:
logger.warning("Configuration not loaded: %s", e)
app/core/cookies.py
-32
View File
@@ -1,32 +0,0 @@
import json
import logging
from pathlib import Path
from typing import Any
from app.core.exceptions import CookiesError
logger = logging.getLogger(__name__)
_REQUIRED_KEYS = ("stel_ssid", "stel_dt", "stel_token", "stel_ton_token")
def load_cookies() -> dict[str, Any]:
cookies_path = Path(__file__).resolve().parents[2] / "cookies.json"
if not cookies_path.exists():
raise CookiesError("cookies.json not found. Create it in the project root and paste your Fragment cookies.")
try:
with cookies_path.open("r", encoding="utf-8") as f:
cookies = json.load(f)
except Exception as exc:
raise CookiesError(f"Failed to read cookies.json: {exc}") from exc
missing = [k for k in _REQUIRED_KEYS if not str(cookies.get(k, "")).strip()]
if missing:
raise CookiesError(
f"cookies.json is missing or has empty values for: {', '.join(missing)}. "
"Open Fragment.com in your browser, copy fresh cookies, and update the file."
)
return cookies
app/core/exceptions.py
-42
View File
@@ -1,42 +0,0 @@
__all__ = [
"ConfigError",
"CookiesError",
"FragmentError",
"HashFetchError",
"RequestError",
"TransactionError",
"UserNotFoundError",
"WalletError",
]
class FragmentError(Exception):
"""Base exception for all Fragment API errors."""
class ConfigError(FragmentError):
"""Raised when .env is missing or required keys are absent."""
class CookiesError(FragmentError):
"""Raised when cookies.json is missing, unreadable, or has empty required fields."""
class HashFetchError(FragmentError):
"""Raised when the Fragment API hash cannot be fetched from the page."""
class UserNotFoundError(FragmentError):
"""Raised when the target Telegram user is not found on Fragment."""
class WalletError(FragmentError):
"""Raised for TON wallet issues (connection, balance, account info)."""
class TransactionError(FragmentError):
"""Raised when a TON transaction fails to build or broadcast."""
class RequestError(FragmentError):
"""Raised when a Fragment API response cannot be parsed."""
app/core/logging.py
-20
View File
@@ -1,20 +0,0 @@
import logging
def setup_logging() -> None:
formatter = logging.Formatter(fmt="[%(asctime)s] - %(levelname)s: %(message)s", datefmt="%d.%m.%y %H:%M:%S")
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
file_handler = logging.FileHandler("FragmentAPI.log", mode="w", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logging.basicConfig(level=logging.DEBUG, handlers=[console_handler, file_handler], force=True)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
app/methods/__init__.py
-5
View File
@@ -1,5 +0,0 @@
from app.methods.premium import buy_premium
from app.methods.stars import buy_stars
from app.methods.ton import topup_ton
__all__ = ["buy_premium", "buy_stars", "topup_ton"]
app/methods/premium.py
-151
View File
@@ -1,151 +0,0 @@
import json
import logging
import time
import httpx
from app.core import (
BASE_HEADERS,
DEVICE,
PREMIUM_PAGE,
FragmentError,
UserNotFoundError,
load_cookies,
)
from app.utils import (
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
logger = logging.getLogger(__name__)
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": PREMIUM_PAGE,
"x-aj-referer": PREMIUM_PAGE,
}
async def search_premium_recipient(
client: httpx.AsyncClient,
fragment_hash: str,
username: str,
months: int,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"query": username,
"months": months,
"method": "searchPremiumGiftRecipient",
},
)
result = parse_json_response(resp, "searchPremiumGiftRecipient")
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(
f"Telegram user '{username}' was not found on Fragment. "
"Make sure the username is correct and the account exists."
)
return recipient
async def init_gift_premium(
client: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
months: int,
) -> str:
await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"mode": "new",
"lv": "false",
"dh": str(int(time.time())),
"method": "updatePremiumState",
},
)
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"recipient": recipient,
"months": months,
"method": "initGiftPremiumRequest",
},
)
result = parse_json_response(resp, "initGiftPremiumRequest")
req_id = result.get("req_id")
if not req_id:
raise FragmentError(
"Fragment did not return a request ID for this Premium purchase. "
"The session may have expired — refresh your cookies."
)
return req_id
async def buy_premium(username: str, months: int, show_sender: bool = True) -> dict:
if months not in (3, 6, 12):
return {
"success": False,
"error": "Invalid duration. Choose 3, 6, or 12 months.",
}
try:
logger.info("Loading session cookies")
cookies = load_cookies()
logger.info("Fetching Fragment session hash")
fragment_hash = await get_fragment_hash(cookies, HEADERS, PREMIUM_PAGE)
# logger.info("Retrieving TON wallet info")
account = await get_account_info()
async with httpx.AsyncClient(cookies=cookies) as client:
logger.info("Searching recipient: %s", username)
recipient = await search_premium_recipient(client, fragment_hash, username, months)
logger.info("Initializing Premium gift request: %s months to %s", months, username)
req_id = await init_gift_premium(client, fragment_hash, recipient, months)
# logger.info("Requesting transaction payload (req_id=%s)", req_id)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": int(show_sender),
"method": "getGiftPremiumLink",
}
transaction = await execute_transaction_request(client, HEADERS, account, tx_data, fragment_hash)
logger.info("Broadcasting transaction to TON blockchain")
tx_hash = await process_transaction(transaction)
logger.info(
"Premium purchase successful: %s months -> %s | tx: %s",
months,
username,
tx_hash,
)
return {
"success": True,
"data": {
"transaction_id": tx_hash,
"username": username,
"months": months,
"timestamp": int(time.time()),
},
}
except FragmentError as exc:
logger.error("Premium purchase failed — %s", exc)
return {"success": False, "error": str(exc)}
except Exception as exc:
logger.exception("Unexpected error during Premium purchase")
return {"success": False, "error": f"Unexpected error: {exc}"}
app/methods/stars.py
-133
View File
@@ -1,133 +0,0 @@
import json
import logging
import time
import httpx
from app.core import (
BASE_HEADERS,
DEVICE,
STARS_PAGE,
FragmentError,
UserNotFoundError,
load_cookies,
)
from app.utils import (
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
logger = logging.getLogger(__name__)
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": STARS_PAGE,
"x-aj-referer": STARS_PAGE,
}
async def search_stars_recipient(
client: httpx.AsyncClient,
fragment_hash: str,
username: str,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"query": username, "quantity": "", "method": "searchStarsRecipient"},
)
result = parse_json_response(resp, "searchStarsRecipient")
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(
f"Telegram user '{username}' was not found on Fragment. "
"Make sure the username is correct and the account exists."
)
return recipient
async def init_buy_stars(
client: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
amount: int,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"recipient": recipient,
"quantity": amount,
"method": "initBuyStarsRequest",
},
)
result = parse_json_response(resp, "initBuyStarsRequest")
req_id = result.get("req_id")
if not req_id:
raise FragmentError(
"Fragment did not return a request ID for this Stars purchase. "
"The session may have expired — refresh your cookies."
)
return req_id
async def buy_stars(username: str, amount: int, show_sender: bool = True) -> dict:
if not isinstance(amount, int) or amount < 50:
return {"success": False, "error": "Amount must be an integer >= 50 stars."}
try:
logger.info("Loading session cookies")
cookies = load_cookies()
logger.info("Fetching Fragment session hash")
fragment_hash = await get_fragment_hash(cookies, HEADERS, STARS_PAGE)
# logger.info("Retrieving TON wallet info")
account = await get_account_info()
async with httpx.AsyncClient(cookies=cookies) as client:
logger.info("Searching recipient: %s", username)
recipient = await search_stars_recipient(client, fragment_hash, username)
logger.info("Initializing Stars purchase request: %s stars to %s", amount, username)
req_id = await init_buy_stars(client, fragment_hash, recipient, amount)
# logger.info("Requesting transaction payload (req_id=%s)", req_id)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": int(show_sender),
"method": "getBuyStarsLink",
}
transaction = await execute_transaction_request(client, HEADERS, account, tx_data, fragment_hash)
logger.info("Broadcasting transaction to TON blockchain")
tx_hash = await process_transaction(transaction)
logger.info(
"Stars purchase successful: %s stars -> %s | tx: %s",
amount,
username,
tx_hash,
)
return {
"success": True,
"data": {
"transaction_id": tx_hash,
"username": username,
"amount": amount,
"timestamp": int(time.time()),
},
}
except FragmentError as exc:
logger.error("Stars purchase failed — %s", exc)
return {"success": False, "error": str(exc)}
except Exception as exc:
logger.exception("Unexpected error during Stars purchase")
return {"success": False, "error": f"Unexpected error: {exc}"}
app/methods/ton.py
-132
View File
@@ -1,132 +0,0 @@
import json
import logging
import time
import httpx
from app.core import (
ADS_PAGE,
BASE_HEADERS,
DEVICE,
FragmentError,
UserNotFoundError,
load_cookies,
)
from app.utils import (
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
logger = logging.getLogger(__name__)
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": ADS_PAGE,
"x-aj-referer": ADS_PAGE,
}
async def search_ads_recipient(
client: httpx.AsyncClient,
fragment_hash: str,
username: str,
) -> str:
await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"mode": "new", "method": "updateAdsTopupState"},
)
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"query": username, "method": "searchAdsTopupRecipient"},
)
result = parse_json_response(resp, "searchAdsTopupRecipient")
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(
f"Telegram user '{username}' was not found on Fragment. "
"Make sure the username is correct and the account exists."
)
return recipient
async def init_ads_topup(
client: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
amount: int,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"recipient": recipient,
"amount": amount,
"method": "initAdsTopupRequest",
},
)
result = parse_json_response(resp, "initAdsTopupRequest")
req_id = result.get("req_id")
if not req_id:
raise FragmentError(
"Fragment did not return a request ID for this TON topup. " "The session may have expired — refresh your cookies."
)
return req_id
async def topup_ton(username: str, amount: int, show_sender: bool = True) -> dict:
if not isinstance(amount, int) or amount < 1:
return {"success": False, "error": "Amount must be an integer >= 1 TON."}
try:
logger.info("Loading session cookies")
cookies = load_cookies()
logger.info("Fetching Fragment session hash")
fragment_hash = await get_fragment_hash(cookies, HEADERS, ADS_PAGE)
# logger.info("Retrieving TON wallet info")
account = await get_account_info()
async with httpx.AsyncClient(cookies=cookies) as client:
logger.info("Searching recipient: %s", username)
recipient = await search_ads_recipient(client, fragment_hash, username)
logger.info("Initializing topup request: %s TON to %s", amount, username)
req_id = await init_ads_topup(client, fragment_hash, recipient, amount)
# logger.info("Requesting transaction payload (req_id=%s)", req_id)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": int(show_sender),
"method": "getAdsTopupLink",
}
transaction = await execute_transaction_request(client, HEADERS, account, tx_data, fragment_hash)
logger.info("Broadcasting transaction to TON blockchain")
tx_hash = await process_transaction(transaction)
logger.info("TON topup successful: %s TON -> %s | tx: %s", amount, username, tx_hash)
return {
"success": True,
"data": {
"transaction_id": tx_hash,
"username": username,
"amount": amount,
"timestamp": int(time.time()),
},
}
except FragmentError as exc:
logger.error("TON topup failed — %s", exc)
return {"success": False, "error": str(exc)}
except Exception as exc:
logger.exception("Unexpected error during TON topup")
return {"success": False, "error": f"Unexpected error: {exc}"}
app/utils/__init__.py
-14
View File
@@ -1,14 +0,0 @@
from app.utils.client import execute_transaction_request, parse_json_response
from app.utils.decoder import clean_decode
from app.utils.hash import get_fragment_hash
from app.utils.wallet import get_account_info, link_wallet, process_transaction
__all__ = [
"clean_decode",
"execute_transaction_request",
"get_account_info",
"get_fragment_hash",
"link_wallet",
"parse_json_response",
"process_transaction",
]
app/utils/client.py
-39
View File
@@ -1,39 +0,0 @@
import logging
from typing import Any
import httpx
from app.core import RequestError, WalletError
from app.utils.wallet import link_wallet
logger = logging.getLogger(__name__)
def parse_json_response(response: httpx.Response, context: str) -> dict[str, Any]:
try:
return response.json()
except Exception as exc:
raise RequestError(f"Fragment API returned an unparseable response for '{context}': {exc}") from exc
async def execute_transaction_request(
client: httpx.AsyncClient,
headers: dict,
account: dict[str, Any],
tx_data: dict[str, Any],
fragment_hash: str,
) -> dict[str, Any]:
url = f"https://fragment.com/api?hash={fragment_hash}"
resp = await client.post(url, headers=headers, data=tx_data)
transaction = parse_json_response(resp, tx_data.get("method", "transaction"))
if transaction.get("need_verify"):
if not await link_wallet(client, headers, account, fragment_hash):
raise WalletError(
"Failed to link your TON wallet to Fragment. " "Make sure the wallet matching your cookies is used."
)
resp = await client.post(url, headers=headers, data=tx_data)
transaction = parse_json_response(resp, tx_data.get("method", "transaction"))
return transaction
app/utils/decoder.py
-36
View File
@@ -1,36 +0,0 @@
import base64
import logging
from pytoniq_core import Cell
logger = logging.getLogger(__name__)
# OLD decoder (manual base64 + regex, kept for reference):
#
# import re, string
# def clean_decode(payload: str) -> str:
# s = re.sub(r'[^A-Za-z0-9+/=]', '', payload.strip())
# s += '=' * (-len(s) % 4)
# text = base64.b64decode(s).decode('utf-8', errors='ignore')
# text = ''.join(c for c in text if c in string.printable or c.isspace())
# match = re.search(r'([0-9]*\s*Telegram .*?Ref#[A-Za-z0-9]+)', text, re.S)
# return match.group(1).strip() if match else text.strip()
def clean_decode(payload: str) -> str:
# Pad and decode base64 → BOC bytes
s = payload.strip()
if not s:
return ""
s += "=" * (-len(s) % 4)
boc = base64.b64decode(s)
# Parse BOC cell and read snake-encoded text (skipping 32-bit op prefix)
cell = Cell.one_from_boc(boc)
sl = cell.begin_parse()
sl.load_uint(32) # op code — always 0 for text comment
result = sl.load_snake_string().strip()
logger.debug("Payload: %s -> %s", payload, result.replace("\n", " "))
return result
app/utils/hash.py
-57
View File
@@ -1,57 +0,0 @@
import logging
import re
from typing import Any
import httpx
from app.core import HashFetchError
logger = logging.getLogger(__name__)
async def get_fragment_hash(
cookies: dict[str, Any],
headers: dict[str, str],
page_url: str,
) -> str:
# Must look like a real browser navigation — not an XHR — otherwise Fragment
# returns JSON (no hash in it) instead of full HTML.
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": "https://fragment.com/",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"upgrade-insecure-requests": "1",
}
)
async with httpx.AsyncClient(cookies=cookies) as client:
response = await client.get(page_url, headers=page_headers)
if response.status_code != 200:
raise HashFetchError(
f"Fragment returned HTTP {response.status_code} for {page_url}. "
"Check that your cookies are valid and not expired."
)
match = re.search(r"(?:https://fragment\.com)?/api\?hash=([a-f0-9]+)", response.text)
if not match:
raise HashFetchError(
f"Fragment hash not found in the page source of {page_url}. "
"The page structure may have changed or you are not logged in."
)
return match.group(1)
app/utils/wallet.py
-107
View File
@@ -1,107 +0,0 @@
import base64
import json
import logging
from typing import Any
import httpx
from tonutils.clients import TonapiClient
from tonutils.types import NetworkGlobalID
from app.core import DEVICE, WALLET_CLASSES, TransactionError, WalletError, config
from app.utils.decoder import clean_decode
logger = logging.getLogger(__name__)
def initialize_ton_client() -> TonapiClient:
return TonapiClient(network=NetworkGlobalID.MAINNET, api_key=config.API_KEY)
async def process_transaction(transaction_data: dict) -> str:
logger.debug("transaction_data: %s", transaction_data)
if "transaction" not in transaction_data or "messages" not in transaction_data["transaction"]:
raise TransactionError(
"Fragment returned an invalid transaction payload. "
"The API response is missing expected 'transaction.messages' data."
)
# TODO: Investigate 406 'inbound external message rejected before smart-contract execution'.
# This happens when the previous transaction's seqno hasn't been confirmed on-chain yet,
# causing the wallet contract to reject the new message.
async with initialize_ton_client() as client:
wallet_cls = WALLET_CLASSES[config.WALLET_VERSION]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=client, mnemonic=config.SEED)
# Check balance before broadcasting
try:
await wallet.refresh()
balance_ton = wallet.balance / 1_000_000_000
if balance_ton < 0.056:
raise WalletError(f"TON wallet balance is too low: {balance_ton:.2f} TON. " "Minimum required is 0.056 TON.")
except WalletError:
raise
except Exception as exc:
raise WalletError(f"Wallet balance check failed: {exc}") from exc
try:
message = transaction_data["transaction"]["messages"][0]
payload = clean_decode(message["payload"])
result = await wallet.transfer(
destination=message["address"],
amount=int(message["amount"]), # nanotons, not TON
body=payload,
)
tx_hash = result.normalized_hash
return tx_hash
except (WalletError, TransactionError):
raise
except Exception as exc:
raise TransactionError(f"Transaction broadcast failed: {exc}") from exc
async def get_account_info() -> dict[str, Any]:
async with initialize_ton_client() as client:
try:
wallet_cls = WALLET_CLASSES[config.WALLET_VERSION]
wallet, pub_key, _, _ = wallet_cls.from_mnemonic(client=client, mnemonic=config.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(f"Failed to retrieve wallet account info: {exc}") from exc
async def link_wallet(
client: httpx.AsyncClient,
headers: dict,
account: dict[str, Any],
fragment_hash: str,
) -> bool:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=headers,
data={
"account": json.dumps(account),
"device": DEVICE,
"method": "linkWallet",
},
)
result = resp.json()
if result.get("ok"):
return True
if "transaction" in result:
try:
await process_transaction(result)
return True
except (TransactionError, WalletError):
return False
return False
cookies.example.json
-6
View File
@@ -1,6 +0,0 @@
{
"stel_ssid": "",
"stel_dt": "",
"stel_token": "",
"stel_ton_token": ""
}
fragmentapi/__init__.py
+44
View File
@@ -0,0 +1,44 @@
# Copyright (c) 2025 bohd4nx
#
# This source code is licensed under the MIT License found in the
# LICENSE file in the root directory of this source tree.
from fragmentapi.client import FragmentClient
from fragmentapi.types import (
AdsTopupResult,
ClientError,
ConfigError,
CookiesError,
FragmentAPIError,
FragmentError,
HashFetchError,
OperationError,
PremiumResult,
RequestError,
StarsResult,
TransactionError,
UnexpectedError,
UserNotFoundError,
VerificationError,
WalletError,
)
__all__ = [
"FragmentClient",
"AdsTopupResult",
"PremiumResult",
"StarsResult",
"ClientError",
"ConfigError",
"CookiesError",
"FragmentAPIError",
"FragmentError",
"HashFetchError",
"OperationError",
"RequestError",
"TransactionError",
"UnexpectedError",
"UserNotFoundError",
"VerificationError",
"WalletError",
]
fragmentapi/client.py
+112
View File
@@ -0,0 +1,112 @@
import json
from fragmentapi.methods.premium import gift_premium
from fragmentapi.methods.stars import gift_stars
from fragmentapi.methods.ton import topup_ton
from fragmentapi.types import (
REQUIRED_COOKIE_KEYS,
SUPPORTED_WALLET_VERSIONS,
AdsTopupResult,
ConfigError,
CookiesError,
PremiumResult,
StarsResult,
WalletVersion,
)
class FragmentClient:
"""
Client for the Fragment.com API.
Args:
seed: 24-word mnemonic phrase for the TON wallet.
api_key: Tonapi API key — get one at https://tonconsole.com.
cookies: Fragment session cookies as a dict or JSON string.
wallet_version: Wallet contract version — ``"V4R2"`` or ``"V5R1"`` (default).
Raises:
ConfigError: If ``seed``, ``api_key``, or ``wallet_version`` are missing or invalid.
CookiesError: If ``cookies`` cannot be parsed or are missing required keys.
Example::
client = FragmentClient(
seed="word1 word2 ...",
api_key="AAABBB...",
cookies={"stel_ssid": "...", "stel_dt": "...", ...},
)
result = await client.gift_premium("@username", months=6)
print(result.transaction_id)
"""
def __init__(
self,
seed: str,
api_key: str,
cookies: dict | str,
wallet_version: str = "V5R1",
) -> None:
missing = [name for name, val in (("seed", seed), ("api_key", api_key)) if not val or not str(val).strip()]
if missing:
raise ConfigError(ConfigError.MISSING_VARS.format(keys=", ".join(missing)))
if isinstance(cookies, str):
try:
cookies = json.loads(cookies)
except Exception as exc:
raise CookiesError(CookiesError.READ_FAILED.format(exc=exc)) from exc
missing_keys = [k for k in REQUIRED_COOKIE_KEYS if not str(cookies.get(k, "")).strip()]
if missing_keys:
raise CookiesError(CookiesError.MISSING_KEYS.format(keys=", ".join(missing_keys)))
version = wallet_version.strip().upper()
if version not in SUPPORTED_WALLET_VERSIONS:
raise ConfigError(
ConfigError.UNSUPPORTED_VERSION.format(version=version, supported=", ".join(sorted(SUPPORTED_WALLET_VERSIONS)))
)
self.seed: str = seed.strip()
self.api_key: str = api_key.strip()
self.cookies: dict = cookies
self.wallet_version: WalletVersion = version # type: ignore[assignment]
async def gift_premium(self, username: str, months: int, show_sender: bool = True) -> PremiumResult:
"""Gift Telegram Premium to a user.
Args:
username: Recipient's Telegram username (with or without ``@``).
months: Duration — ``3``, ``6``, or ``12``.
show_sender: Show your name as the gift sender. Defaults to ``True``.
Returns:
:class:`PremiumResult` with ``transaction_id``, ``username``, ``months``, ``timestamp``.
"""
return await gift_premium(self, username, months, show_sender)
async def gift_stars(self, username: str, amount: int, show_sender: bool = True) -> StarsResult:
"""Gift Telegram Stars to a user.
Args:
username: Recipient's Telegram username (with or without ``@``).
amount: Number of stars — 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``, ``stars``, ``timestamp``.
"""
return await gift_stars(self, username, amount, show_sender)
async def topup_ton(self, username: str, amount: int, show_sender: bool = True) -> AdsTopupResult:
"""Top up Telegram Ads balance with TON.
Args:
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``, ``amount``, ``timestamp``.
"""
return await topup_ton(self, username, amount, show_sender)
fragmentapi/methods/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from fragmentapi.methods.premium import gift_premium
from fragmentapi.methods.stars import gift_stars
from fragmentapi.methods.ton import topup_ton
__all__ = ["gift_premium", "gift_stars", "topup_ton"]
fragmentapi/methods/premium.py
+119
View File
@@ -0,0 +1,119 @@
import json
import time
from typing import TYPE_CHECKING
import httpx
from fragmentapi.types import (
BASE_HEADERS,
DEVICE,
PREMIUM_PAGE,
ConfigError,
FragmentAPIError,
FragmentError,
PremiumResult,
UnexpectedError,
UserNotFoundError,
)
from fragmentapi.utils import (
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
if TYPE_CHECKING:
from fragmentapi.client import FragmentClient
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": PREMIUM_PAGE,
"x-aj-referer": PREMIUM_PAGE,
}
async def _search_recipient(
session: httpx.AsyncClient,
fragment_hash: str,
username: str,
months: int,
) -> str:
resp = await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"query": username,
"months": months,
"method": "searchPremiumGiftRecipient",
},
)
result = parse_json_response(resp, "searchPremiumGiftRecipient")
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(UserNotFoundError.NOT_FOUND.format(username=username))
return recipient
async def _init_request(
session: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
months: int,
) -> str:
await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"mode": "new",
"lv": "false",
"dh": str(int(time.time())),
"method": "updatePremiumState",
},
)
resp = await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"recipient": recipient,
"months": months,
"method": "initGiftPremiumRequest",
},
)
result = parse_json_response(resp, "initGiftPremiumRequest")
req_id = result.get("req_id")
if not req_id:
raise FragmentAPIError(FragmentAPIError.NO_REQUEST_ID.format(context="Premium purchase"))
return req_id
async def gift_premium(client: "FragmentClient", username: str, months: int, show_sender: bool = True) -> PremiumResult:
if months not in (3, 6, 12):
raise ConfigError(ConfigError.INVALID_MONTHS)
try:
fragment_hash = await get_fragment_hash(client.cookies, HEADERS, PREMIUM_PAGE)
account = await get_account_info(client)
async with httpx.AsyncClient(cookies=client.cookies) as session:
recipient = await _search_recipient(session, fragment_hash, username, months)
req_id = await _init_request(session, fragment_hash, recipient, months)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": int(show_sender),
"method": "getGiftPremiumLink",
}
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)
except FragmentError:
raise
except Exception as exc:
raise UnexpectedError(UnexpectedError.UNEXPECTED.format(exc=exc)) from exc
fragmentapi/methods/stars.py
+103
View File
@@ -0,0 +1,103 @@
import json
from typing import TYPE_CHECKING
import httpx
from fragmentapi.types import (
BASE_HEADERS,
DEVICE,
STARS_PAGE,
ConfigError,
FragmentAPIError,
FragmentError,
StarsResult,
UnexpectedError,
UserNotFoundError,
)
from fragmentapi.utils import (
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
if TYPE_CHECKING:
from fragmentapi.client import FragmentClient
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": STARS_PAGE,
"x-aj-referer": STARS_PAGE,
}
async def _search_recipient(
session: httpx.AsyncClient,
fragment_hash: str,
username: str,
) -> str:
resp = await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"query": username, "quantity": "", "method": "searchStarsRecipient"},
)
result = parse_json_response(resp, "searchStarsRecipient")
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(UserNotFoundError.NOT_FOUND.format(username=username))
return recipient
async def _init_request(
session: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
amount: int,
) -> str:
resp = await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"recipient": recipient,
"quantity": amount,
"method": "initBuyStarsRequest",
},
)
result = parse_json_response(resp, "initBuyStarsRequest")
req_id = result.get("req_id")
if not req_id:
raise FragmentAPIError(FragmentAPIError.NO_REQUEST_ID.format(context="Stars purchase"))
return req_id
async def gift_stars(client: "FragmentClient", username: str, amount: int, show_sender: bool = True) -> StarsResult:
if not isinstance(amount, int) or not (50 <= amount <= 1_000_000):
raise ConfigError(ConfigError.INVALID_STARS_AMOUNT)
try:
fragment_hash = await get_fragment_hash(client.cookies, HEADERS, STARS_PAGE)
account = await get_account_info(client)
async with httpx.AsyncClient(cookies=client.cookies) as session:
recipient = await _search_recipient(session, fragment_hash, username)
req_id = await _init_request(session, fragment_hash, recipient, amount)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": int(show_sender),
"method": "getBuyStarsLink",
}
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)
except FragmentError:
raise
except Exception as exc:
raise UnexpectedError(UnexpectedError.UNEXPECTED.format(exc=exc)) from exc
fragmentapi/methods/ton.py
+108
View File
@@ -0,0 +1,108 @@
import json
from typing import TYPE_CHECKING
import httpx
from fragmentapi.types import (
BASE_HEADERS,
DEVICE,
TON_PAGE,
AdsTopupResult,
ConfigError,
FragmentAPIError,
FragmentError,
UnexpectedError,
UserNotFoundError,
)
from fragmentapi.utils import (
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
if TYPE_CHECKING:
from fragmentapi.client import FragmentClient
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": TON_PAGE,
"x-aj-referer": TON_PAGE,
}
async def _search_recipient(
session: httpx.AsyncClient,
fragment_hash: str,
username: str,
) -> str:
await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"mode": "new", "method": "updateAdsTopupState"},
)
resp = await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"query": username, "method": "searchAdsTopupRecipient"},
)
result = parse_json_response(resp, "searchAdsTopupRecipient")
recipient = result.get("found", {}).get("recipient")
if not recipient:
raise UserNotFoundError(UserNotFoundError.NOT_FOUND.format(username=username))
return recipient
async def _init_request(
session: httpx.AsyncClient,
fragment_hash: str,
recipient: str,
amount: int,
) -> str:
resp = await session.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={
"recipient": recipient,
"amount": amount,
"method": "initAdsTopupRequest",
},
)
result = parse_json_response(resp, "initAdsTopupRequest")
req_id = result.get("req_id")
if not req_id:
raise FragmentAPIError(FragmentAPIError.NO_REQUEST_ID.format(context="TON topup"))
return req_id
async def topup_ton(client: "FragmentClient", username: str, amount: int, show_sender: bool = True) -> AdsTopupResult:
if not isinstance(amount, int) or not (1 <= amount <= 1_000_000_000):
raise ConfigError(ConfigError.INVALID_TON_AMOUNT)
try:
fragment_hash = await get_fragment_hash(client.cookies, HEADERS, TON_PAGE)
account = await get_account_info(client)
async with httpx.AsyncClient(cookies=client.cookies) as session:
recipient = await _search_recipient(session, fragment_hash, username)
req_id = await _init_request(session, fragment_hash, recipient, amount)
tx_data = {
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": int(show_sender),
"method": "getAdsTopupLink",
}
transaction = await execute_transaction_request(session, HEADERS, tx_data, fragment_hash)
tx_hash = await process_transaction(client, transaction)
return AdsTopupResult(transaction_id=tx_hash, username=username, amount=amount)
except FragmentError:
raise
except Exception as exc:
raise UnexpectedError(UnexpectedError.UNEXPECTED.format(exc=exc)) from exc
fragmentapi/types/__init__.py
+59
View File
@@ -0,0 +1,59 @@
from fragmentapi.types.constants import (
BASE_HEADERS,
DEVICE,
PREMIUM_PAGE,
REQUIRED_COOKIE_KEYS,
STARS_PAGE,
SUPPORTED_WALLET_VERSIONS,
TON_PAGE,
WALLET_CLASSES,
WalletVersion,
)
from fragmentapi.types.exceptions import (
ClientError,
ConfigError,
CookiesError,
FragmentAPIError,
FragmentError,
HashFetchError,
OperationError,
RequestError,
TransactionError,
UnexpectedError,
UserNotFoundError,
VerificationError,
WalletError,
)
from fragmentapi.types.results import AdsTopupResult, PremiumResult, StarsResult
__all__ = [
# constants
"BASE_HEADERS",
"DEVICE",
"PREMIUM_PAGE",
"REQUIRED_COOKIE_KEYS",
"STARS_PAGE",
"SUPPORTED_WALLET_VERSIONS",
"TON_PAGE",
"WALLET_CLASSES",
"WalletVersion",
# client exceptions
"ClientError",
"ConfigError",
"CookiesError",
# fragment exceptions
"FragmentAPIError",
"FragmentError",
"HashFetchError",
"OperationError",
"RequestError",
"TransactionError",
"UnexpectedError",
"UserNotFoundError",
"VerificationError",
"WalletError",
# result types
"AdsTopupResult",
"PremiumResult",
"StarsResult",
]
app/core/constants.py → fragmentapi/types/constants.py
+4 -1
View File
@@ -10,10 +10,13 @@ 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, type] = {"V4R2": WalletV4R2, "V5R1": WalletV5R1}
# Required Fragment session cookie keys
REQUIRED_COOKIE_KEYS: tuple[str, ...] = ("stel_ssid", "stel_dt", "stel_token", "stel_ton_token")
# Fragment page URLs
STARS_PAGE: str = "https://fragment.com/stars/buy"
PREMIUM_PAGE: str = "https://fragment.com/premium/gift"
ADS_PAGE: str = "https://fragment.com/ads/topup"
TON_PAGE: str = "https://fragment.com/ads/topup"
# Tonkeeper device fingerprint — serialized once, reused in every tx_data payload.
DEVICE: str = json.dumps(
fragmentapi/types/exceptions.py
+106
View File
@@ -0,0 +1,106 @@
class FragmentError(Exception):
"""Base exception for all fragmentapi library errors."""
class ClientError(FragmentError):
"""Raised for client configuration and setup issues (bad params, invalid cookies)."""
class ConfigError(ClientError):
"""Raised when required client parameters are missing or invalid."""
MISSING_VARS = "Missing required parameter(s): {keys}."
UNSUPPORTED_VERSION = "Unsupported wallet_version '{version}'. Must be one of: {supported}."
INVALID_MONTHS = "Invalid duration. Choose 3, 6, or 12 months."
INVALID_STARS_AMOUNT = "Amount must be an integer between 50 and 1 000 000 stars."
INVALID_TON_AMOUNT = "Amount must be an integer between 1 and 1 000 000 000 TON."
class CookiesError(ClientError):
"""Raised when cookies are unreadable or missing required fields."""
READ_FAILED = "Failed to parse cookies: {exc}"
MISSING_KEYS = (
"Cookies are missing or have empty values for: {keys}. " "Open Fragment.com in your browser and copy fresh cookies."
)
class FragmentAPIError(FragmentError):
"""Raised for errors returned by Fragment's API responses."""
NO_REQUEST_ID = (
"Fragment did not return a request ID for '{context}'. " "The session may have expired — refresh your cookies."
)
class HashFetchError(FragmentAPIError):
"""Raised when the Fragment API hash cannot be fetched from the page."""
BAD_STATUS = "Fragment returned HTTP {status} for {url}. " "Check that your cookies are valid and not expired."
NOT_FOUND = (
"Fragment hash not found in the page source of {url}. " "The page structure may have changed or you are not logged in."
)
class UserNotFoundError(FragmentAPIError):
"""Raised when the target Telegram user is not found on Fragment."""
NOT_FOUND = (
"Telegram user '{username}' was not found on Fragment. " "Make sure the username is correct and the account exists."
)
class TransactionError(FragmentAPIError):
"""Raised when a TON transaction fails to build or broadcast."""
INVALID_PAYLOAD = (
"Fragment returned an invalid transaction payload. " "The API response is missing expected 'transaction.messages' data."
)
BROADCAST_FAILED = "Transaction broadcast failed: {exc}"
class RequestError(FragmentAPIError):
"""Raised when a Fragment API response cannot be parsed."""
UNPARSEABLE = "Fragment API returned an unparseable response for '{context}': {exc}"
class VerificationError(FragmentAPIError):
"""Raised when Fragment requires KYC verification before proceeding."""
KYC_REQUIRED = "Fragment requires identity (KYC) verification. " "Complete it at https://fragment.com/my/profile and retry."
class OperationError(FragmentError):
"""Raised for runtime operation failures unrelated to Fragment's API."""
class WalletError(OperationError):
"""Raised for TON wallet issues (connection, balance, account info)."""
LOW_BALANCE = "TON wallet balance is too low: {balance:.2f} TON. Minimum required is 0.056 TON."
BALANCE_CHECK_FAILED = "Wallet balance check failed: {exc}"
ACCOUNT_INFO_FAILED = "Failed to retrieve wallet account info: {exc}"
class UnexpectedError(OperationError):
"""Raised when an unexpected error occurs during an API call."""
UNEXPECTED = "An unexpected error occurred: {exc}"
__all__ = [
"FragmentError",
"ClientError",
"ConfigError",
"CookiesError",
"FragmentAPIError",
"HashFetchError",
"UserNotFoundError",
"TransactionError",
"RequestError",
"VerificationError",
"OperationError",
"WalletError",
"UnexpectedError",
]
fragmentapi/types/results.py
+34
View File
@@ -0,0 +1,34 @@
import time
from dataclasses import dataclass, field
__all__ = ["AdsTopupResult", "PremiumResult", "StarsResult"]
@dataclass
class PremiumResult:
"""Result of a successful Telegram Premium gift."""
transaction_id: str
username: str
months: int
timestamp: int = field(default_factory=lambda: int(time.time()))
@dataclass
class StarsResult:
"""Result of a successful Telegram Stars purchase."""
transaction_id: str
username: str
stars: int
timestamp: int = field(default_factory=lambda: int(time.time()))
@dataclass
class AdsTopupResult:
"""Result of a successful Telegram Ads balance top-up."""
transaction_id: str
username: str
amount: int
timestamp: int = field(default_factory=lambda: int(time.time()))
fragmentapi/utils/__init__.py
+16
View File
@@ -0,0 +1,16 @@
from fragmentapi.utils.client import (
execute_transaction_request,
get_fragment_hash,
parse_json_response,
)
from fragmentapi.utils.decoder import clean_decode
from fragmentapi.utils.wallet import get_account_info, process_transaction
__all__ = [
"clean_decode",
"execute_transaction_request",
"get_account_info",
"get_fragment_hash",
"parse_json_response",
"process_transaction",
]
fragmentapi/utils/client.py
+107
View File
@@ -0,0 +1,107 @@
import re
from typing import Any
import httpx
from fragmentapi.types import HashFetchError, RequestError, VerificationError
async def get_fragment_hash(
cookies: dict[str, Any],
headers: dict[str, str],
page_url: str,
) -> str:
"""Fetch the API hash from a Fragment page.
Fragment embeds a short-lived hash in each page's HTML that must be
included in every subsequent API request. This function loads the page
as a real browser navigation (not XHR) so Fragment returns full HTML.
Args:
cookies: Active Fragment session cookies.
headers: Base headers for the relevant Fragment page.
page_url: URL of the Fragment page to fetch the hash from.
Returns:
Lowercase hex hash string.
Raises:
HashFetchError: If the page returns a non-200 status or the hash
is not found in the response HTML.
"""
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": "https://fragment.com/",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"upgrade-insecure-requests": "1",
}
)
async with httpx.AsyncClient(cookies=cookies) as session:
response = await session.get(page_url, headers=page_headers)
if response.status_code != 200:
raise HashFetchError(HashFetchError.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 HashFetchError(HashFetchError.NOT_FOUND.format(url=page_url))
return match.group(1)
def parse_json_response(response: httpx.Response, context: str) -> dict[str, Any]:
"""Parse a Fragment API JSON response.
Args:
response: The HTTP response object.
context: Human-readable name of the API method, used in error messages.
Returns:
Parsed response as a dict.
Raises:
RequestError: If the response body cannot be decoded as JSON.
"""
try:
return response.json()
except Exception as exc:
raise RequestError(RequestError.UNPARSEABLE.format(context=context, exc=exc)) from exc
async def execute_transaction_request(
session: httpx.AsyncClient,
headers: dict,
tx_data: dict[str, Any],
fragment_hash: str,
) -> dict[str, Any]:
"""Post a transaction request to the Fragment API.
Args:
session: Active httpx session with Fragment cookies.
headers: Page-specific HTTP headers.
tx_data: Form data payload for the API method.
fragment_hash: Short-lived hash from the Fragment page.
Returns:
Parsed API response dict containing transaction data.
Raises:
VerificationError: If Fragment requires KYC verification.
RequestError: If the response cannot be parsed.
"""
url = f"https://fragment.com/api?hash={fragment_hash}"
resp = await session.post(url, headers=headers, data=tx_data)
transaction = parse_json_response(resp, tx_data.get("method", "transaction"))
if transaction.get("need_verify"):
raise VerificationError(VerificationError.KYC_REQUIRED)
return transaction
fragmentapi/utils/decoder.py
+20
View File
@@ -0,0 +1,20 @@
import base64
from pytoniq_core import Cell
from fragmentapi.types import RequestError
def clean_decode(payload: str) -> str:
s = payload.strip()
if not s:
return ""
s += "=" * (-len(s) % 4)
try:
boc = base64.b64decode(s)
cell = Cell.one_from_boc(boc)
sl = cell.begin_parse()
sl.load_uint(32) # op code — always 0 for text comment
return sl.load_snake_string().strip()
except Exception as exc:
raise RequestError(RequestError.UNPARSEABLE.format(context="payload decode", exc=exc)) from exc
fragmentapi/utils/wallet.py
+69
View File
@@ -0,0 +1,69 @@
import base64
from typing import TYPE_CHECKING, Any
from tonutils.clients import TonapiClient
from tonutils.types import NetworkGlobalID
from fragmentapi.types import WALLET_CLASSES, TransactionError, WalletError
from fragmentapi.utils.decoder import clean_decode
if TYPE_CHECKING:
from fragmentapi.client import FragmentClient
def _init_ton_client(client: "FragmentClient") -> TonapiClient:
return TonapiClient(network=NetworkGlobalID.MAINNET, api_key=client.api_key)
async def process_transaction(client: "FragmentClient", transaction_data: dict) -> str:
if "transaction" not in transaction_data or "messages" not in transaction_data["transaction"]:
raise TransactionError(TransactionError.INVALID_PAYLOAD)
# TODO: Investigate 406 'inbound external message rejected before smart-contract execution'.
# This happens when the previous transaction's seqno hasn't been confirmed on-chain yet,
# causing the wallet contract to reject the new message.
async with _init_ton_client(client) as ton:
wallet_cls = WALLET_CLASSES[client.wallet_version]
wallet, _, _, _ = wallet_cls.from_mnemonic(client=ton, mnemonic=client.seed)
# Check balance before broadcasting
try:
await wallet.refresh()
balance_ton = wallet.balance / 1_000_000_000
if balance_ton < 0.056:
raise WalletError(WalletError.LOW_BALANCE.format(balance=balance_ton))
except WalletError:
raise
except Exception as exc:
raise WalletError(WalletError.BALANCE_CHECK_FAILED.format(exc=exc)) from exc
try:
message = transaction_data["transaction"]["messages"][0]
payload = clean_decode(message["payload"])
result = await wallet.transfer(
destination=message["address"],
amount=int(message["amount"]), # nanotons, not TON
body=payload,
)
return result.normalized_hash
except (WalletError, TransactionError):
raise
except Exception as exc:
raise TransactionError(TransactionError.BROADCAST_FAILED.format(exc=exc)) from exc
async def get_account_info(client: "FragmentClient") -> dict[str, Any]:
async with _init_ton_client(client) 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
main.py
-66
View File
@@ -1,66 +0,0 @@
import asyncio
import logging
from app.core import setup_logging
from app.methods import buy_premium, buy_stars, topup_ton
logger = logging.getLogger(__name__)
async def topup_ton_example():
logger.info("Starting TON topup example")
# @bohd4nx - target username, 100 - TON amount (integer 1-1000000000 (one billion))
# show_sender=True — recipient sees who sent the topup
result = await topup_ton("@bohd4nx", 100, show_sender=True)
if result["success"]:
pass # Transaction successful, details are logged in the method
else:
logger.error(f"TON topup failed: {result['error']}")
async def buy_premium_example():
logger.info("Starting Premium purchase example")
# @bohd4nx - target username, 12 - months duration (3, 6, or 12 only)
# show_sender=True — recipient sees who gifted the Premium
result = await buy_premium("@bohd4nx", 12, show_sender=True)
if result["success"]:
pass # Transaction successful, details are logged in the method
else:
logger.error(f"Premium purchase failed: {result['error']}")
async def buy_stars_example():
logger.info("Starting Stars purchase example")
# @bohd4nx - target username, 1000000 - stars amount (integer 50-1000000 (one million))
# show_sender=True — recipient sees who sent the Stars
result = await buy_stars("@bohd4nx", 1000000, show_sender=True)
if result["success"]:
pass # Transaction successful, details are logged in the method
else:
logger.error(f"Stars purchase failed: {result['error']}")
async def main():
setup_logging()
logger.info("Starting Fragment API by @bohd4nx - examples")
await topup_ton_example()
await buy_premium_example()
await buy_stars_example()
logger.info("All examples completed")
if __name__ == "__main__":
logger.info("Fragment API by @bohd4nx - Usage Examples")
logger.info("Supported username formats: @username, username")
logger.info("Limits: TON minimum 1, Premium 3/6/12 months, Stars minimum 50")
logger.info("Setup: Copy .env.example to .env and fill all fields")
asyncio.run(main())
pyproject.toml
+40
View File
@@ -1,3 +1,43 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "fragmentapi"
version = "2026.1.0"
description = "Python library for the Fragment.com API — gift Telegram Stars, Premium, and top up TON Ads balance."
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.12"
authors = [{ name = "bohd4nx", url = "https://github.com/bohd4nx" }]
keywords = ["fragment", "telegram", "ton", "stars", "premium", "crypto", "blockchain"]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Natural Language :: English",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.12",
"Framework :: AsyncIO",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Internet",
"Topic :: Office/Business :: Financial",
"Typing :: Typed",
]
dependencies = [
"httpx==0.28.1",
"tonutils[pytoniq]==2.0.0",
]
[project.urls]
Homepage = "https://github.com/bohd4nx/FragmentAPI"
Repository = "https://github.com/bohd4nx/FragmentAPI"
Issues = "https://github.com/bohd4nx/FragmentAPI/issues"
[tool.hatch.build.targets.wheel]
packages = ["fragmentapi"]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["[0-9][0-9][0-9]_test_*.py"]
requirements.txt
-2
View File
@@ -1,4 +1,2 @@
python-dotenv==1.2.2
asyncio==4.0.0
httpx==0.28.1
tonutils[pytoniq]==2.0.0
tests/001_test_decode.py
+12 -7
View File
@@ -1,11 +1,11 @@
"""Tests for clean_decode() — BOC-encoded Fragment payloads decode to
human-readable UTF-8 with the Telegram label and Ref# intact."""
"""Tests for clean_decode() — BOC-encoded Fragment payloads decode to UTF-8."""
import re
import pytest
from app.utils.decoder import clean_decode
from fragmentapi.types import RequestError
from fragmentapi.utils.decoder import clean_decode
PAYLOADS = [
pytest.param(
@@ -24,12 +24,17 @@ PAYLOADS = [
@pytest.mark.parametrize("payload", PAYLOADS)
def test_payload(payload: str) -> None:
def test_decode_payload(payload: str) -> None:
result = clean_decode(payload)
assert "Telegram" in result
assert re.search(r"Ref#[A-Za-z0-9]+", result), f"no Ref# in {result!r}"
assert all(ord(c) <= 127 for c in result), f"non-ASCII chars in {result!r}"
assert all(ord(c) < 128 for c in result), f"non-ASCII chars in {result!r}"
def test_empty_input_returns_string() -> None:
assert isinstance(clean_decode(""), str)
def test_empty_payload_returns_empty_string() -> None:
assert clean_decode("") == ""
def test_invalid_payload_raises_request_error() -> None:
with pytest.raises(RequestError):
clean_decode("!!!not-valid-base64!!!")
tests/002_test_client.py
+81
View File
@@ -0,0 +1,81 @@
"""Unit tests for FragmentClient — init validation and cookie parsing (no network calls)."""
import json
import pytest
from fragmentapi import FragmentClient
from fragmentapi.types import ConfigError, CookiesError
VALID_SEED = "abandon " * 23 + "about"
VALID_API_KEY = "test_api_key"
VALID_COOKIES = {
"stel_ssid": "x",
"stel_dt": "x",
"stel_token": "x",
"stel_ton_token": "x",
}
def test_valid_init() -> None:
client = FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=VALID_COOKIES)
assert client.seed == VALID_SEED.strip()
assert client.api_key == VALID_API_KEY
assert client.wallet_version == "V5R1"
def test_wallet_version_v4r2() -> None:
client = FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=VALID_COOKIES, wallet_version="V4R2")
assert client.wallet_version == "V4R2"
def test_wallet_version_is_case_insensitive() -> None:
client = FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=VALID_COOKIES, wallet_version="v5r1")
assert client.wallet_version == "V5R1"
def test_missing_seed_raises() -> None:
with pytest.raises(ConfigError):
FragmentClient(seed="", api_key=VALID_API_KEY, cookies=VALID_COOKIES)
def test_whitespace_only_seed_raises() -> None:
with pytest.raises(ConfigError):
FragmentClient(seed=" ", api_key=VALID_API_KEY, cookies=VALID_COOKIES)
def test_missing_api_key_raises() -> None:
with pytest.raises(ConfigError):
FragmentClient(seed=VALID_SEED, api_key="", cookies=VALID_COOKIES)
def test_unsupported_wallet_version_raises() -> None:
with pytest.raises(ConfigError):
FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=VALID_COOKIES, wallet_version="V3R2")
def test_cookies_as_json_string() -> None:
client = FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=json.dumps(VALID_COOKIES))
assert client.cookies == VALID_COOKIES
def test_invalid_cookies_json_raises() -> None:
with pytest.raises(CookiesError):
FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies="{not valid json}")
def test_missing_cookie_key_raises() -> None:
with pytest.raises(CookiesError):
FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies={"stel_ssid": "x"})
def test_empty_cookie_value_raises() -> None:
bad = {**VALID_COOKIES, "stel_token": ""}
with pytest.raises(CookiesError):
FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=bad)
def test_whitespace_cookie_value_raises() -> None:
bad = {**VALID_COOKIES, "stel_ton_token": " "}
with pytest.raises(CookiesError):
FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=bad)
tests/002_test_hash.py → tests/003_test_hash.py
+2 -2
View File
@@ -5,8 +5,8 @@ import re
import pytest
from app.core.constants import BASE_HEADERS, STARS_PAGE
from app.utils.hash import get_fragment_hash
from fragmentapi.types import BASE_HEADERS, STARS_PAGE
from fragmentapi.utils import get_fragment_hash
@pytest.mark.asyncio
tests/conftest.py
+10 -6
View File
@@ -1,13 +1,17 @@
import pytest
import json
from pathlib import Path
from app.core.cookies import load_cookies
from app.core.exceptions import CookiesError
import pytest
@pytest.fixture
def cookies():
"""Load Fragment cookies; skip the test if they are unavailable."""
"""Load Fragment cookies from cookies.json; skip the test if unavailable."""
cookies_path = Path(__file__).resolve().parents[1] / "cookies.json"
if not cookies_path.exists():
pytest.skip("cookies.json not found")
try:
return load_cookies()
except CookiesError as exc:
with cookies_path.open("r", encoding="utf-8") as f:
return json.load(f)
except Exception as exc:
pytest.skip(f"Cookies unavailable — {exc}")