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: migrate to tonutils v2, pytest, pyproject.toml

Browse Source 
- tonutils 2.0.0: TonapiClient → ToncenterV3Client, wallet.transfer()
- classes → plain async functions across all utils and methods
- core/: constants.py, cookies.py, exceptions.py extracted
- DEVICE constant in constants.py (single source of truth)
- account/device serialised with json.dumps() in all tx payloads
- tests: unittest → pytest, conftest.py fixtures, 001/002 naming
- pyproject.toml: pytest + ruff config
- min balance check: 0.056 TON
This commit is contained in:
bohd4nx
2026-03-05 04:08:35 +02:00
parent b1eae1dcb7
commit e3706f01bf
23 changed files with 631 additions and 565 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1 -3
View File
@@ -1,7 +1,5 @@
# Fragment.com cookies - copy from browser after login (Header String format)
# How to get fragment hash: open devtools -> Network -> api?hash=<your_fragment_hash_here>
# HASH = "your_fragment_hash_here" --- IGNORE ---
# Hash is now fetched dynamically, so this line is no longer needed.
# Hash is now fetched dynamically
# TON wallet seed phrase - 12 or 24 words separated by spaces
SEED = "your_ton_wallet_seed_phrase_here"
app/__meta__.py
-13
View File
@@ -1,13 +0,0 @@
APP_NAME = "FragmentAPI"
APP_TITLE = "Fragment API by @bohd4nx"
APP_VERSION = "2025.1.2"
APP_AUTHOR = "Bohdan (bohd4nx)"
APP_TIMESTAMP = "2025-11-24T12:00:00Z"
__all__ = [
"APP_NAME",
"APP_TITLE",
"APP_VERSION",
"APP_AUTHOR",
"APP_TIMESTAMP",
]
app/core/__init__.py
+32 -2
View File
@@ -1,4 +1,34 @@
from app.core.config import Config, config
from app.core.config import config
from app.core.constants import ADS_PAGE, BASE_HEADERS, DEVICE, PREMIUM_PAGE, STARS_PAGE
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__ = ["Config", "config", "logger", "setup_logging"]
__all__ = [
"ADS_PAGE",
"BASE_HEADERS",
"DEVICE",
"PREMIUM_PAGE",
"STARS_PAGE",
"ConfigError",
"CookiesError",
"FragmentError",
"HashFetchError",
"RequestError",
"TransactionError",
"UserNotFoundError",
"WalletError",
"config",
"load_cookies",
"logger",
"setup_logging",
]
app/core/config.py
+22 -18
View File
@@ -5,34 +5,38 @@ from pathlib import Path
from dotenv import load_dotenv
from app.core.exceptions import ConfigError
logger = logging.getLogger(__name__)
class Config:
def __init__(self):
SEED: str
API_KEY: str
def __init__(self) -> None:
env_path = Path(__file__).resolve().parents[2] / ".env"
if not env_path.exists():
logger.error(".env file not found!")
sys.exit(1)
raise ConfigError(
".env file not found. "
"Copy .env.example to .env and fill in SEED and API_KEY."
)
load_dotenv(env_path)
required_keys = ["SEED", "API_KEY"]
missing_keys: list[str] = []
for key in required_keys:
value = os.getenv(key, "").strip()
if not value:
missing_keys.append(key)
setattr(self, key, value)
if missing_keys:
logger.error(f"Missing required environment variables: {', '.join(missing_keys)}")
logger.error("Create .env file based on .env.example and fill all fields")
sys.exit(1)
logger.info("Configuration loaded successfully")
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)}. "
"Open .env and fill in all required fields."
)
self.SEED = os.getenv("SEED", "").strip()
self.API_KEY = os.getenv("API_KEY", "").strip()
try:
config = Config()
except ConfigError as e:
logger.error("Configuration error: %s", e)
sys.exit(1)
app/core/constants.py
+38
View File
@@ -0,0 +1,38 @@
import json
# 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"
# Tonkeeper device fingerprint — serialized once, reused in every tx_data payload.
DEVICE: str = json.dumps({
"platform": "iphone",
"appName": "Tonkeeper",
"appVersion": "5.5.2",
"maxProtocolVersion": 2,
"features": [
"SendTransaction",
{"name": "SendTransaction", "maxMessages": 255},
{"name": "SignData", "types": ["text", "binary", "cell"]},
],
})
# Base HTTP headers — shared across all Fragment API requests.
# Each method merges these with its own "referer" and "x-aj-referer".
BASE_HEADERS: dict[str, str] = {
"accept": "application/json, text/javascript, */*; q=0.01",
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": "en-US,en;q=0.9,uk;q=0.8,ru;q=0.7",
"content-type": "application/x-www-form-urlencoded; charset=UTF-8",
"origin": "https://fragment.com",
"priority": "u=1, i",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"user-agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.5 Mobile/15E148 Safari/604.1"
),
"x-requested-with": "XMLHttpRequest",
}
app/core/cookies.py
+35
View File
@@ -0,0 +1,35 @@
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
@@ -0,0 +1,42 @@
__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
+1 -5
View File
@@ -1,7 +1,5 @@
import logging
from app.__meta__ import APP_NAME
def setup_logging() -> None:
formatter = logging.Formatter(
@@ -14,7 +12,7 @@ def setup_logging() -> None:
console_handler.setFormatter(formatter)
file_handler = logging.FileHandler(
f"{APP_NAME}.log",
"FragmentAPI.log",
mode="w",
encoding="utf-8"
)
@@ -26,8 +24,6 @@ def setup_logging() -> None:
force=True
)
logging.getLogger("aiogram.dispatcher").setLevel(logging.INFO)
logging.getLogger("aiogram.event").setLevel(logging.ERROR)
logging.getLogger("httpx").setLevel(logging.INFO)
logging.getLogger("httpcore").setLevel(logging.WARNING)
app/methods/__init__.py
+4 -4
View File
@@ -1,5 +1,5 @@
from app.methods.premium import FragmentPremium
from app.methods.stars import FragmentStars
from app.methods.ton import FragmentTon
from app.methods.premium import buy_premium
from app.methods.stars import buy_stars
from app.methods.ton import topup_ton
__all__ = ['FragmentTon', 'FragmentPremium', 'FragmentStars']
__all__ = ['buy_premium', 'buy_stars', 'topup_ton']
app/methods/premium.py
+86 -103
View File
@@ -1,133 +1,116 @@
import base64
import json
import logging
import time
import httpx
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from app.core import config
from app.core import load_cookies
from app.core.constants import BASE_HEADERS, DEVICE, PREMIUM_PAGE
from app.core.exceptions import FragmentError, UserNotFoundError
from app.utils import (
TransactionProcessor,
WalletLinker,
ApiClient,
clean_decode,
parse_json_response,
load_cookies,
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
logger = logging.getLogger(__name__)
class FragmentPremium:
def __init__(self):
self.headers = {
"accept": "application/json, text/javascript, */*; q=0.01",
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": "en-US,en;q=0.9,uk;q=0.8,ru;q=0.7",
"content-type": "application/x-www-form-urlencoded; charset=UTF-8",
"origin": "https://fragment.com",
"referer": "https://fragment.com/premium/gift",
"user-agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.5 Mobile/15E148 Safari/604.1",
"x-requested-with": "XMLHttpRequest",
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": PREMIUM_PAGE,
"x-aj-referer": PREMIUM_PAGE,
}
self.cookies = load_cookies()
self.transaction_processor = TransactionProcessor(clean_decode)
self.wallet_linker = WalletLinker(self.headers, self.cookies, self.transaction_processor)
self.api_client = ApiClient(self.headers, self.cookies, self.wallet_linker)
@staticmethod
async def _get_account_info():
client = TonapiClient(api_key=config.API_KEY, is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.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.hex(),
"chain": "-239",
"walletStateInit": base64.b64encode(boc).decode()
}
async def buy_premium(self, username, months):
if months not in [3, 6, 12]:
return {"success": False, "error": "Invalid duration. Use 3, 6, or 12 months"}
fragment_hash = await get_fragment_hash(
self.cookies,
self.headers,
"https://fragment.com/premium/gift",
async def search_premium_recipient(
client: httpx.AsyncClient,
fragment_hash: str,
cookies: dict,
username: str,
months: int,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
data={"query": username, "months": months, "method": "searchPremiumGiftRecipient"},
)
if not fragment_hash:
raise RuntimeError("Failed to fetch Fragment hash")
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
account = await self._get_account_info()
async def init_gift_premium(
client: httpx.AsyncClient,
fragment_hash: str,
cookies: dict,
recipient: str,
months: int,
) -> str:
await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
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, cookies=cookies,
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) -> dict:
if months not in (3, 6, 12):
return {"success": False, "error": "Invalid duration. Choose 3, 6, or 12 months."}
try:
cookies = load_cookies()
fragment_hash = await get_fragment_hash(cookies, HEADERS, PREMIUM_PAGE)
account = await get_account_info()
async with httpx.AsyncClient() as client:
search_data = {"query": username, "months": months, "method": "searchPremiumGiftRecipient"}
search_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=search_data)
search_result, error = parse_json_response(search_resp, logger, "search")
if search_result is None:
return {"success": False, "error": f"Invalid response from Fragment API: {error}"}
recipient = search_result.get("found", {}).get("recipient")
if not recipient:
return {"success": False, "error": "User not found"}
update_data = {"mode": "new", "lv": "false", "dh": str(int(time.time())), "method": "updatePremiumState"}
await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=update_data)
init_data = {"recipient": recipient, "months": months, "method": "initGiftPremiumRequest"}
init_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=init_data)
init_result, error = parse_json_response(init_resp, logger, "init")
if init_result is None:
return {"success": False, "error": f"Invalid response from Fragment API: {error}"}
req_id = init_result.get("req_id")
if not req_id:
return {"success": False, "error": "Failed to initialize purchase"}
recipient = await search_premium_recipient(client, fragment_hash, cookies, username, months)
req_id = await init_gift_premium(client, fragment_hash, cookies, recipient, months)
tx_data = {
'account': account,
'device': {"appVersion": "5.4.3", "platform": "iphone",
"features": ["SendTransaction", {"maxMessages": 255, "name": "SendTransaction"},
{"types": ["text", "binary", "cell"], "name": "SignData"}],
"appName": "Tonkeeper", "maxProtocolVersion": 2},
'transaction': 1,
'id': req_id,
'show_sender': 1,
'ref': "OprzztcdJ",
'method': 'getGiftPremiumLink'
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": 1,
"method": "getGiftPremiumLink",
}
transaction = await execute_transaction_request(client, HEADERS, cookies, account, tx_data, fragment_hash)
request_success, transaction_result = await self.api_client.execute_transaction_request(
tx_data,
account,
fragment_hash,
)
if not request_success:
return transaction_result
success, error, tx_hash = await self.transaction_processor.process_transaction(transaction_result)
if success:
tx_hash = await process_transaction(transaction)
return {
"success": True,
"data": {
"transaction_id": tx_hash,
"username": username,
"months": months,
"timestamp": int(time.time())
}
"timestamp": int(time.time()),
},
}
return {"success": False, "error": error}
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
+80 -95
View File
@@ -1,125 +1,110 @@
import base64
import json
import logging
import time
import httpx
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from app.core import config
from app.core import load_cookies
from app.core.constants import BASE_HEADERS, DEVICE, STARS_PAGE
from app.core.exceptions import FragmentError, UserNotFoundError
from app.utils import (
TransactionProcessor,
WalletLinker,
ApiClient,
clean_decode,
parse_json_response,
load_cookies,
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
logger = logging.getLogger(__name__)
class FragmentStars:
def __init__(self):
self.headers = {
"accept": "application/json, text/javascript, */*; q=0.01",
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": "en-US,en;q=0.9,uk;q=0.8,ru;q=0.7",
"content-type": "application/x-www-form-urlencoded; charset=UTF-8",
"origin": "https://fragment.com",
"referer": "https://fragment.com/stars/buy",
"user-agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.5 Mobile/15E148 Safari/604.1",
"x-requested-with": "XMLHttpRequest",
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": STARS_PAGE,
"x-aj-referer": STARS_PAGE,
}
self.cookies = load_cookies()
self.transaction_processor = TransactionProcessor(clean_decode)
self.wallet_linker = WalletLinker(self.headers, self.cookies, self.transaction_processor)
self.api_client = ApiClient(self.headers, self.cookies, self.wallet_linker)
@staticmethod
async def _get_account_info():
client = TonapiClient(api_key=config.API_KEY, is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.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.hex(),
"chain": "-239",
"walletStateInit": base64.b64encode(boc).decode()
}
async def buy_stars(self, username, amount):
if amount < 50 or not isinstance(amount, int):
return {"success": False, "error": "Amount must be an integer >= 50 stars"}
fragment_hash = await get_fragment_hash(
self.cookies,
self.headers,
"https://fragment.com/stars/buy",
async def search_stars_recipient(
client: httpx.AsyncClient,
fragment_hash: str,
cookies: dict,
username: str,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
data={"query": username, "quantity": "", "method": "searchStarsRecipient"},
)
if not fragment_hash:
raise RuntimeError("Failed to fetch Fragment hash")
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
account = await self._get_account_info()
async def init_buy_stars(
client: httpx.AsyncClient,
fragment_hash: str,
cookies: dict,
recipient: str,
amount: int,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
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) -> dict:
if not isinstance(amount, int) or amount < 50:
return {"success": False, "error": "Amount must be an integer >= 50 stars."}
try:
cookies = load_cookies()
fragment_hash = await get_fragment_hash(cookies, HEADERS, STARS_PAGE)
account = await get_account_info()
async with httpx.AsyncClient() as client:
search_data = {"query": username, "quantity": "", "method": "searchStarsRecipient"}
search_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=search_data)
search_result, error = parse_json_response(search_resp, logger, "search")
if search_result is None:
return {"success": False, "error": f"Invalid response from Fragment API: {error}"}
recipient = search_result.get("found", {}).get("recipient")
if not recipient:
return {"success": False, "error": "User not found"}
init_data = {"recipient": recipient, "quantity": amount, "method": "initBuyStarsRequest"}
init_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=init_data)
init_result, error = parse_json_response(init_resp, logger, "init")
if init_result is None:
return {"success": False, "error": f"Invalid response from Fragment API: {error}"}
req_id = init_result.get("req_id")
if not req_id:
return {"success": False, "error": "Failed to initialize purchase"}
recipient = await search_stars_recipient(client, fragment_hash, cookies, username)
req_id = await init_buy_stars(client, fragment_hash, cookies, recipient, amount)
tx_data = {
'account': account,
'device': "iPhone15,2",
'transaction': 1,
'id': req_id,
'show_sender': 0,
'method': 'getBuyStarsLink'
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": 1,
"method": "getBuyStarsLink",
}
transaction = await execute_transaction_request(client, HEADERS, cookies, account, tx_data, fragment_hash)
request_success, transaction_result = await self.api_client.execute_transaction_request(
tx_data,
account,
fragment_hash,
)
if not request_success:
return transaction_result
success, error, tx_hash = await self.transaction_processor.process_transaction(transaction_result)
if success:
tx_hash = await process_transaction(transaction)
return {
"success": True,
"data": {
"transaction_id": tx_hash,
"username": username,
"amount": amount,
"timestamp": int(time.time())
}
"timestamp": int(time.time()),
},
}
return {"success": False, "error": error}
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
+85 -102
View File
@@ -1,132 +1,115 @@
import base64
import json
import logging
import time
import httpx
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from app.core import config
from app.core import load_cookies
from app.core.constants import ADS_PAGE, BASE_HEADERS, DEVICE
from app.core.exceptions import FragmentError, UserNotFoundError
from app.utils import (
TransactionProcessor,
WalletLinker,
ApiClient,
clean_decode,
parse_json_response,
load_cookies,
execute_transaction_request,
get_account_info,
get_fragment_hash,
parse_json_response,
process_transaction,
)
logger = logging.getLogger(__name__)
class FragmentTon:
def __init__(self):
self.headers = {
"accept": "application/json, text/javascript, */*; q=0.01",
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": "en-US,en;q=0.9,uk;q=0.8,ru;q=0.7",
"content-type": "application/x-www-form-urlencoded; charset=UTF-8",
"origin": "https://fragment.com",
"referer": "https://fragment.com/ads/topup",
"user-agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.5 Mobile/15E148 Safari/604.1",
"x-requested-with": "XMLHttpRequest",
# Page-specific headers
HEADERS: dict[str, str] = {
**BASE_HEADERS,
"referer": ADS_PAGE,
"x-aj-referer": ADS_PAGE,
}
self.cookies = load_cookies()
self.transaction_processor = TransactionProcessor(clean_decode)
self.wallet_linker = WalletLinker(self.headers, self.cookies, self.transaction_processor)
self.api_client = ApiClient(self.headers, self.cookies, self.wallet_linker)
@staticmethod
async def _get_account_info():
client = TonapiClient(api_key=config.API_KEY, is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.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.hex(),
"chain": "-239",
"walletStateInit": base64.b64encode(boc).decode()
}
async def topup_ton(self, username, amount):
if amount < 1 or not isinstance(amount, int):
return {"success": False, "error": "Amount must be an integer >= 1 TON"}
fragment_hash = await get_fragment_hash(
self.cookies,
self.headers,
"https://fragment.com/ads/topup",
async def search_ads_recipient(
client: httpx.AsyncClient,
fragment_hash: str,
cookies: dict,
username: str,
) -> str:
await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
data={"mode": "new", "method": "updateAdsTopupState"},
)
if not fragment_hash:
raise RuntimeError("Failed to fetch Fragment hash")
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
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
account = await self._get_account_info()
async def init_ads_topup(
client: httpx.AsyncClient,
fragment_hash: str,
cookies: dict,
recipient: str,
amount: int,
) -> str:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS, cookies=cookies,
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) -> dict:
if not isinstance(amount, int) or amount < 1:
return {"success": False, "error": "Amount must be an integer >= 1 TON."}
try:
cookies = load_cookies()
fragment_hash = await get_fragment_hash(cookies, HEADERS, ADS_PAGE)
account = await get_account_info()
async with httpx.AsyncClient() as client:
update_data = {"mode": "new", "method": "updateAdsTopupState"}
await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=update_data)
search_data = {"query": username, "method": "searchAdsTopupRecipient"}
search_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=search_data)
search_result, error = parse_json_response(search_resp, logger, "search")
if search_result is None:
return {"success": False, "error": f"Invalid response from Fragment API: {error}"}
recipient = search_result.get("found", {}).get("recipient")
if not recipient:
return {"success": False, "error": "User not found"}
init_data = {"recipient": recipient, "amount": amount, "method": "initAdsTopupRequest"}
init_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=init_data)
init_result, error = parse_json_response(init_resp, logger, "init")
if init_result is None:
return {"success": False, "error": f"Invalid response from Fragment API: {error}"}
req_id = init_result.get("req_id")
if not req_id:
return {"success": False, "error": "Failed to initialize topup"}
recipient = await search_ads_recipient(client, fragment_hash, cookies, username)
req_id = await init_ads_topup(client, fragment_hash, cookies, recipient, amount)
tx_data = {
'account': account,
'device': {"appVersion": "5.4.3", "platform": "iphone",
"features": ["SendTransaction", {"maxMessages": 255, "name": "SendTransaction"},
{"types": ["text", "binary", "cell"], "name": "SignData"}],
"appName": "Tonkeeper", "maxProtocolVersion": 2},
'transaction': 1,
'id': req_id,
'show_sender': 1,
'method': 'getAdsTopupLink'
"account": json.dumps(account),
"device": DEVICE,
"transaction": 1,
"id": req_id,
"show_sender": 1,
"method": "getAdsTopupLink",
}
transaction = await execute_transaction_request(client, HEADERS, cookies, account, tx_data, fragment_hash)
request_success, transaction_result = await self.api_client.execute_transaction_request(
tx_data,
account,
fragment_hash,
)
if not request_success:
return transaction_result
success, error, tx_hash = await self.transaction_processor.process_transaction(transaction_result)
if success:
tx_hash = await process_transaction(transaction)
return {
"success": True,
"data": {
"transaction_id": tx_hash,
"username": username,
"amount": amount,
"timestamp": int(time.time())
}
"timestamp": int(time.time()),
},
}
return {"success": False, "error": error}
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
+8 -9
View File
@@ -1,16 +1,15 @@
from app.utils.client import ApiClient, parse_json_response
from app.utils.cookies import load_cookies
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.transaction import TransactionProcessor
from app.utils.wallet import WalletLinker
from app.utils.transaction import process_transaction
from app.utils.wallet import get_account_info, link_wallet
__all__ = [
'TransactionProcessor',
'WalletLinker',
'ApiClient',
'clean_decode',
'execute_transaction_request',
'get_account_info',
'get_fragment_hash',
'link_wallet',
'parse_json_response',
'load_cookies',
'get_fragment_hash'
'process_transaction',
]
app/utils/client.py
+28 -33
View File
@@ -3,47 +3,42 @@ from typing import Any
import httpx
from app.core.exceptions import RequestError, WalletError
from app.utils.wallet import link_wallet
def parse_json_response(
response: httpx.Response,
logger: logging.Logger,
context: str,
) -> tuple[dict[str, Any] | None, str | None]:
logger = logging.getLogger(__name__)
def parse_json_response(response: httpx.Response, context: str) -> dict[str, Any]:
try:
return response.json(), None
except Exception as e:
logger.error(f"Failed to parse {context} response: {e}")
logger.error(f"Response content: {response.content[:200]}")
return None, str(e)
return response.json()
except Exception as exc:
raise RequestError(
f"Fragment API returned an unparseable response for '{context}': {exc}"
) from exc
class ApiClient:
def __init__(self, headers: dict, cookies: dict, wallet_linker):
self.headers = headers
self.cookies = cookies
self.wallet_linker = wallet_linker
async def execute_transaction_request(
self,
tx_data: dict[str, Any],
client: httpx.AsyncClient,
headers: dict,
cookies: dict,
account: dict[str, Any],
tx_data: dict[str, Any],
fragment_hash: str,
) -> tuple[bool, dict[str, Any]]:
async with httpx.AsyncClient() as client:
tx_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=tx_data)
transaction, error = parse_json_response(tx_resp, logging.getLogger(__name__), "transaction")
if transaction is None:
return False, {"success": False, "error": f"Invalid response from Fragment API: {error}"}
) -> dict[str, Any]:
url = f"https://fragment.com/api?hash={fragment_hash}"
resp = await client.post(url, headers=headers, cookies=cookies, data=tx_data)
transaction = parse_json_response(resp, tx_data.get("method", "transaction"))
if transaction.get("need_verify"):
if not await self.wallet_linker.link_wallet(account, fragment_hash):
return False, {"success": False, "error": "Failed to link wallet"}
if not await link_wallet(client, headers, cookies, 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, cookies=cookies, data=tx_data)
transaction = parse_json_response(resp, tx_data.get("method", "transaction"))
tx_resp = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=tx_data)
transaction, error = parse_json_response(tx_resp, logging.getLogger(__name__), "transaction")
if transaction is None:
return False, {"success": False, "error": f"Invalid response from Fragment API: {error}"}
return transaction
return True, transaction
app/utils/cookies.py
-35
View File
@@ -1,35 +0,0 @@
import json
import logging
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
def load_cookies() -> dict[str, Any]:
cookies_path = Path(__file__).resolve().parents[2] / "cookies.json"
if not cookies_path.exists():
logger.error("cookies.json file not found!")
return {}
try:
with cookies_path.open("r", encoding="utf-8") as file:
cookies = json.load(file)
required_keys = ["stel_ssid", "stel_dt", "stel_token", "stel_ton_token"]
missing_or_empty = [
key for key in required_keys
if not str(cookies.get(key, "")).strip()
]
if missing_or_empty:
logger.warning(
"cookies.json has missing or empty values: %s",
", ".join(missing_or_empty)
)
return cookies
except Exception as exc:
logger.error(f"Failed to load cookies.json: {exc}")
return {}
app/utils/decoder.py
+10 -24
View File
@@ -7,34 +7,20 @@ logger = logging.getLogger(__name__)
def clean_decode(payload: str) -> str:
logger.debug(f"Original payload: {payload}")
logger.debug("Original payload: %s", payload)
# Decode raw bytes first
clean = ''.join(c for c in payload if c.isalnum() or c in '+/=')
clean += '=' * (-len(clean) % 4)
raw_bytes = base64.b64decode(clean)
logger.debug(f"Raw decoded bytes: {raw_bytes}")
# 1. Clean Base64
# Strip non-Base64 chars and decode
s = re.sub(r'[^A-Za-z0-9+/=]', '', payload.strip())
s += '=' * (-len(s) % 4)
text = base64.b64decode(s).decode('utf-8', errors='ignore')
# 2. Base64 -> bytes
b = base64.b64decode(s)
# Keep only printable characters
text = ''.join(c for c in text if c in string.printable or c.isspace())
# 3. Decode UTF-8, ignoring invalid bytes
t = b.decode('utf-8', errors='ignore')
# 4. Remove binary characters, keep only printable + whitespace
t = ''.join(c for c in t if c in string.printable or c.isspace())
# 5. Extract the main text with Ref#
match = re.search(r'([0-9]*\s*Telegram .*?Ref#[A-Za-z0-9]+)', t, re.S)
if match:
result = match.group(1).strip()
else:
result = t.strip()
logger.debug(f"Cleaned result: {result}")
# Extract "Telegram … Ref#XXXX" block
match = re.search(r'([0-9]*\s*Telegram .*?Ref#[A-Za-z0-9]+)', text, re.S)
result = match.group(1).strip() if match else text.strip()
logger.debug("Decoded result: %s", result)
return result
app/utils/hash.py
+17 -13
View File
@@ -4,6 +4,8 @@ from typing import Any
import httpx
from app.core.exceptions import HashFetchError
logger = logging.getLogger(__name__)
@@ -11,25 +13,27 @@ async def get_fragment_hash(
cookies: dict[str, Any],
headers: dict[str, str],
page_url: str,
) -> str | None:
request_headers = {
) -> str:
page_headers = {
**headers,
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept-language": headers.get("accept-language") or headers.get("Accept-Language", "en-US,en;q=0.9"),
"user-agent": headers.get("user-agent") or headers.get("User-Agent", ""),
"referer": "https://fragment.com/",
}
async with httpx.AsyncClient(cookies=cookies) as client:
response = await client.get(page_url, headers=request_headers)
response = await client.get(page_url, headers=page_headers)
if response.status_code != 200:
logger.error("Failed to fetch Fragment page for hash: %s", response.status_code)
return None
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."
)
text = response.text
match = re.search(r"(?:https://fragment\.com)?/api\?hash=([a-f0-9]+)", text)
if match:
return match.group(1)
logger.error("Failed to extract Fragment hash from page")
return None
app/utils/transaction.py
+28 -37
View File
@@ -1,58 +1,49 @@
import logging
from tonutils.client import TonapiClient, ToncenterV3Client
from tonutils.client import ToncenterV3Client
from tonutils.wallet import WalletV5R1
from tonutils.wallet.messages import TransferMessage
from app.core import config
from app.core.exceptions import TransactionError, WalletError
from app.utils.decoder import clean_decode
logger = logging.getLogger(__name__)
class TransactionProcessor:
def __init__(self, clean_decode_func):
self._clean_decode = clean_decode_func
async def process_transaction(transaction_data: dict) -> str:
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."
)
@staticmethod
async def _check_wallet_balance() -> tuple[bool, str | None]:
client = ToncenterV3Client(is_testnet=False, rps=1, max_retries=1)
client = ToncenterV3Client(api_key=config.API_KEY, is_testnet=False)
wallet, _, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=config.SEED)
# Check balance before broadcasting
try:
balance = await wallet.balance()
if float(balance) < 0.056:
raise WalletError(
f"TON wallet balance is too low: {balance} TON. "
"Minimum required is 0.056 TON."
)
except WalletError:
raise
except Exception as exc:
return False, f"Wallet balance check failed: {exc}"
try:
if float(balance) <= 0:
return False, "Wallet balance is zero"
except Exception:
pass
return True, None
async def process_transaction(self, transaction_data: dict) -> tuple[bool, str | None, str | None]:
if "transaction" not in transaction_data or "messages" not in transaction_data["transaction"]:
return False, "Invalid transaction", None
ready, reason = await self._check_wallet_balance()
if not ready:
return False, reason or "Wallet is not ready", None
client = TonapiClient(api_key=config.API_KEY, is_testnet=False)
wallet, _, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=config.SEED)
raise WalletError(f"Wallet balance check failed: {exc}") from exc
try:
message = transaction_data["transaction"]["messages"][0]
payload = self._clean_decode(message["payload"])
payload = clean_decode(message["payload"])
messages = [TransferMessage(
return await wallet.transfer(
destination=message["address"],
amount=int(message["amount"]) / 1000000000,
body=payload
)]
amount=int(message["amount"]) / 1_000_000_000,
body=payload,
)
except (WalletError, TransactionError):
raise
except Exception as exc:
raise TransactionError(f"Transaction broadcast failed: {exc}") from exc
tx_hash = await wallet.batch_transfer_messages(messages=messages)
return True, None, tx_hash
except Exception as e:
return False, str(e), None
app/utils/wallet.py
+48 -17
View File
@@ -1,31 +1,62 @@
import base64
import json
import logging
from typing import Any
import httpx
from tonutils.client import ToncenterV3Client
from tonutils.wallet import WalletV5R1
from app.core import config
from app.core.constants import DEVICE
from app.core.exceptions import TransactionError, WalletError
from app.utils.transaction import process_transaction
logger = logging.getLogger(__name__)
class WalletLinker:
def __init__(self, headers: dict, cookies: dict, transaction_processor):
self.headers = headers
self.cookies = cookies
self.transaction_processor = transaction_processor
async def link_wallet(self, account: dict[str, Any], fragment_hash: str) -> bool:
async with httpx.AsyncClient() as client:
data = {
'account': account,
'device': "iPhone15,2",
'method': 'linkWallet'
async def get_account_info() -> dict[str, Any]:
try:
client = ToncenterV3Client(api_key=config.API_KEY, is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.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.hex(),
"chain": "-239",
"walletStateInit": base64.b64encode(boc).decode(),
}
except Exception as exc:
raise WalletError(f"Failed to retrieve wallet account info: {exc}") from exc
response = await client.post(f"https://fragment.com/api?hash={fragment_hash}",
headers=self.headers, cookies=self.cookies, data=data)
result = response.json()
async def link_wallet(
client: httpx.AsyncClient,
headers: dict,
cookies: dict,
account: dict[str, Any],
fragment_hash: str,
) -> bool:
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=headers,
cookies=cookies,
data={
"account": json.dumps(account),
"device": DEVICE,
"method": "linkWallet",
},
)
result = resp.json()
if result.get("ok"):
return True
if "transaction" in result:
success, _, _ = await self.transaction_processor.process_transaction(result)
return success
try:
await process_transaction(result)
return True
except (TransactionError, WalletError):
return False
return False
main.py
+7 -10
View File
@@ -1,8 +1,8 @@
import asyncio
import logging
from app.methods import FragmentPremium, FragmentStars, FragmentTon
from app.core import setup_logging
from app.methods import buy_premium, buy_stars, topup_ton
logger = logging.getLogger(__name__)
@@ -10,9 +10,8 @@ logger = logging.getLogger(__name__)
async def topup_ton_example():
logger.info("Starting TON topup example")
ton_client = FragmentTon()
# @bohd4nx - target username, 5 - TON amount (integer 1-1000000000 (one billion))
result = await ton_client.topup_ton("@bohd4nx", 100)
# @bohd4nx - target username, 100 - TON amount (integer 1-1000000000 (one billion))
result = await topup_ton("@bohd4nx", 100)
if result["success"]:
data = result["data"]
@@ -25,9 +24,8 @@ async def topup_ton_example():
async def buy_premium_example():
logger.info("Starting Premium purchase example")
premium_client = FragmentPremium()
# @bohd4nx - target username, 6 - months duration (3, 6, or 12 only)
result = await premium_client.buy_premium("@bohd4nx", 12)
# @bohd4nx - target username, 12 - months duration (3, 6, or 12 only)
result = await buy_premium("@bohd4nx", 12)
if result["success"]:
data = result["data"]
@@ -40,9 +38,8 @@ async def buy_premium_example():
async def buy_stars_example():
logger.info("Starting Stars purchase example")
stars_client = FragmentStars()
# @bohd4nx - target username, 50 - stars amount (integer 50-1000000 (one million))
result = await stars_client.buy_stars("@bohd4nx", 1000000)
# @bohd4nx - target username, 1000000 - stars amount (integer 50-1000000 (one million))
result = await buy_stars("@bohd4nx", 1000000)
if result["success"]:
data = result["data"]
pyproject.toml
+17
View File
@@ -0,0 +1,17 @@
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["[0-9][0-9][0-9]_test_*.py"]
asyncio_mode = "auto"
addopts = "-v --tb=short"
[tool.ruff]
target-version = "py312"
line-length = 100
src = ["app", "tests"]
[tool.ruff.lint]
select = ["E", "W", "F", "I", "UP", "B", "C4", "RUF"]
ignore = ["E501"]
[tool.ruff.lint.isort]
known-first-party = ["app"]
requirements.txt
+2 -2
View File
@@ -1,4 +1,4 @@
python-dotenv==1.2.1
python-dotenv==1.2.2
asyncio==4.0.0
httpx==0.28.1
tonutils==0.5.6
tonutils==2.0.0