Private
Public Access
forked from FOSS/AutoShop-Djimbo-Simple
304 lines
11 KiB
Python
304 lines
11 KiB
Python
# - *- 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")
|
|
|
|
if status and "errors" not in response and self.merchant_id == "None":
|
|
return True, ded(
|
|
f"""
|
|
<b>🟢 Lolzteam функционирует ✅</b>
|
|
➖➖➖➖➖➖➖➖➖➖
|
|
▪️ Токен: <code>{self.token}</code>
|
|
▪️ Айди: <code>{response['user']['user_id']}</code>
|
|
"""
|
|
)
|
|
|
|
elif status and "errors" not in response:
|
|
# ! Проверка этим способом не работает для мерчантов, у которых нет отдельного
|
|
# ! баланса на маркете, и поэтому используется способ ниже
|
|
# balances = response["user"]["balances"]
|
|
|
|
# for balance in balances:
|
|
# if int(self.merchant_id) == balance["merchant_id"]:
|
|
# merchant_valid = True
|
|
|
|
# ! Создание оплаты и проверка возвращаемого статуса (если статус положительный - то и мерчант валидный)
|
|
bill_message, bill_link, bill_receipt = await self.bill(
|
|
1, is_test=True
|
|
) # Тестовый инвойс
|
|
|
|
if bill_message:
|
|
return True, ded(
|
|
f"""
|
|
<b>🟢 Lolzteam полностью функционирует ✅</b>
|
|
➖➖➖➖➖➖➖➖➖➖
|
|
▪️ Токен: <code>{self.token}</code>
|
|
▪️ Айди: <code>{response['user']['user_id']}</code>
|
|
▪️ Айди Мерчанта: <code>{self.merchant_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],
|
|
is_test: bool = False,
|
|
) -> 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": is_test, # Тестовый режим для отладки
|
|
}
|
|
|
|
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",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
url = base_url + path
|
|
|
|
try:
|
|
if method == "POST":
|
|
response = await session.post(url=url, headers=headers, json=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)
|