houseassassin/Solo_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #9 from hteppl/main

Browse Source 
Changes from upstream
This commit is contained in:
Oleg Kraev
2025-01-17 21:51:08 +03:00
committed by GitHub
parent 466c2dcf31 381f3e0b67
commit 75d3102c79
3 changed files with 119 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backup.py
+3 -1
View File
@@ -4,7 +4,7 @@ from datetime import datetime
from aiogram.types import BufferedInputFile
from config import ADMIN_ID, BACK_DIR, DB_NAME, DB_PASSWORD, DB_USER, PG_HOST
from config import ADMIN_ID, BACK_DIR, DB_NAME, DB_PASSWORD, DB_USER, PG_HOST, PG_PORT
from logger import logger
@@ -43,6 +43,8 @@ def _create_database_backup() -> (str | None, Exception | None):
DB_USER,
"-h",
PG_HOST,
"-p",
PG_PORT,
"-F",
"c",
"-f",
handlers/keys/subscriptions.py
+52 -49
View File
@@ -10,6 +10,16 @@ from config import DATABASE_URL, PROJECT_NAME, SUB_MESSAGE, SUPERNODE, TRANSITIO
from database import get_servers_from_db
from logger import logger
# Глобальная переменная для пула соединений
db_pool = None
async def init_db_pool():
"""
Инициализация пула соединений, если он ещё не создан.
"""
global db_pool
if not db_pool:
db_pool = await asyncpg.create_pool(dsn=DATABASE_URL, min_size=5, max_size=20)
async def fetch_url_content(url, tg_id):
try:
@@ -33,44 +43,42 @@ async def fetch_url_content(url, tg_id):
logger.error(f"Ошибка при получении {url} для tg_id: {tg_id}: {e}")
return []
async def combine_unique_lines(urls, tg_id, query_string):
if SUPERNODE:
logger.info(f"Режим SUPERNODE активен. Возвращаем первую ссылку для tg_id: {tg_id}")
urls_with_query = [f"{urls[0]}?{query_string}"] if urls else []
return await fetch_url_content(urls_with_query[0], tg_id) if urls_with_query else []
if not urls:
return []
url_with_query = f"{urls[0]}?{query_string}" if query_string else urls[0]
return await fetch_url_content(url_with_query, tg_id)
all_lines = []
logger.info(
f"Начинаем объединение подписок для tg_id: {tg_id}, запрос: {query_string}"
)
urls_with_query = [f"{url}?{query_string}" for url in urls]
urls_with_query = [f"{url}?{query_string}" if query_string else url for url in urls]
logger.info(f"Составлены URL-адреса: {urls_with_query}")
for url in urls_with_query:
lines = await fetch_url_content(url, tg_id)
all_lines.extend(lines)
tasks = [fetch_url_content(url, tg_id) for url in urls_with_query]
results = await asyncio.gather(*tasks)
all_lines = set()
for lines in results:
all_lines.update(filter(None, lines))
all_lines = list(set(filter(None, all_lines)))
logger.info(
f"Объединено {len(all_lines)} строк после фильтрации и удаления дубликатов для tg_id: {tg_id}"
)
return all_lines
return list(all_lines)
transition_date = datetime.strptime(TRANSITION_DATE_STR, "%Y-%m-%d %H:%M:%S")
transition_timestamp_ms = int(transition_date.timestamp() * 1000)
transition_timestamp_ms_adjusted = transition_timestamp_ms - (3 * 60 * 60 * 1000)
logger.info(
f"Время перехода (с поправкой на часовой пояс): {transition_timestamp_ms_adjusted}"
)
async def handle_old_subscription(request):
email = request.match_info.get("email")
@@ -83,8 +91,10 @@ async def handle_old_subscription(request):
logger.info(f"Обработка запроса для старого клиента с email: {email}")
conn = await asyncpg.connect(DATABASE_URL)
try:
# Инициализируем пул соединений
await init_db_pool()
async with db_pool.acquire() as conn:
key_info = await conn.fetchrow(
"SELECT created_at, server_id FROM keys WHERE email = $1", email
)
@@ -121,35 +131,31 @@ async def handle_old_subscription(request):
status=400,
)
servers = await get_servers_from_db()
cluster_servers = servers.get(cluster_name, [])
logger.info(f"Сервера в кластере: {cluster_servers}")
servers = await get_servers_from_db()
cluster_servers = servers.get(cluster_name, [])
logger.info(f"Сервера в кластере: {cluster_servers}")
urls = []
for server in cluster_servers:
server_subscription_url = f"{server['subscription_url']}/{email}"
urls.append(server_subscription_url)
urls = [
f"{server['subscription_url']}/{email}" for server in cluster_servers
]
combined_subscriptions = await combine_unique_lines(urls, email, "")
combined_subscriptions = await combine_unique_lines(urls, email, "")
base64_encoded = base64.b64encode(
"\n".join(combined_subscriptions).encode("utf-8")
).decode("utf-8")
base64_encoded = base64.b64encode(
"\n".join(combined_subscriptions).encode("utf-8")
).decode("utf-8")
encoded_project_name = f"{PROJECT_NAME} - {SUB_MESSAGE}"
headers = {
"Content-Type": "text/plain; charset=utf-8",
"Content-Disposition": "inline",
"profile-update-interval": "7",
"profile-title": "base64:"
+ base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"),
}
encoded_project_name = f"{PROJECT_NAME} - {SUB_MESSAGE}"
headers = {
"Content-Type": "text/plain; charset=utf-8",
"Content-Disposition": "inline",
"profile-update-interval": "7",
"profile-title": "base64:"
+ base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"),
}
logger.info(f"Возвращаем объединенные подписки для email: {email}")
return web.Response(text=base64_encoded, headers=headers)
finally:
await conn.close()
logger.info(f"Возвращаем объединенные подписки для email: {email}")
return web.Response(text=base64_encoded, headers=headers)
async def handle_new_subscription(request):
@@ -165,9 +171,10 @@ async def handle_new_subscription(request):
logger.info(f"Обработка запроса для нового клиента: email={email}, tg_id={tg_id}")
conn = await asyncpg.connect(DATABASE_URL)
try:
# Инициализируем пул соединений
await init_db_pool()
async with db_pool.acquire() as conn:
client_data = await conn.fetchrow(
"SELECT tg_id, server_id FROM keys WHERE email = $1", email
)
@@ -189,16 +196,12 @@ async def handle_new_subscription(request):
status=403,
)
finally:
await conn.close()
servers = await get_servers_from_db()
cluster_servers = servers.get(cluster_name, [])
urls = []
for server in cluster_servers:
server_subscription_url = f"{server['subscription_url']}/{email}"
urls.append(server_subscription_url)
urls = [
f"{server['subscription_url']}/{email}" for server in cluster_servers
]
query_string = request.query_string
logger.info(f"Извлечен query string: {query_string}")
handlers/notifications.py
+64 -29
View File
@@ -1,5 +1,6 @@
import asyncio
from datetime import datetime, timedelta
import pytz
import asyncpg
from aiogram import Bot, Router, types
@@ -17,6 +18,7 @@ from config import (
RENEWAL_PLANS,
TOTAL_GB,
TRIAL_TIME,
EXPIRED_KEYS_CHECK_INTERVAL
)
from database import (
add_blocked_user,
@@ -56,6 +58,24 @@ async def check_users_and_update_blocked(bot: Bot):
await conn.close()
async def periodic_expired_keys_check(bot: Bot):
"""Периодическая проверка истекших ключей с кастомным интервалом."""
while True:
conn = None
try:
conn = await asyncpg.connect(DATABASE_URL)
current_time = int(datetime.utcnow().timestamp() * 1000)
await handle_expired_keys(bot, conn, current_time)
logger.info("✅ Проверка истекших ключей выполнена.")
except Exception as e:
logger.error(f"❌ Ошибка в periodic_expired_keys_check: {e}")
finally:
if conn:
await conn.close()
await asyncio.sleep(EXPIRED_KEYS_CHECK_INTERVAL)
async def notify_expiring_keys(bot: Bot):
conn = None
@@ -64,25 +84,19 @@ async def notify_expiring_keys(bot: Bot):
logger.info("Подключение к базе данных успешно.")
current_time = int(datetime.utcnow().timestamp() * 1000)
threshold_time_10h = int(
(datetime.utcnow() + timedelta(hours=10)).timestamp() * 1000
)
threshold_time_24h = int(
(datetime.utcnow() + timedelta(days=1)).timestamp() * 1000
)
threshold_time_10h = int((datetime.utcnow() + timedelta(hours=10)).timestamp() * 1000)
threshold_time_24h = int((datetime.utcnow() + timedelta(days=1)).timestamp() * 1000)
logger.info("Начало обработки уведомлений.")
await notify_inactive_trial_users(bot, conn)
await asyncio.sleep(1)
await asyncio.sleep(0.5)
await check_online_users()
await asyncio.sleep(1)
await asyncio.sleep(0.5)
await notify_10h_keys(bot, conn, current_time, threshold_time_10h)
await asyncio.sleep(1)
await asyncio.sleep(0.5)
await notify_24h_keys(bot, conn, current_time, threshold_time_24h)
await asyncio.sleep(1)
await handle_expired_keys(bot, conn, current_time)
await asyncio.sleep(1)
await asyncio.sleep(0.5)
except Exception as e:
logger.error(f"Ошибка при отправке уведомлений: {e}")
@@ -92,6 +106,7 @@ async def notify_expiring_keys(bot: Bot):
logger.info("Соединение с базой данных закрыто.")
async def is_bot_blocked(bot: Bot, chat_id: int) -> bool:
if DEV_MODE:
return False
@@ -137,8 +152,10 @@ async def process_10h_record(record, bot, conn):
email = record["email"]
expiry_time = record["expiry_time"]
expiry_date = datetime.utcfromtimestamp(expiry_time / 1000)
current_date = datetime.utcnow()
moscow_tz = pytz.timezone("Europe/Moscow")
expiry_date = datetime.fromtimestamp(expiry_time / 1000, tz=moscow_tz)
current_date = datetime.now(moscow_tz)
time_left = expiry_date - current_date
days_left_message = (
@@ -211,8 +228,10 @@ async def process_24h_record(record, bot, conn):
email = record["email"]
expiry_time = record["expiry_time"]
expiry_date = datetime.utcfromtimestamp(expiry_time / 1000)
current_date = datetime.utcnow()
moscow_tz = pytz.timezone("Europe/Moscow")
expiry_date = datetime.fromtimestamp(expiry_time / 1000, tz=moscow_tz)
current_date = datetime.now(moscow_tz)
time_left = expiry_date - current_date
days_left_message = (
@@ -368,13 +387,15 @@ async def process_key(record, bot, conn):
email = record["email"]
balance = await get_balance(tg_id)
expiry_time = record["expiry_time"]
expiry_date = datetime.utcfromtimestamp(expiry_time / 1000)
current_date = datetime.utcnow()
moscow_tz = pytz.timezone("Europe/Moscow")
expiry_date = datetime.fromtimestamp(expiry_time / 1000, tz=moscow_tz)
current_date = datetime.now(moscow_tz)
time_left = expiry_date - current_date
logger.info(
f"Время истечения ключа: {expiry_time} (UTC: {expiry_date}), "
f"Текущее время (UTC): {current_date}, "
f"Время истечения ключа: {expiry_time} (МСК: {expiry_date}), "
f"Текущее время (МСК): {current_date}, "
f"Оставшееся время: {time_left}"
)
@@ -391,7 +412,8 @@ async def process_key(record, bot, conn):
try:
if AUTO_RENEW_KEYS and balance >= RENEWAL_PLANS["1"]["price"]:
await update_balance(tg_id, -RENEWAL_PLANS["1"]["price"])
new_expiry_time = int((datetime.utcnow() + timedelta(days=30)).timestamp() * 1000)
new_expiry_time = int((datetime.now(moscow_tz) + timedelta(days=30)).timestamp() * 1000)
await update_key_expiry(client_id, new_expiry_time)
servers = await get_servers_from_db()
@@ -410,23 +432,35 @@ async def process_key(record, bot, conn):
)
logger.info(f"Флаги notified сброшены для клиента {client_id}.")
await bot.send_message(tg_id, text=KEY_RENEWED, reply_markup=keyboard)
logger.info(f"Уведомление об успешном продлении отправлено клиенту {tg_id}.")
try:
await bot.send_message(tg_id, text=KEY_RENEWED, reply_markup=keyboard)
logger.info(f"Уведомление об успешном продлении отправлено клиенту {tg_id}.")
except Exception as e:
logger.error(f"Не удалось отправить уведомление о продлении клиенту {tg_id}: {e}")
else:
message_expired = "Ваша подписка истекла. Пополните баланс для продления."
await bot.send_message(tg_id, text=message_expired, reply_markup=keyboard)
logger.info(f"Уведомление об истечении подписки отправлено пользователю {tg_id}.")
try:
await bot.send_message(tg_id, text=message_expired, reply_markup=keyboard)
logger.info(f"Уведомление об истечении подписки отправлено пользователю {tg_id}.")
except Exception as e:
logger.error(f"Не удалось отправить уведомление об истечении клиенту {tg_id}: {e}")
if AUTO_DELETE_EXPIRED_KEYS:
servers = await get_servers_from_db()
for cluster_id in servers:
await delete_key_from_cluster(cluster_id, email, client_id)
logger.info(f"Клиент {client_id} удален из кластера {cluster_id}.")
try:
await delete_key_from_cluster(cluster_id, email, client_id)
logger.info(f"Клиент {client_id} удален из кластера {cluster_id}.")
except Exception as e:
logger.error(f"Ошибка при удалении клиента {client_id} из кластера {cluster_id}: {e}")
await delete_key(client_id)
logger.info(f"Ключ {client_id} удалён из базы данных.")
try:
await delete_key(client_id)
logger.info(f"Ключ {client_id} удалён из базы данных.")
except Exception as e:
logger.error(f"Ошибка при удалении ключа {client_id} из базы данных: {e}")
else:
logger.info(f"Ключ {client_id} НЕ был удалён (AUTO_DELETE_EXPIRED_KEYS=False).")
@@ -434,6 +468,7 @@ async def process_key(record, bot, conn):
logger.error(f"Ошибка при обработке ключа для клиента {tg_id}: {e}")
async def check_online_users():
servers = await get_servers_from_db()