205 lines
8.7 KiB
Python
205 lines
8.7 KiB
Python
import os
|
||
from typing import Any
|
||
import uuid
|
||
from requests.auth import HTTPBasicAuth
|
||
from yookassa import Configuration, Payment
|
||
import aiohttp
|
||
import mysql.connector
|
||
import requests
|
||
from dotenv import load_dotenv
|
||
|
||
|
||
load_dotenv()
|
||
|
||
|
||
def get_contract_ret_tron(hash_user: str, amount: int) -> bool:
|
||
""" URL для запроса инфо о транзакции tron """
|
||
# GET запрос к API
|
||
response = requests.get(f"https://apilist.tronscanapi.com/api/transaction-info?hash={hash_user}")
|
||
response.raise_for_status()
|
||
data = response.json() # Получаем ответ в джисон
|
||
if 'trc20TransferInfo' not in data or len(data['trc20TransferInfo']) == 0:
|
||
return False # если нет ключа или он 0, прерываем функцию
|
||
amount_hash = amount * 1000000 # реальную сумму надо умножить. так указано в джон ответах
|
||
transfer_info = data['trc20TransferInfo'][0]
|
||
if transfer_info.get('to_address') == os.getenv('TO_ADRESS'):
|
||
if transfer_info.get('amount_str') == str(amount_hash):
|
||
return data.get('contractRet') == "SUCCESS"
|
||
return False
|
||
|
||
|
||
def get_contract_ret_arbitrum(hash_user: str, amount: int) -> bool:
|
||
""" URL для запроса инфо о транзакции Arbitrum """
|
||
url = "https://api.arbiscan.io/api"
|
||
params = {
|
||
'module': 'proxy',
|
||
'action': 'eth_getTransactionReceipt',
|
||
'txhash': hash_user,
|
||
'apikey': os.getenv('ARBITRUM_API_KEY')}
|
||
response = requests.get(url, params=params)
|
||
data = response.json() # Получаем ответ в джисон
|
||
if 'result' not in data or data['result'] is None:
|
||
return False # Прерывание функции, если result нет или пустой
|
||
if 'logs' not in data['result'] or len(data['result']['logs']) == 0:
|
||
return False # Прерывание функции, если logs нет или пустая
|
||
amount_hash = amount * 1000000 # реальную сумму надо умножить. так указано в джон ответах
|
||
# в джисон ответе данные в hex формате. ниже идет логика перевода в читабельный формат для проверок
|
||
hex_data = data['result']['logs'][0]['data']
|
||
hex_address = data['result']['logs'][0]['topics'][2]
|
||
hex_status = data['result']['status']
|
||
decimal_data = int(hex_data, 16)
|
||
decimal_status = int(hex_status, 16)
|
||
address = hex_address[-40:]
|
||
if decimal_data == amount_hash:
|
||
if address == os.getenv('ARBITRUM_ADRESS'):
|
||
return decimal_status == 1
|
||
return False
|
||
|
||
|
||
def create_connection() -> Any:
|
||
""" подключения к базе данных mysql """
|
||
connection = mysql.connector.connect(
|
||
host=os.getenv('DB_HOST'),
|
||
user=os.getenv('DB_USER'),
|
||
password=os.getenv('DB_PASSWORD'),
|
||
database=os.getenv('DB_DATABASE'),)
|
||
return connection
|
||
|
||
|
||
async def get_usd_rub_course() -> None:
|
||
"""Получает курс USD/RUB с openexchangerates.org"""
|
||
from database_queries import adding_well
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(
|
||
"https://openexchangerates.org/api/latest.json?"
|
||
"app_id=488b07ebd6964a71b838ea93fbebcd31&base=USD&symbols=RUB") as response:
|
||
data = await response.json()
|
||
usd_rub_course = round(data["rates"]["RUB"], 2)
|
||
conn = create_connection()
|
||
cursor = conn.cursor()
|
||
adding_well(cursor, usd_rub_course)
|
||
conn.commit()
|
||
|
||
|
||
async def payment_order_yukassa(price_rub: int, bot_username: str) -> dict:
|
||
""" Формирование платежки в юкассе. Формат вывода - словарь"""
|
||
Configuration.account_id = os.getenv('ACCOUNT_ID_YK')
|
||
Configuration.secret_key = os.getenv('SECRET_KEY_YK')
|
||
payment = Payment.create({
|
||
"amount": {
|
||
"value": price_rub,
|
||
"currency": "RUB"},
|
||
"confirmation": {
|
||
"type": "redirect",
|
||
"return_url": f"https://t.me/{bot_username}"},
|
||
"capture": True,
|
||
"description": "Оплата подписки"}, uuid.uuid4())
|
||
return payment
|
||
|
||
|
||
def check_payment_status(payment_id: str) -> str:
|
||
""" Проверка статуса платежа по айди платежки """
|
||
shop_id = os.getenv('ACCOUNT_ID_YK')
|
||
secret_key = os.getenv('SECRET_KEY_YK')
|
||
url = f"https://api.yookassa.ru/v3/payments/{payment_id}"
|
||
headers = {"Content-Type": "application/json"}
|
||
response = requests.get(url, auth=HTTPBasicAuth(shop_id, secret_key), headers=headers) # сам запрос
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return data['status']
|
||
else:
|
||
raise Exception(f"Ошибка при проверке статуса платежа: {response.status_code}, {response.text}")
|
||
|
||
|
||
async def checking_and_deleting_users() -> None:
|
||
""" Получает список id пользователй и бот удаляет их из чата, если они там есть """
|
||
from database_queries import check_expired_subscriptions, reset_user_data
|
||
from main import bot_chat
|
||
list_user_in_table = check_expired_subscriptions()
|
||
print(list_user_in_table)
|
||
for id_user in list_user_in_table:
|
||
try:
|
||
await bot_chat.ban_chat_member(-4226821992, id_user)
|
||
reset_user_data(id_user)
|
||
except Exception as e:
|
||
print(f"Ошибка: {e}")
|
||
|
||
|
||
def check_table_exists_users():
|
||
""" Проверяет, существует ли таблица users в базе данных. Если таблицы нет, создает ее """
|
||
conn = create_connection()
|
||
cursor = conn.cursor()
|
||
# Запрос для проверки существования таблицы
|
||
cursor.execute(f"""
|
||
SELECT 1
|
||
FROM information_schema.tables
|
||
WHERE table_schema = '{os.getenv('DB_DATABASE')}'
|
||
AND table_name = 'users';
|
||
""")
|
||
exists = cursor.fetchone() is not None
|
||
# Если таблица не существует, создаем ее
|
||
if not exists:
|
||
cursor.execute(f"""
|
||
CREATE TABLE users (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
telegram_id BIGINT NOT NULL,
|
||
telegram_username VARCHAR(255) NOT NULL,
|
||
zennolab_id VARCHAR(255) DEFAULT NULL,
|
||
sub_datetime_end BIGINT UNSIGNED DEFAULT NULL,
|
||
threads_count INT DEFAULT NULL,
|
||
telegram_thread_id BIGINT DEFAULT NULL,
|
||
question_time INT DEFAULT NULL,
|
||
count_streams INT DEFAULT NULL,
|
||
subscription_option VARCHAR(255) DEFAULT NULL,
|
||
subscription_end_date DATE DEFAULT NULL);""")
|
||
conn.commit()
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def check_table_exists_usd_rub():
|
||
""" Проверяет, существует ли таблица USD_RUB в базе данных. Если таблицы нет, создает ее """
|
||
conn = create_connection()
|
||
cursor = conn.cursor()
|
||
# Запрос для проверки существования таблицы
|
||
cursor.execute(f"""
|
||
SELECT 1
|
||
FROM information_schema.tables
|
||
WHERE table_schema = '{os.getenv('DB_DATABASE')}'
|
||
AND table_name = 'USD_RUB';
|
||
""")
|
||
exists = cursor.fetchone() is not None
|
||
# Если таблица не существует, создаем ее
|
||
if not exists:
|
||
cursor.execute(f"""
|
||
CREATE TABLE USD_RUB (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
well FLOAT DEFAULT 0);""")
|
||
conn.commit()
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def check_table_exists_successful_hash():
|
||
""" Проверяет, существует ли таблица successful_hash в базе данных. Если таблицы нет, создает ее """
|
||
conn = create_connection()
|
||
cursor = conn.cursor()
|
||
# Запрос для проверки существования таблицы
|
||
cursor.execute(f"""
|
||
SELECT 1
|
||
FROM information_schema.tables
|
||
WHERE table_schema = '{os.getenv('DB_DATABASE')}'
|
||
AND table_name = 'successful_hash';
|
||
""")
|
||
exists = cursor.fetchone() is not None
|
||
# Если таблица не существует, создаем ее
|
||
if not exists:
|
||
cursor.execute(f"""
|
||
CREATE TABLE successful_hash (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
hash_users VARCHAR(255) NOT NULL,
|
||
telegram_id BIGINT NOT NULL);""")
|
||
conn.commit()
|
||
cursor.close()
|
||
conn.close()
|