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

refactor: streamline error messages and improve code formatting for consistency

Browse Source
This commit is contained in:
bohd4nx
2026-03-09 03:32:45 +02:00
parent c3de8a8ca3
commit 6c869428aa
10 changed files with 67 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/core/config.py
+1 -2
View File
@@ -37,8 +37,7 @@ class Config:
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))}."
f"Unsupported WALLET_VERSION '{version}'. " f"Must be one of: {', '.join(sorted(SUPPORTED_WALLET_VERSIONS))}."
)
self.WALLET_VERSION: WalletVersion = version # type: ignore[assignment]
app/core/cookies.py
+1 -3
View File
@@ -14,9 +14,7 @@ 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."
)
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:
app/core/logging.py
+1 -3
View File
@@ -2,9 +2,7 @@ import logging
def setup_logging() -> None:
formatter = logging.Formatter(
fmt="[%(asctime)s] - %(levelname)s: %(message)s", datefmt="%d.%m.%y %H:%M:%S"
)
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)
app/methods/premium.py
+19 -7
View File
@@ -39,7 +39,11 @@ async def search_premium_recipient(
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"query": username, "months": months, "method": "searchPremiumGiftRecipient"},
data={
"query": username,
"months": months,
"method": "searchPremiumGiftRecipient",
},
)
result = parse_json_response(resp, "searchPremiumGiftRecipient")
recipient = result.get("found", {}).get("recipient")
@@ -70,7 +74,11 @@ async def init_gift_premium(
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"recipient": recipient, "months": months, "method": "initGiftPremiumRequest"},
data={
"recipient": recipient,
"months": months,
"method": "initGiftPremiumRequest",
},
)
result = parse_json_response(resp, "initGiftPremiumRequest")
req_id = result.get("req_id")
@@ -84,7 +92,10 @@ async def init_gift_premium(
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."}
return {
"success": False,
"error": "Invalid duration. Choose 3, 6, or 12 months.",
}
try:
logger.info("Loading session cookies")
@@ -112,14 +123,15 @@ async def buy_premium(username: str, months: int) -> dict:
"show_sender": 1,
"method": "getGiftPremiumLink",
}
transaction = await execute_transaction_request(
client, HEADERS, account, tx_data, fragment_hash
)
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
"Premium purchase successful: %s months -> %s | tx: %s",
months,
username,
tx_hash,
)
return {
"success": True,
app/methods/stars.py
+12 -5
View File
@@ -59,7 +59,11 @@ async def init_buy_stars(
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"recipient": recipient, "quantity": amount, "method": "initBuyStarsRequest"},
data={
"recipient": recipient,
"quantity": amount,
"method": "initBuyStarsRequest",
},
)
result = parse_json_response(resp, "initBuyStarsRequest")
req_id = result.get("req_id")
@@ -101,13 +105,16 @@ async def buy_stars(username: str, amount: int) -> dict:
"show_sender": 1,
"method": "getBuyStarsLink",
}
transaction = await execute_transaction_request(
client, HEADERS, account, tx_data, fragment_hash
)
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)
logger.info(
"Stars purchase successful: %s stars -> %s | tx: %s",
amount,
username,
tx_hash,
)
return {
"success": True,
"data": {
app/methods/ton.py
+15 -7
View File
@@ -4,7 +4,14 @@ import time
import httpx
from app.core import ADS_PAGE, BASE_HEADERS, DEVICE, FragmentError, UserNotFoundError, load_cookies
from app.core import (
ADS_PAGE,
BASE_HEADERS,
DEVICE,
FragmentError,
UserNotFoundError,
load_cookies,
)
from app.utils import (
execute_transaction_request,
get_account_info,
@@ -57,14 +64,17 @@ async def init_ads_topup(
resp = await client.post(
f"https://fragment.com/api?hash={fragment_hash}",
headers=HEADERS,
data={"recipient": recipient, "amount": amount, "method": "initAdsTopupRequest"},
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."
"Fragment did not return a request ID for this TON topup. " "The session may have expired — refresh your cookies."
)
return req_id
@@ -99,9 +109,7 @@ async def topup_ton(username: str, amount: int) -> dict:
"show_sender": 1,
"method": "getAdsTopupLink",
}
transaction = await execute_transaction_request(
client, HEADERS, account, tx_data, fragment_hash
)
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)
app/utils/client.py
+2 -5
View File
@@ -13,9 +13,7 @@ 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
raise RequestError(f"Fragment API returned an unparseable response for '{context}': {exc}") from exc
async def execute_transaction_request(
@@ -33,8 +31,7 @@ async def execute_transaction_request(
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."
"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"))
app/utils/hash.py
+7 -1
View File
@@ -20,7 +20,13 @@ async def get_fragment_hash(
k: v
for k, v in headers.items()
if k
not in ("accept", "accept-encoding", "content-type", "x-requested-with", "x-aj-referer")
not in (
"accept",
"accept-encoding",
"content-type",
"x-requested-with",
"x-aj-referer",
)
}
page_headers.update(
{
app/utils/wallet.py
+1 -4
View File
@@ -38,10 +38,7 @@ async def process_transaction(transaction_data: dict) -> str:
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."
)
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:
pyproject.toml
+8 -6
View File
@@ -4,14 +4,16 @@ python_files = ["[0-9][0-9][0-9]_test_*.py"]
asyncio_mode = "auto"
addopts = "-v --tb=short"
[tool.black]
line-length = 128
target-version = ["py312"]
[tool.ruff]
target-version = "py312"
line-length = 100
src = ["app", "tests"]
line-length = 128
[tool.ruff.lint]
select = ["E", "W", "F", "I", "UP", "B", "C4", "RUF"]
# E — pycodestyle errors, F — pyflakes, W — warnings, I — isort
select = ["E", "F", "W", "I"]
# E501 — line too long (covered by line-length above)
ignore = ["E501"]
[tool.ruff.lint.isort]
known-first-party = ["app"]