Lolz/AutoShop-Wednesday
Private
Public Access
4
0
Fork 1
forked from FOSS/AutoShop-Djimbo-Simple
Code Activity

feat: add Lolzteam payment integration

Browse Source
This commit is contained in:
Tom Riddle
2026-07-16 04:40:02 +05:00
parent cb56f6757e
commit 573dae70fb
13 changed files with 735 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tgbot/services/api_cryptobot.py
+69 -46
View File
@@ -13,20 +13,31 @@ from tgbot.utils.misc.bot_models import ARS
from tgbot.utils.misc_functions import send_admins
# Список поддерживаемых валют
ALLOW_CURRENCIES = ['BTC', 'ETH', 'LTC', 'USDT', 'USDC', 'TRX', 'TON', 'BNB', 'SOL', 'DOGE']
ALLOW_CURRENCIES = [
"BTC",
"ETH",
"LTC",
"USDT",
"USDC",
"TRX",
"TON",
"BNB",
"SOL",
"DOGE",
]
# АПИ для работы с CryptoBot
class CryptobotAPI:
# Настройка клиента CryptoBot
def __init__(
self,
bot: Bot,
arSession: ARS,
update: Optional[Union[Message, CallbackQuery]] = None,
token: str = "None",
adding: bool = False,
skipping_error: bool = False,
self,
bot: Bot,
arSession: ARS,
update: Optional[Union[Message, CallbackQuery]] = None,
token: str = "None",
adding: bool = False,
skipping_error: bool = False,
):
self.bot = bot
self.arSession = arSession
@@ -38,12 +49,12 @@ class CryptobotAPI:
# Инициализация данных
@classmethod
async def connect(
cls,
bot: Bot,
arSession: ARS,
update: Optional[Union[Message, CallbackQuery]] = None,
token: Optional[str] = None,
skipping_error: bool = False,
cls,
bot: Bot,
arSession: ARS,
update: Optional[Union[Message, CallbackQuery]] = None,
token: Optional[str] = None,
skipping_error: bool = False,
) -> "CryptobotAPI":
adding = token is not None
@@ -75,21 +86,23 @@ class CryptobotAPI:
await send_admins(
self.bot,
f"<b>🔷 CryptoBot недоступен. Как можно быстрее его замените</b>\n"
f"❗️ Ошибка: <code>{error_code}</code>"
f"❗️ Ошибка: <code>{error_code}</code>",
)
# Проверка кассы/кошелька
async def check(self) -> Tuple[bool, str]:
status, response = await self._request("getMe")
if status and response['ok']:
return True, ded(f"""
if status and response["ok"]:
return True, ded(
f"""
<b>🔷 CryptoBot кошелёк полностью функционирует ✅</b>
➖➖➖➖➖➖➖➖➖➖
▪️ Токен: <code>{self.token}</code>
▪️ Айди: <code>{response['result']['app_id']}</code>
▪️ Имя: <code>{response['result']['name']}</code>
""")
"""
)
return False, "<b>🔷 Не удалось проверить CryptoBot кошелёк ❌</b>"
@@ -97,17 +110,17 @@ class CryptobotAPI:
async def balance(self) -> str:
status, response = await self._request("getBalance")
if status and response['ok']:
if status and response["ok"]:
save_currencies = []
response_balances = sorted(
response['result'],
response["result"],
reverse=True,
key=lambda balance: to_number(balance['available']),
key=lambda balance: to_number(balance["available"]),
)
for currency in response_balances:
if currency['currency_code'] in ALLOW_CURRENCIES:
if currency["currency_code"] in ALLOW_CURRENCIES:
save_currencies.append(
f"▪️ {currency['currency_code']}: <code>{currency['available']}</code>"
)
@@ -123,24 +136,26 @@ class CryptobotAPI:
return "<b>🔷 Не удалось получить баланс CryptoBot кошелька ❌</b>"
# Создание счета на оплату
async def bill(self, pay_amount: Union[float, int]) -> Tuple[Union[str, bool], str, str]:
async def bill(
self, pay_amount: Union[float, int]
) -> Tuple[Union[str, bool], str, str]:
assets_currencies = ",".join(ALLOW_CURRENCIES)
payload = {
'currency_type': 'fiat',
'fiat': 'RUB',
'amount': str(pay_amount),
'expires_in': 10800,
'accepted_assets': assets_currencies
"currency_type": "fiat",
"fiat": "RUB",
"amount": str(pay_amount),
"expires_in": 10800,
"accepted_assets": assets_currencies,
}
status, response = await self._request("createInvoice", payload)
if status and response['ok']:
if status and response["ok"]:
bill_message = ded(f"""
<b>💰 Пополнение баланса</b>
➖➖➖➖➖➖➖➖➖➖
▪️ Для пополнения баланса, нажмите на кнопку ниже
▪️ Для пополнения баланса, нажмите на кнопку ниже
<code>Перейти к оплате</code> и оплатите выставленный вам счёт
▪️ У вас имеется 3 часа на оплату счета
▪️ Сумма пополнения: <code>{pay_amount}₽</code>
@@ -148,46 +163,52 @@ class CryptobotAPI:
❗️ После оплаты, нажмите на <code>Проверить оплату</code>
""")
return bill_message, response['result']['mini_app_invoice_url'], response['result']['invoice_id']
return (
bill_message,
response["result"]["mini_app_invoice_url"],
response["result"]["invoice_id"],
)
return False, "", ""
# Проверка счета на оплату
async def bill_check(self, bill_receipt: Optional[Union[str, int]] = None, records: int = 1) -> Tuple[int, float]:
async def bill_check(
self, bill_receipt: Optional[Union[str, int]] = None, records: int = 1
) -> Tuple[int, float]:
payload = {
'invoice_ids': f'{bill_receipt}',
'fiat': 'RUB',
"invoice_ids": f"{bill_receipt}",
"fiat": "RUB",
}
status, response = await self._request("getInvoices", payload)
pay_status, pay_amount = 1, 0
if status and response['ok']:
get_invoice = response['result']['items'][0]
if status and response["ok"]:
get_invoice = response["result"]["items"][0]
if get_invoice['status'] == "active":
if get_invoice["status"] == "active":
pay_status = 2
elif get_invoice['status'] == "expired":
elif get_invoice["status"] == "expired":
pay_status = 3
else:
pay_status = 0
pay_amount = to_number(get_invoice['amount'])
pay_amount = to_number(get_invoice["amount"])
return pay_status, pay_amount
# Генерация запроса
async def _request(
self,
method: str,
data: Optional[Dict[str, Any]] = None,
self,
method: str,
data: Optional[Dict[str, Any]] = None,
) -> Tuple[bool, Any]:
session = await self.arSession.get_session()
base_url = 'https://pay.crypt.bot/api/'
base_url = "https://pay.crypt.bot/api/"
headers = {
'Crypto-Pay-API-Token': self.token,
'Content-Type': 'application/x-www-form-urlencoded',
"Crypto-Pay-API-Token": self.token,
"Content-Type": "application/x-www-form-urlencoded",
}
url = base_url + method
@@ -204,7 +225,9 @@ class CryptobotAPI:
if response.status == 200:
return True, response_data
else:
await self.error_notification(f"{response.status} - {str(response_data)}")
await self.error_notification(
f"{response.status} - {str(response_data)}"
)
return False, response_data
except ClientConnectorCertificateError:
tgbot/services/api_lolzteam.py
+294
View File
@@ -0,0 +1,294 @@
# - *- coding: utf- 8 - *-
import json
from typing import Any, Dict, Optional, Tuple, Union
from aiogram import Bot
from aiogram.types import CallbackQuery, Message
from aiohttp import ClientConnectorCertificateError
from tgbot.database import Paymentsx
from tgbot.utils.const_functions import ded, to_number, get_unix, gen_id
from tgbot.utils.misc.bot_logging import bot_logger
from tgbot.utils.misc.bot_models import ARS
from tgbot.utils.misc_functions import send_admins
# Список поддерживаемых валют
ALLOW_CURRENCIES = [
"rub",
"uah",
"kzt",
"byn",
"usd",
"eur",
"gbp",
"cny",
"try",
"jpy",
"brl",
]
# АПИ для работы с Lolzteam
class LolzteamAPI:
# Настройка клиента Lolzteam
def __init__(
self,
bot: Bot,
arSession: ARS,
update: Optional[Union[Message, CallbackQuery]] = None,
token: str = "None",
merchant_id: int = "None",
adding: bool = False,
skipping_error: bool = False,
):
self.bot = bot
self.arSession = arSession
self.update = update
self.token = token
self.merchant_id = merchant_id
self.adding = adding
self.skipping_error = skipping_error
# Инициализация данных
@classmethod
async def connect(
cls,
bot: Bot,
arSession: ARS,
update: Optional[Union[Message, CallbackQuery]] = None,
token: Optional[str] = None,
merchant_id: Optional[int] = None,
skipping_error: bool = False,
) -> "LolzteamAPI":
adding = token is not None
if token is None or merchant_id is None:
payments = await Paymentsx().get()
adding = False
if token is None:
token = payments.lolzteam_token
if merchant_id is None:
merchant_id = payments.lolzteam_merchant_id
return cls(
bot=bot,
arSession=arSession,
update=update,
token=token,
merchant_id=merchant_id,
adding=adding,
skipping_error=skipping_error,
)
# Уведомления о неработоспособности кассы/кошелька
async def error_notification(self, error_code: str = "Unknown"):
bot_logger.warning("Lolzteam недоступен: %s", error_code)
if not self.skipping_error:
if self.adding and self.update is not None:
await self.update.edit_text(
f"<b>🟢 Не удалось добавить Lolzteam кассу ❌</b>\n"
f"❗️ Ошибка: <code>{error_code}</code>"
)
else:
await send_admins(
self.bot,
f"<b>🟢 Lolzteam недоступен. Как можно быстрее его замените</b>\n"
f"❗️ Ошибка: <code>{error_code}</code>",
)
# Проверка кассы/кошелька
async def check(self) -> Tuple[bool, str]:
status, response = await self._request("me", method="GET")
merchant_valid = False
if status and "errors" not in response and self.merchant_id:
balances = response["user"]["balances"]
for balance in balances:
if int(self.merchant_id) == balance["merchant_id"]:
merchant_valid = True
if merchant_valid:
return True, ded(
f"""
<b>🟢 Lolzteam полностью функционирует ✅</b>
➖➖➖➖➖➖➖➖➖➖
▪️ Токен: <code>{self.token}</code>
▪️ Айди: <code>{response['user']['user_id']}</code>
▪️ Айди Мерчанта: <code>{self.merchant_id}</code>
"""
)
elif status and "errors" not in response and not self.merchant_id:
return True, ded(
f"""
<b>🟢 Lolzteam полностью функционирует ✅</b>
➖➖➖➖➖➖➖➖➖➖
▪️ Токен: <code>{self.token}</code>
▪️ Айди: <code>{response['user']['user_id']}</code>
"""
)
return False, "<b>🟢 Не удалось проверить Lolzteam ❌</b>"
# Получение баланса
async def balance(self) -> str:
status, response = await self._request("me", method="GET")
if status and "errors" not in response:
merchants = []
response_balances = sorted(
response["user"]["balances"],
reverse=True,
key=lambda balance: to_number(balance["balance"]),
)
for balance in response_balances:
if balance["merchant_id"] == self.merchant_id:
title = balance["title"]
custom_title = balance["custom_title"]
merchants.append(
f"▪️ {title if not custom_title else custom_title}: <code>{balance['balance']} ₽</code>"
)
merchants = "\n".join(merchants)
return ded(f"""
<b>🟢 Баланс Lolzteam составляет</b>
➖➖➖➖➖➖➖➖➖➖
{merchants}
""")
return "<b>🟢 Не удалось получить баланс Lolzteam ❌</b>"
# Создание счета на оплату
async def bill(
self, pay_amount: Union[float, int]
) -> Tuple[Union[str, bool], str, str]:
pro_payment_id = get_unix()
eu_payment_id = gen_id(12)
payment_id = str(pro_payment_id) + str(eu_payment_id)
bot_data = await self.bot.get_me()
bot_username = bot_data.username
bot_url = f"https://telegram.me/{bot_username}"
payload = {
"currency": "rub",
"amount": pay_amount,
"payment_id": payment_id,
"comment": f"Пополнение #{payment_id}",
"url_success": bot_url,
"merchant_id": self.merchant_id,
"lifetime": 10800,
"is_test": True, # Тестовый режим для отладки
}
status, response = await self._request("invoice", payload)
if status and "errors" not in response:
bill_message = ded(f"""
<b>💰 Пополнение баланса</b>
➖➖➖➖➖➖➖➖➖➖
▪️ Для пополнения баланса, нажмите на кнопку ниже
<code>Перейти к оплате</code> и оплатите выставленный вам счёт
▪️ У вас имеется 3 часа на оплату счета
▪️ Сумма пополнения: <code>{pay_amount}₽</code>
➖➖➖➖➖➖➖➖➖➖
❗️ После оплаты, нажмите на <code>Проверить оплату</code>
""")
return (
bill_message,
response["invoice"]["url"],
response["invoice"]["invoice_id"],
)
return False, "", ""
# Проверка счета на оплату
async def bill_check(
self, bill_receipt: Optional[Union[str, int]] = None, records: int = 1
) -> Tuple[int, float]:
payload = {"invoice_id": int(bill_receipt)}
status, response = await self._request("invoice", payload, method="GET")
pay_status, pay_amount = 1, 0
if status and "errors" not in response:
get_invoice = response["invoice"]
status = get_invoice["status"]
time = response["system_info"]["time"]
expires_at = get_invoice["expires_at"]
# not_paid + time < expires_at
if (status == "not_paid") and (time < expires_at):
pay_status = 2
# not_paid + time > expires_at
elif (status == "expired") and (time > expires_at):
pay_status = 3
# paid
else:
pay_status = 0
pay_amount = to_number(get_invoice["amount"])
return pay_status, pay_amount
# Генерация запроса
async def _request(
self,
path: str,
data: Optional[Dict[str, Any]] = None,
method: str = "POST",
) -> Tuple[bool, Any]:
session = await self.arSession.get_session()
method = method.upper()
base_url = "https://prod-api.lzt.market/"
headers = {
"Authorization": f"Bearer {self.token}",
"Accept": "application/json",
}
url = base_url + path
try:
if method == "POST":
response = await session.post(url=url, headers=headers, data=data)
elif method == "GET":
response = await session.get(url=url, headers=headers, params=data)
response_data = json.loads((await response.read()).decode())
if response.status == 200:
return True, response_data
else:
await self.error_notification(
f"{response.status} - {str(response_data)}"
)
return False, response_data
except ClientConnectorCertificateError:
bot_logger.warning("Ошибка SSL при запросе Lolzteam", exc_info=True)
await self.error_notification("CERTIFICATE_VERIFY_FAILED")
return False, "CERTIFICATE_VERIFY_FAILED"
except Exception as ex:
bot_logger.warning("Ошибка запроса Lolzteam", exc_info=True)
await self.error_notification(str(ex))
return False, str(ex)