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

Merge pull request #198 from Capybara-z/patch-2

Browse Source 
Fix notification
This commit is contained in:
Vladislav Lisitsyn
2025-05-07 20:42:32 +03:00
committed by GitHub
parent 9cf30b3953 811f7276c8
commit b027d329dd
4 changed files with 285 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handlers/keys/subscriptions.py
+79 -47
View File
@@ -12,10 +12,10 @@ from aiohttp import web
from config import (
DATABASE_URL,
RANDOM_SUBSCRIPTIONS,
PROJECT_NAME,
SUPERNODE,
SUPPORT_CHAT_URL,
TOTAL_GB,
USERNAME_BOT,
USE_COUNTRY_SELECTION,
)
@@ -24,33 +24,43 @@ from handlers.utils import convert_to_bytes
from logger import logger
async def fetch_url_content(url: str, identifier: str) -> list[str]:
async def fetch_url_content(url: str, identifier: str) -> tuple[list[str], dict[str, str]]:
try:
timeout = aiohttp.ClientTimeout(total=5)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url, ssl=False) as response:
if response.status == 200:
content = await response.text()
return base64.b64decode(content).decode("utf-8").split("\n")
return []
except Exception:
return []
lines = base64.b64decode(content).decode("utf-8").split("\n")
headers = {k.lower(): v for k, v in response.headers.items()}
logger.debug(f"Fetched {url}: {len(lines)} lines, headers: {headers}")
return lines, headers
return [], {}
except Exception as e:
logger.error(f"Error fetching URL {url}: {e}")
return [], {}
async def combine_unique_lines(urls: list[str], identifier: str, query_string: str) -> list[str]:
async def combine_unique_lines(urls: list[str], identifier: str, query_string: str) -> tuple[list[str], list[dict[str, str]]]:
if SUPERNODE:
logger.info(f"Режим SUPERNODE активен. Возвращаем первую ссылку для идентификатора: {identifier}")
if not urls:
return []
return [], []
url_with_query = f"{urls[0]}?{query_string}" if query_string else urls[0]
return await fetch_url_content(url_with_query, identifier)
lines, headers = await fetch_url_content(url_with_query, identifier)
return lines, [headers]
urls_with_query = [f"{url}?{query_string}" if query_string else url for url in urls]
tasks = [fetch_url_content(url, identifier) for url in urls_with_query]
results = await asyncio.gather(*tasks, return_exceptions=True)
all_lines = set()
for lines in results:
all_lines.update(filter(None, lines))
return list(all_lines)
all_headers = []
for result in results:
if isinstance(result, tuple):
lines, headers = result
all_lines.update(filter(None, lines))
all_headers.append(headers)
return list(all_lines), all_headers
async def get_subscription_urls(server_id: str, email: str, conn, include_remnawave_key: str = None) -> list[str]:
@@ -72,38 +82,59 @@ async def get_subscription_urls(server_id: str, email: str, conn, include_remnaw
return urls
def calculate_traffic(cleaned_subscriptions: list[str], expiry_time_ms: int | None) -> str:
def calculate_traffic(cleaned_subscriptions: list[str], expiry_time_ms: int | None, headers_list: list[dict[str, str]]) -> str:
logger.debug(f"Calculating traffic with subscriptions: {cleaned_subscriptions}, headers: {headers_list}")
expire_timestamp = int(expiry_time_ms / 1000) if expiry_time_ms else 0
if TOTAL_GB != 0:
country_remaining = {}
for line in cleaned_subscriptions:
if "#" not in line:
continue
try:
_, meta = line.split("#", 1)
except ValueError:
continue
parts = meta.split("-")
country = parts[0].strip()
remaining_str = parts[1].strip() if len(parts) == 2 else ""
if remaining_str:
remaining_str = remaining_str.replace(",", ".")
m_total = re.search(r"([\d\.]+)\s*([GMKTB]B)", remaining_str, re.IGNORECASE)
if m_total:
value = float(m_total.group(1))
unit = m_total.group(2).upper()
remaining_bytes = convert_to_bytes(value, unit)
country_remaining[country] = remaining_bytes
num_countries = len(country_remaining)
issued_per_country = TOTAL_GB
total_traffic_bytes = issued_per_country * num_countries
consumed_traffic_bytes = total_traffic_bytes - sum(country_remaining.values())
if consumed_traffic_bytes < 0:
consumed_traffic_bytes = 0
else:
consumed_traffic_bytes = 1
total_traffic_bytes = 0
return f"upload=0; download={consumed_traffic_bytes}; total={total_traffic_bytes}; expire={expire_timestamp}"
upload = 0
download = 0
total = 0
for headers in headers_list:
userinfo = headers.get("subscription-userinfo", "")
if userinfo:
parts = userinfo.split(";")
for part in parts:
part = part.strip()
if part.startswith("upload="):
upload += int(part.split("=")[1])
elif part.startswith("download="):
download += int(part.split("=")[1])
elif part.startswith("total="):
total += int(part.split("=")[1])
logger.debug(f"Processed Subscription-Userinfo: {userinfo}")
country_remaining = {}
for line in cleaned_subscriptions:
if "#" not in line:
continue
try:
_, meta = line.split("#", 1)
except ValueError:
continue
parts = meta.split("-")
country = parts[0].strip()
remaining_str = parts[1].strip() if len(parts) == 2 else ""
if remaining_str:
remaining_str = remaining_str.replace(",", ".")
m_total = re.search(r"([\d\.]+)\s*([GMKTB]B)", remaining_str, re.IGNORECASE)
if m_total:
value = float(m_total.group(1))
unit = m_total.group(2).upper()
remaining_bytes = convert_to_bytes(value, unit)
country_remaining[country] = remaining_bytes
logger.debug(f"Found traffic: {value}{unit} for {country}")
consumed_traffic_bytes = total - sum(country_remaining.values()) if country_remaining else download
if consumed_traffic_bytes < 0:
consumed_traffic_bytes = 0
download = max(download, consumed_traffic_bytes)
if download == 0 and total == 0 and not country_remaining:
download = 1
result = f"upload={upload}; download={download}; total={total}; expire={expire_timestamp}"
logger.debug(f"Calculated subscription-userinfo: {result}")
return result
def clean_subscription_line(line: str) -> str:
@@ -204,15 +235,16 @@ async def handle_subscription(request: web.Request) -> web.Response:
time_left = format_time_left(expiry_time_ms)
urls = await get_subscription_urls(
server_id, email, conn, include_remnawave_key=client_data.get("remnawave_link")
server_id, email, conn, include_remnawave_key=client_data.get("remnawave_key")
)
if not urls:
return web.Response(text="❌ Сервер не найден.", status=404)
query_string = request.query_string
combined_subscriptions = await combine_unique_lines(urls, tg_id or email, query_string)
random.shuffle(combined_subscriptions)
combined_subscriptions, headers_list = await combine_unique_lines(urls, tg_id or email, query_string)
if RANDOM_SUBSCRIPTIONS:
random.shuffle(combined_subscriptions)
cleaned_subscriptions = [clean_subscription_line(line) for line in combined_subscriptions]
@@ -220,7 +252,7 @@ async def handle_subscription(request: web.Request) -> web.Response:
subscription_info = f"📄 Подписка: {email} - {time_left}"
user_agent = request.headers.get("User-Agent", "")
subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms)
subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms, headers_list)
headers = prepare_headers(user_agent, PROJECT_NAME, subscription_info, subscription_userinfo)
return web.Response(text=base64_encoded, headers=headers)
handlers/notifications/general_notifications.py
+77 -30
View File
@@ -1,11 +1,8 @@
import asyncio
from datetime import datetime, timedelta
import asyncpg
import pytz
from aiogram import Bot, Router
import pytz
from config import (
DATABASE_URL,
@@ -33,7 +30,10 @@ from database import (
update_key_expiry,
)
from handlers.keys.key_utils import delete_key_from_cluster, renew_key_in_cluster
from handlers.notifications.notify_kb import build_notification_expired_kb, build_notification_kb
from handlers.notifications.notify_kb import (
build_notification_expired_kb,
build_notification_kb,
)
from handlers.texts import (
KEY_DELETED_MSG,
KEY_EXPIRED_DELAY_HOURS_MINUTES_MSG,
@@ -47,17 +47,13 @@ from handlers.texts import (
)
from handlers.utils import format_hours, format_minutes
from logger import logger
from handlers.utils import format_hours, format_months, format_minutes
from .notify_utils import send_messages_with_limit, send_notification
from .special_notifications import notify_inactive_trial_users, notify_users_no_traffic
router = Router()
moscow_tz = pytz.timezone("Europe/Moscow")
notification_lock = asyncio.Lock()
@@ -77,7 +73,6 @@ async def periodic_notifications(bot: Bot):
try:
conn = await asyncpg.connect(DATABASE_URL)
current_time = int(datetime.now(moscow_tz).timestamp() * 1000)
threshold_time_10h = int((datetime.now(moscow_tz) + timedelta(hours=10)).timestamp() * 1000)
threshold_time_24h = int((datetime.now(moscow_tz) + timedelta(days=1)).timestamp() * 1000)
@@ -91,17 +86,35 @@ async def periodic_notifications(bot: Bot):
keys = []
if not TRIAL_TIME_DISABLE:
await notify_inactive_trial_users(bot, conn)
try:
await notify_inactive_trial_users(bot, conn)
except Exception as e:
logger.error(f"Ошибка в notify_inactive_trial_users: {e}")
await asyncio.sleep(0.5)
await notify_24h_keys(bot, conn, current_time, threshold_time_24h, keys)
try:
await notify_24h_keys(bot, conn, current_time, threshold_time_24h, keys)
except Exception as e:
logger.error(f"Ошибка в notify_24h_keys: {e}")
await asyncio.sleep(1)
await notify_10h_keys(bot, conn, current_time, threshold_time_10h, keys)
try:
await notify_10h_keys(bot, conn, current_time, threshold_time_10h, keys)
except Exception as e:
logger.error(f"Ошибка в notify_10h_keys: {e}")
await asyncio.sleep(1)
await handle_expired_keys(bot, conn, current_time, keys)
try:
await handle_expired_keys(bot, conn, current_time, keys)
except Exception as e:
logger.error(f"Ошибка в handle_expired_keys: {e}")
await asyncio.sleep(0.5)
if NOTIFY_INACTIVE_TRAFFIC:
await notify_users_no_traffic(bot, conn, current_time, keys)
try:
await notify_users_no_traffic(bot, conn, current_time, keys)
except Exception as e:
logger.error(f"Ошибка в notify_users_no_traffic: {e}")
await asyncio.sleep(0.5)
logger.info("Завершена обработка уведомлений")
@@ -162,7 +175,11 @@ async def notify_24h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int,
)
if NOTIFY_RENEW:
await process_auto_renew_or_notify(bot, conn, key, notification_id, 1, "notify_24h.jpg", notification_text)
try:
await process_auto_renew_or_notify(bot, conn, key, notification_id, 1, "notify_24h.jpg", notification_text)
except Exception as e:
logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {e}")
continue
else:
keyboard = build_notification_kb(email)
messages.append({
@@ -170,12 +187,21 @@ async def notify_24h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int,
"text": notification_text,
"photo": "notify_24h.jpg",
"keyboard": keyboard,
"notification_id": notification_id,
})
await add_notification(tg_id, notification_id, session=conn)
if messages:
await send_messages_with_limit(bot, messages)
logger.info(f"Отправлено {len(messages)} уведомлений об истечении подписки через 24 часа.")
results = await send_messages_with_limit(bot, messages, conn=conn)
sent_count = 0
for msg, result in zip(messages, results):
tg_id = msg["tg_id"]
if result:
await add_notification(tg_id, msg["notification_id"], session=conn)
sent_count += 1
logger.info(f"📢 Отправлено уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.")
else:
logger.warning(f"📢 Не удалось отправить уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.")
logger.info(f"Отправлено {sent_count} уведомлений об истечении подписки через 24 часа.")
logger.info("Обработка всех уведомлений за 24 часа завершена.")
await asyncio.sleep(1)
@@ -233,6 +259,7 @@ async def notify_10h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int,
)
except Exception as e:
logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {e}")
continue
else:
keyboard = build_notification_kb(email)
messages.append({
@@ -240,12 +267,21 @@ async def notify_10h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int,
"text": notification_text,
"photo": "notify_10h.jpg",
"keyboard": keyboard,
"notification_id": notification_id,
})
await add_notification(tg_id, notification_id, session=conn)
if messages:
await send_messages_with_limit(bot, messages)
logger.info(f"Отправлено {len(messages)} уведомлений об истечении подписки через 10 часов.")
results = await send_messages_with_limit(bot, messages, conn=conn)
sent_count = 0
for msg, result in zip(messages, results):
tg_id = msg["tg_id"]
if result:
await add_notification(tg_id, msg["notification_id"], session=conn)
sent_count += 1
logger.info(f"📢 Отправлено уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.")
else:
logger.warning(f"📢 Не удалось отправить уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.")
logger.info(f"Отправлено {sent_count} уведомлений об истечении подписки через 10 часов.")
logger.info("Обработка всех уведомлений за 10 часов завершена.")
await asyncio.sleep(1)
@@ -317,6 +353,8 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time:
"text": KEY_DELETED_MSG.format(email=email),
"photo": "notify_expired.jpg",
"keyboard": keyboard,
"notification_id": notification_id,
"email": email,
})
except Exception as e:
logger.error(f"Ошибка удаления ключа {client_id} для пользователя {tg_id}: {e}")
@@ -350,12 +388,23 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time:
"text": delay_message,
"photo": "notify_expired.jpg",
"keyboard": keyboard,
"notification_id": notification_id,
"email": email,
})
await add_notification(tg_id, notification_id, session=conn)
if messages:
await send_messages_with_limit(bot, messages)
logger.info(f"Отправлено {len(messages)} уведомлений об истекших ключах.")
results = await send_messages_with_limit(bot, messages, conn=conn)
sent_count = 0
for msg, result in zip(messages, results):
tg_id = msg["tg_id"]
email = msg["email"]
if result:
await add_notification(tg_id, msg["notification_id"], session=conn)
sent_count += 1
logger.info(f"📢 Отправлено уведомление об истекшем ключе для подписки {email} пользователю {tg_id}.")
else:
logger.warning(f"📢 Не удалось отправить уведомление об истекшем ключе для подписки {email} пользователю {tg_id}.")
logger.info(f"Отправлено {sent_count} уведомлений об истекших ключах.")
logger.info("Обработка истекших ключей завершена.")
await asyncio.sleep(1)
@@ -404,7 +453,7 @@ async def process_auto_renew_or_notify(
total_gb = int(renewal_period_months * TOTAL_GB * 1024**3)
logger.info(
f"[Автопродление] Продление подписки {email} на {renewal_period_months} мес. для пользователя {tg_id}. Баланс: {balance}, списываем: {renewal_cost}"
f"Продление подписки {email} на {renewal_period_months} мес. для пользователя {tg_id}. Баланс: {balance}, списываем: {renewal_cost}"
)
try:
@@ -420,9 +469,7 @@ async def process_auto_renew_or_notify(
)
renewed_message = KEY_RENEWED.format(
email=email,
months_formatted=months_formatted,
expiry_date=formatted_expiry_date
email=email, months=renewal_period_months, expiry_date=formatted_expiry_date
)
keyboard = build_notification_expired_kb()
@@ -439,9 +486,9 @@ async def process_auto_renew_or_notify(
logger.error(f"❌ Ошибка при продлении ключа {client_id} для пользователя {tg_id}: {e}")
else:
keyboard = build_notification_kb(email)
await add_notification(tg_id, notification_id, session=conn)
result = await send_notification(bot, tg_id, standard_photo, standard_caption, keyboard)
if result:
logger.info(f"📢 Отправлено уведомление об истекающей подписке {email} пользователю {tg_id}.")
await add_notification(tg_id, notification_id, session=conn)
else:
logger.warning(f"📢 Не удалось отправить уведомление об истекающей подписке {email} пользователю {tg_id}.")
handlers/notifications/notify_utils.py
+61 -14
View File
@@ -2,29 +2,75 @@ import asyncio
import os
import aiofiles
import asyncpg
from aiogram import Bot
from aiogram.exceptions import TelegramForbiddenError, TelegramRetryAfter
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter
from aiogram.types import BufferedInputFile, InlineKeyboardMarkup
from database import create_blocked_user
from logger import logger
async def send_messages_with_limit(bot: Bot, messages: list[dict], messages_per_second: int = 25):
async def send_messages_with_limit(
bot: Bot,
messages: list[dict],
conn: asyncpg.Connection = None,
source_file: str = None,
messages_per_second: int = 25
):
"""
Отправляет сообщения с ограничением по количеству сообщений в секунду.
Возвращает список результатов отправки (True для успеха, False для ошибки).
"""
batch_size = messages_per_second
results = []
for i in range(0, len(messages), batch_size):
batch = messages[i : i + batch_size]
tasks = []
for msg in batch:
tasks.append(send_notification(bot, msg["tg_id"], msg.get("photo"), msg["text"], msg.get("keyboard")))
try:
await asyncio.gather(*tasks, return_exceptions=True)
except Exception as e:
logger.error(f"⚠ Ошибка при отправке сообщений в батче: {e}")
tasks.append(send_notification(
bot,
msg["tg_id"],
msg.get("photo"),
msg["text"],
msg.get("keyboard")
))
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
processed_results = []
for msg, result in zip(batch, batch_results):
tg_id = msg["tg_id"]
if isinstance(result, bool) and result:
processed_results.append(True)
elif isinstance(result, TelegramForbiddenError):
logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.")
if source_file == "special_notifications" and conn:
try:
await create_blocked_user(tg_id, conn)
logger.info(f"Пользователь {tg_id} добавлен в blocked_users.")
except Exception:
pass
processed_results.append(False)
elif isinstance(result, TelegramBadRequest) and "chat not found" in str(result).lower():
logger.warning(f"🚫 Чат не найден для пользователя {tg_id}.")
if source_file == "special_notifications" and conn:
try:
await create_blocked_user(tg_id, conn)
logger.info(f"Пользователь {tg_id} добавлен в blocked_users.")
except Exception:
pass
processed_results.append(False)
else:
logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id}.")
if source_file == "special_notifications" and conn:
try:
await create_blocked_user(tg_id, conn)
logger.info(f"Пользователь {tg_id} добавлен в blocked_users.")
except Exception:
pass
processed_results.append(False)
results.extend(processed_results)
await asyncio.sleep(1.0)
return results
def rate_limited_send(func):
@@ -38,13 +84,16 @@ def rate_limited_send(func):
await asyncio.sleep(retry_in)
except TelegramForbiddenError:
tg_id = kwargs.get("tg_id") or args[1]
logger.warning(f"Пользователь {tg_id} заблокировал бота.")
logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.")
return False
except TelegramBadRequest:
tg_id = kwargs.get("tg_id") or args[1]
logger.warning(f"🚫 Чат не найден для пользователя {tg_id}.")
return False
except Exception as e:
tg_id = kwargs.get("tg_id") or args[1]
logger.error(f"❌ Ошибка отправки сообщения пользователю {tg_id}: {e}")
return False
return wrapper
@@ -85,8 +134,7 @@ async def _send_photo_notification(
buffered_photo = BufferedInputFile(image_data, filename=image_filename)
await bot.send_photo(tg_id, buffered_photo, caption=caption, reply_markup=keyboard)
return True
except TelegramForbiddenError:
logger.error(f"Пользователь {tg_id} заблокировал бота")
except (TelegramForbiddenError, TelegramBadRequest):
return False
except Exception as e:
logger.error(f"Ошибка отправки фото для пользователя {tg_id}: {e}")
@@ -104,8 +152,7 @@ async def _send_text_notification(
try:
await bot.send_message(tg_id, caption, reply_markup=keyboard)
return True
except TelegramForbiddenError:
logger.error(f"Пользователь {tg_id} заблокировал бота")
except (TelegramForbiddenError, TelegramBadRequest):
return False
except Exception as e:
logger.error(f"Неизвестная ошибка при отправке сообщения для пользователя {tg_id}: {e}")
handlers/notifications/special_notifications.py
+68 -60
View File
@@ -1,32 +1,32 @@
import asyncio
from datetime import datetime, timedelta
import asyncpg
import pytz
from aiogram import Bot, Router, types
from aiogram.exceptions import TelegramForbiddenError
from aiogram.utils.keyboard import InlineKeyboardBuilder
from config import NOTIFY_EXTRA_DAYS, NOTIFY_INACTIVE, NOTIFY_INACTIVE_TRAFFIC, SUPPORT_CHAT_URL, TRIAL_TIME
from database import (
add_notification,
check_notifications_bulk,
create_blocked_user,
from config import (
NOTIFY_EXTRA_DAYS,
NOTIFY_INACTIVE,
NOTIFY_INACTIVE_TRAFFIC,
SUPPORT_CHAT_URL,
TRIAL_TIME,
)
from database import add_notification, check_notifications_bulk, create_blocked_user
from handlers.buttons import MAIN_MENU
from handlers.keys.key_utils import get_user_traffic
from handlers.texts import TRIAL_INACTIVE_BONUS_MSG, TRIAL_INACTIVE_FIRST_MSG, ZERO_TRAFFIC_MSG
from handlers.texts import (
TRIAL_INACTIVE_BONUS_MSG,
TRIAL_INACTIVE_FIRST_MSG,
ZERO_TRAFFIC_MSG,
)
from handlers.utils import format_days
from logger import logger
from handlers.utils import format_days
from .notify_utils import send_messages_with_limit, send_notification
router = Router()
moscow_tz = pytz.timezone("Europe/Moscow")
@@ -37,19 +37,15 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection):
Если прошло 24 часа и триал не активирован, отправляется уведомление с бонусом +2 дня.
"""
logger.info("Проверка пользователей, не активировавших пробный период...")
users = await check_notifications_bulk("inactive_trial", NOTIFY_INACTIVE, conn)
logger.info(f"Найдено {len(users)} неактивных пользователей для уведомления.")
messages = []
for user in users:
tg_id = user["tg_id"]
username = user["username"]
first_name = user["first_name"]
last_name = user["last_name"]
display_name = username or first_name or last_name or "Пользователь"
builder = InlineKeyboardBuilder()
builder.row(
types.InlineKeyboardButton(
@@ -59,9 +55,7 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection):
)
builder.row(types.InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
keyboard = builder.as_markup()
trial_extended = user["last_notification_time"] is not None
if trial_extended:
total_days = NOTIFY_EXTRA_DAYS + TRIAL_TIME
message = TRIAL_INACTIVE_BONUS_MSG.format(
@@ -74,21 +68,30 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection):
message = TRIAL_INACTIVE_FIRST_MSG.format(
display_name=display_name, trial_time_formatted=format_days(TRIAL_TIME)
)
try:
await bot.send_message(tg_id, message, reply_markup=keyboard)
logger.info(f"📩 Отправлено уведомление неактивному пользователю {tg_id}.")
await add_notification(tg_id, "inactive_trial", session=conn)
except TelegramForbiddenError:
logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}. Добавляем в blocked_users.")
await create_blocked_user(tg_id, conn)
except Exception as e:
logger.error(f"⚠ Ошибка при отправке уведомления пользователю {tg_id}: {e}")
await asyncio.sleep(1)
messages.append({
"tg_id": tg_id,
"text": message,
"keyboard": keyboard,
"notification_id": "inactive_trial",
})
if messages:
results = await send_messages_with_limit(
bot,
messages,
conn=conn,
source_file="special_notifications",
messages_per_second=25
)
sent_count = 0
for msg, result in zip(messages, results):
tg_id = msg["tg_id"]
if result:
await add_notification(tg_id, msg["notification_id"], session=conn)
sent_count += 1
logger.info(f"📩 Отправлено уведомление неактивному пользователю {tg_id}.")
else:
logger.warning(f"📩 Не удалось отправить уведомление неактивному пользователю {tg_id}.")
logger.info(f"Отправлено {sent_count} уведомлений неактивным пользователям.")
logger.info("✅ Проверка пользователей с неактивным пробным периодом завершена.")
@@ -99,8 +102,8 @@ async def notify_users_no_traffic(bot: Bot, conn: asyncpg.Connection, current_ti
но исключает пользователей, у которых подписка недавно продлилась.
"""
logger.info("Проверка пользователей с нулевым трафиком...")
current_dt = datetime.fromtimestamp(current_time / 1000, tz=moscow_tz)
messages = []
for key in keys:
tg_id = key.get("tg_id")
@@ -124,7 +127,6 @@ async def notify_users_no_traffic(bot: Bot, conn: asyncpg.Connection, current_ti
expiry_dt = pytz.utc.localize(datetime.fromtimestamp(expiry_time / 1000)).astimezone(moscow_tz)
renewal_threshold = expiry_dt - timedelta(days=30)
renewal_recent = current_dt - renewal_threshold < timedelta(hours=NOTIFY_INACTIVE_TRAFFIC)
if renewal_recent:
continue
@@ -144,39 +146,45 @@ async def notify_users_no_traffic(bot: Bot, conn: asyncpg.Connection, current_ti
total_traffic = sum(
value if isinstance(value, int | float) else 0 for value in traffic_data.get("traffic", {}).values()
)
logger.info(f"Ключ для {email}: общий трафик: {total_traffic} ГБ")
try:
await conn.execute(
"UPDATE keys SET notified = TRUE WHERE tg_id = $1 AND client_id = $2", tg_id, client_id
)
except Exception as e:
logger.error(f"Ошибка обновления notified для пользователя {tg_id} (client_id: {client_id}): {e}")
continue
if total_traffic == 0:
logger.info(f"⚠ У пользователя {tg_id} ({email}) 0 ГБ трафика. Отправляем уведомление.")
builder = InlineKeyboardBuilder()
builder.row(types.InlineKeyboardButton(text="🔧 Написать в поддержку", url=SUPPORT_CHAT_URL))
builder.row(types.InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
keyboard = builder.as_markup()
message = ZERO_TRAFFIC_MSG.format(email=email)
messages.append({
"tg_id": tg_id,
"text": message,
"keyboard": keyboard,
"client_id": client_id,
})
try:
result = await send_notification(bot, tg_id, None, message, keyboard)
await conn.execute(
"UPDATE keys SET notified = TRUE WHERE tg_id = $1 AND client_id = $2", tg_id, client_id
)
if result:
logger.info(f"📩 Отправлено уведомление пользователю {tg_id} о нулевом трафике.")
else:
logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id} о нулевом трафике.")
except TelegramForbiddenError:
logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.")
await create_blocked_user(tg_id, conn)
except Exception as e:
logger.error(f"⚠ Ошибка при отправке уведомления пользователю {tg_id}: {e}")
else:
try:
await conn.execute(
"UPDATE keys SET notified = TRUE WHERE tg_id = $1 AND client_id = $2", tg_id, client_id
)
logger.info(f"Ключ для {email} имеет трафик. Обновлено notified = TRUE.")
except Exception as e:
logger.error(f"Ошибка обновления notified для пользователя {tg_id}: {e}")
if messages:
results = await send_messages_with_limit(
bot,
messages,
conn=conn,
source_file="special_notifications",
messages_per_second=25
)
sent_count = 0
for msg, result in zip(messages, results):
tg_id = msg["tg_id"]
if result:
sent_count += 1
logger.info(f"📩 Отправлено уведомление пользователю {tg_id} о нулевом трафике.")
else:
logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id} о нулевом трафике.")
logger.info(f"Отправлено {sent_count} уведомлений о нулевом трафике.")
logger.info("✅ Обработка пользователей с нулевым трафиком завершена.")