From d299a1f8042d4e5dbba69f68d9cc0e501d8c7afe Mon Sep 17 00:00:00 2001 From: bohd4nx Date: Sun, 18 Jan 2026 06:41:08 +0200 Subject: [PATCH] feat: Refactor configuration handling and enhance logging setup; update imports and improve error handling in API methods --- app/__meta__.py | 17 ++++++++++---- app/core/__init__.py | 5 +++-- app/core/config.py | 25 +++++++++++---------- app/core/logging.py | 33 +++++++++++++++++++++++++++ app/methods/premium.py | 46 ++++++++++++++++---------------------- app/methods/stars.py | 48 +++++++++++++++++----------------------- app/methods/ton.py | 46 ++++++++++++++++---------------------- app/utils/__init__.py | 4 ++-- app/utils/client.py | 32 +++++++++++++++++++++------ app/utils/decoder.py | 11 +-------- app/utils/transaction.py | 13 ++++++----- app/utils/wallet.py | 11 ++++----- main.py | 5 +++-- 13 files changed, 164 insertions(+), 132 deletions(-) create mode 100644 app/core/logging.py diff --git a/app/__meta__.py b/app/__meta__.py index 54cefbb..f5240f6 100644 --- a/app/__meta__.py +++ b/app/__meta__.py @@ -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", +] diff --git a/app/core/__init__.py b/app/core/__init__.py index ac3258e..00a22a3 100644 --- a/app/core/__init__.py +++ b/app/core/__init__.py @@ -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"] diff --git a/app/core/config.py b/app/core/config.py index be8ce34..5fe99d7 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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() diff --git a/app/core/logging.py b/app/core/logging.py new file mode 100644 index 0000000..2f49e44 --- /dev/null +++ b/app/core/logging.py @@ -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__) diff --git a/app/methods/premium.py b/app/methods/premium.py index 819e010..f3fad52 100644 --- a/app/methods/premium.py +++ b/app/methods/premium.py @@ -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: diff --git a/app/methods/stars.py b/app/methods/stars.py index 2c78fa0..93b110a 100644 --- a/app/methods/stars.py +++ b/app/methods/stars.py @@ -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: diff --git a/app/methods/ton.py b/app/methods/ton.py index e76fb27..634ca8f 100644 --- a/app/methods/ton.py +++ b/app/methods/ton.py @@ -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: diff --git a/app/utils/__init__.py b/app/utils/__init__.py index cb3ab54..a980904 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -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'] diff --git a/app/utils/client.py b/app/utils/client.py index 3c39164..24345bc 100644 --- a/app/utils/client.py +++ b/app/utils/client.py @@ -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() diff --git a/app/utils/decoder.py b/app/utils/decoder.py index 2fff9a8..a1093f8 100644 --- a/app/utils/decoder.py +++ b/app/utils/decoder.py @@ -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") diff --git a/app/utils/transaction.py b/app/utils/transaction.py index 764a0b5..c4a6dab 100644 --- a/app/utils/transaction.py +++ b/app/utils/transaction.py @@ -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] diff --git a/app/utils/wallet.py b/app/utils/wallet.py index 68ed0c3..bd262b0 100644 --- a/app/utils/wallet.py +++ b/app/utils/wallet.py @@ -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() diff --git a/main.py b/main.py index ad8e312..18ce897 100644 --- a/main.py +++ b/main.py @@ -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()