FOSS/FragmentAPI
0
0
Fork 0
mirror of https://github.com/bohd4nx/FragmentAPI.git synced 2026-07-25 06:14:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Refactor configuration handling and enhance logging setup; update imports and improve error handling in API methods

Browse Source
This commit is contained in:
bohd4nx
2026-01-18 06:41:08 +02:00
parent 76473993e2
commit d299a1f804
13 changed files with 164 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/__meta__.py
+13 -4
View File
@@ -1,4 +1,13 @@
__title__ = "Fragment API by @bohd4nx"
__version__ = "2025.1.2"
__author__ = "Bohdan (bohd4nx)"
__timestamp__ = "2025-11-24T12:00:00Z"
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
+3 -2
View File
@@ -1,3 +1,4 @@
from app.core.config import Config
from app.core.config import Config, config
from app.core.logging import logger, setup_logging
__all__ = ['Config']
__all__ = ["Config", "config", "logger", "setup_logging"]
app/core/config.py
+13 -12
View File
@@ -1,27 +1,31 @@
import logging
import os
import sys
from typing import Dict
from pathlib import Path
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
logger = logging.getLogger(__name__)
class Config:
def __init__(self):
required_keys = ['COOKIES', 'SEED', 'HASH', 'API_KEY']
missing_keys = []
env_path = Path(__file__).resolve().parents[2] / ".env"
self._config = {}
if not env_path.exists():
logger.error(".env file not found!")
sys.exit(1)
load_dotenv(env_path)
required_keys = ["COOKIES", "SEED", "HASH", "API_KEY"]
missing_keys: list[str] = []
for key in required_keys:
value = os.getenv(key, '').strip()
value = os.getenv(key, "").strip()
if not value:
missing_keys.append(key)
self._config[key.lower()] = value
setattr(self, key, value)
if missing_keys:
logger.error(f"Missing required environment variables: {', '.join(missing_keys)}")
@@ -30,8 +34,5 @@ class Config:
logger.info("Configuration loaded successfully")
def get_config(self) -> Dict[str, str]:
return self._config.copy()
def get(self, key: str, default: str = '') -> str:
return self._config.get(key, default)
config = Config()
app/core/logging.py
+33
View File
@@ -0,0 +1,33 @@
import logging
from app.__meta__ import APP_NAME
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(
f"{APP_NAME}.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("aiogram.dispatcher").setLevel(logging.INFO)
logging.getLogger("aiogram.event").setLevel(logging.ERROR)
logger = logging.getLogger(__name__)
app/methods/premium.py
+19 -27
View File
@@ -6,36 +6,34 @@ import httpx
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from app.core.config import Config
from app.utils import TransactionProcessor, WalletLinker, ApiClient, clean_decode
from app.core import config
from app.utils import TransactionProcessor, WalletLinker, ApiClient, clean_decode, parse_json_response
logger = logging.getLogger(__name__)
class FragmentPremium:
def __init__(self):
config_reader = Config()
self.config = config_reader.get_config()
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',
'cookie': self.config['cookies'],
'cookie': config.COOKIES,
'origin': 'https://fragment.com',
'referer': 'https://fragment.com/premium/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'
}
self.transaction_processor = TransactionProcessor(self.config, clean_decode)
self.wallet_linker = WalletLinker(self.config, self.headers, self.transaction_processor)
self.api_client = ApiClient(self.config, self.headers, self.wallet_linker)
self.transaction_processor = TransactionProcessor(clean_decode)
self.wallet_linker = WalletLinker(self.headers, self.transaction_processor)
self.api_client = ApiClient(self.headers, self.wallet_linker)
async def _get_account_info(self):
client = TonapiClient(api_key=self.config['api_key'], is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=self.config['seed'])
@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 {
@@ -53,34 +51,28 @@ class FragmentPremium:
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={self.config['hash']}",
search_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=search_data)
try:
search_result = search_resp.json()
except Exception as e:
logger.error(f"Failed to parse search response: {e}")
return {"success": False, "error": f"Invalid response from Fragment API: {str(e)}"}
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={self.config['hash']}",
await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=update_data)
init_data = {"recipient": recipient, "months": months, "method": "initGiftPremiumRequest"}
init_resp = await client.post(f"https://fragment.com/api?hash={self.config['hash']}",
init_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=init_data)
try:
init_result = init_resp.json()
except Exception as e:
logger.error(f"Failed to parse init response: {e}")
logger.error(f"Response status: {init_resp.status_code}")
logger.error(f"Response content: {init_resp.content[:200]}")
return {"success": False, "error": f"Invalid response from Fragment API: {str(e)}"}
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:
app/methods/stars.py
+20 -28
View File
@@ -6,36 +6,34 @@ import httpx
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from app.core.config import Config
from app.utils import TransactionProcessor, WalletLinker, ApiClient, clean_decode
from app.core import config
from app.utils import TransactionProcessor, WalletLinker, ApiClient, clean_decode, parse_json_response
logger = logging.getLogger(__name__)
class FragmentStars:
def __init__(self):
config_reader = Config()
self.config = config_reader.get_config()
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',
'cookie': self.config['cookies'],
'cookie': config.COOKIES,
'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'
}
self.transaction_processor = TransactionProcessor(self.config, clean_decode)
self.wallet_linker = WalletLinker(self.config, self.headers, self.transaction_processor)
self.api_client = ApiClient(self.config, self.headers, self.wallet_linker)
self.transaction_processor = TransactionProcessor(clean_decode)
self.wallet_linker = WalletLinker(self.headers, self.transaction_processor)
self.api_client = ApiClient(self.headers, self.wallet_linker)
async def _get_account_info(self):
client = TonapiClient(api_key=self.config['api_key'], is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=self.config['seed'])
@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 {
@@ -46,37 +44,31 @@ class FragmentStars:
}
async def buy_stars(self, username, amount):
if amount < 50:
return {"success": False, "error": "Minimum amount is 50 stars"}
if amount < 50 or not isinstance(amount, int):
return {"success": False, "error": "Amount must be an integer >= 50 stars"}
account = await self._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={self.config['hash']}",
search_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=search_data)
try:
search_result = search_resp.json()
except Exception as e:
logger.error(f"Failed to parse search response: {e}")
return {"success": False, "error": f"Invalid response from Fragment API: {str(e)}"}
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={self.config['hash']}",
init_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=init_data)
try:
init_result = init_resp.json()
except Exception as e:
logger.error(f"Failed to parse init response: {e}")
logger.error(f"Response status: {init_resp.status_code}")
logger.error(f"Response content: {init_resp.content[:200]}")
return {"success": False, "error": f"Invalid response from Fragment API: {str(e)}"}
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:
app/methods/ton.py
+19 -27
View File
@@ -6,36 +6,34 @@ import httpx
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from app.core.config import Config
from app.utils import TransactionProcessor, WalletLinker, ApiClient, clean_decode
from app.core import config
from app.utils import TransactionProcessor, WalletLinker, ApiClient, clean_decode, parse_json_response
logger = logging.getLogger(__name__)
class FragmentTon:
def __init__(self):
config_reader = Config()
self.config = config_reader.get_config()
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',
'cookie': self.config['cookies'],
'cookie': config.COOKIES,
'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'
}
self.transaction_processor = TransactionProcessor(self.config, clean_decode)
self.wallet_linker = WalletLinker(self.config, self.headers, self.transaction_processor)
self.api_client = ApiClient(self.config, self.headers, self.wallet_linker)
self.transaction_processor = TransactionProcessor(clean_decode)
self.wallet_linker = WalletLinker(self.headers, self.transaction_processor)
self.api_client = ApiClient(self.headers, self.wallet_linker)
async def _get_account_info(self):
client = TonapiClient(api_key=self.config['api_key'], is_testnet=False)
wallet, pub_key, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=self.config['seed'])
@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 {
@@ -53,34 +51,28 @@ class FragmentTon:
async with httpx.AsyncClient() as client:
update_data = {"mode": "new", "method": "updateAdsTopupState"}
await client.post(f"https://fragment.com/api?hash={self.config['hash']}",
await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=update_data)
search_data = {"query": username, "method": "searchAdsTopupRecipient"}
search_resp = await client.post(f"https://fragment.com/api?hash={self.config['hash']}",
search_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=search_data)
try:
search_result = search_resp.json()
except Exception as e:
logger.error(f"Failed to parse search response: {e}")
return {"success": False, "error": f"Invalid response from Fragment API: {str(e)}"}
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={self.config['hash']}",
init_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=init_data)
try:
init_result = init_resp.json()
except Exception as e:
logger.error(f"Failed to parse init response: {e}")
logger.error(f"Response status: {init_resp.status_code}")
logger.error(f"Response content: {init_resp.content[:200]}")
return {"success": False, "error": f"Invalid response from Fragment API: {str(e)}"}
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:
app/utils/__init__.py
+2 -2
View File
@@ -1,6 +1,6 @@
from app.utils.client import ApiClient
from app.utils.client import ApiClient, parse_json_response
from app.utils.decoder import clean_decode
from app.utils.transaction import TransactionProcessor
from app.utils.wallet import WalletLinker
__all__ = ['TransactionProcessor', 'WalletLinker', 'ApiClient', 'clean_decode']
__all__ = ['TransactionProcessor', 'WalletLinker', 'ApiClient', 'clean_decode', 'parse_json_response']
app/utils/client.py
+25 -7
View File
@@ -1,18 +1,36 @@
from typing import Dict, Any, Tuple
from typing import Any
import logging
import httpx
from app.core import config
def parse_json_response(
response: httpx.Response,
logger: logging.Logger,
context: str,
) -> tuple[dict[str, Any] | None, str | None]:
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)
class ApiClient:
def __init__(self, config: dict, headers: dict, wallet_linker):
self.config = config
def __init__(self, headers: dict, wallet_linker):
self.headers = headers
self.wallet_linker = wallet_linker
async def execute_transaction_request(self, tx_data: Dict[str, Any], account: Dict[str, Any]) -> Tuple[
bool, Dict[str, Any]]:
async def execute_transaction_request(
self,
tx_data: dict[str, Any],
account: dict[str, Any],
) -> tuple[bool, dict[str, Any]]:
async with httpx.AsyncClient() as client:
tx_resp = await client.post(f"https://fragment.com/api?hash={self.config['hash']}",
tx_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=tx_data)
transaction = tx_resp.json()
@@ -20,7 +38,7 @@ class ApiClient:
if not await self.wallet_linker.link_wallet(account):
return False, {"success": False, "error": "Failed to link wallet"}
tx_resp = await client.post(f"https://fragment.com/api?hash={self.config['hash']}",
tx_resp = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=tx_data)
transaction = tx_resp.json()
app/utils/decoder.py
+1 -10
View File
@@ -28,7 +28,7 @@ def clean_decode(payload: str) -> str:
# 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 main text with Ref#
# 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()
@@ -38,12 +38,3 @@ def clean_decode(payload: str) -> str:
logger.debug(f"Cleaned result: {result}")
return result
# payloads = [
# "te6ccgEBAgEALwABTgAAAAAxMDAwMDAwIFRlbGVncmFtIFN0YXJzIAoKUmVmI1RQb01wegEABkM3ZQ",
# "te6ccgEBAgEANAABTgAAAABUZWxlZ3JhbSBQcmVtaXVtIGZvciAxIHllYXIgCgpSZWYjcgEAEE9OQnM2cmNt",
# "te6ccgEBAgEAMAABTgAAAABUZWxlZ3JhbSBhY2NvdW50IHRvcCB1cCAKClJlZiNrMXpDRQEACFkxd3g"
# ]
# for p in payloads:
# logger.debug("\n" + clean_decode(p) + "\n")
app/utils/transaction.py
+7 -6
View File
@@ -1,24 +1,25 @@
import logging
from typing import Tuple, Optional
from tonutils.client import TonapiClient
from tonutils.wallet import WalletV5R1
from tonutils.wallet.messages import TransferMessage
from app.core import config
logger = logging.getLogger(__name__)
class TransactionProcessor:
def __init__(self, config: dict, clean_decode_func):
self.config = config
def __init__(self, clean_decode_func):
self._clean_decode = clean_decode_func
async def process_transaction(self, transaction_data: dict) -> Tuple[bool, Optional[str], Optional[str]]:
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
client = TonapiClient(api_key=self.config['api_key'], is_testnet=False)
wallet, _, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=self.config['seed'])
client = TonapiClient(api_key=config.API_KEY, is_testnet=False)
wallet, _, _, _ = WalletV5R1.from_mnemonic(client=client, mnemonic=config.SEED)
try:
message = transaction_data["transaction"]["messages"][0]
app/utils/wallet.py
+6 -5
View File
@@ -1,15 +1,16 @@
from typing import Dict, Any
from typing import Any
import httpx
from app.core import config
class WalletLinker:
def __init__(self, config: dict, headers: dict, transaction_processor):
self.config = config
def __init__(self, headers: dict, transaction_processor):
self.headers = headers
self.transaction_processor = transaction_processor
async def link_wallet(self, account: Dict[str, Any]) -> bool:
async def link_wallet(self, account: dict[str, Any]) -> bool:
async with httpx.AsyncClient() as client:
data = {
'account': account,
@@ -17,7 +18,7 @@ class WalletLinker:
'method': 'linkWallet'
}
response = await client.post(f"https://fragment.com/api?hash={self.config['hash']}",
response = await client.post(f"https://fragment.com/api?hash={config.HASH}",
headers=self.headers, data=data)
result = response.json()
main.py
+3 -2
View File
@@ -1,9 +1,9 @@
import asyncio
import logging
from app.methods import FragmentTon, FragmentPremium, FragmentStars
from app.methods import FragmentPremium, FragmentStars, FragmentTon
from app.core import setup_logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@@ -53,6 +53,7 @@ async def buy_stars_example():
async def main():
setup_logging()
logger.info("Starting Fragment API by @bohd4nx - examples")
await topup_ton_example()