From 995e2f8016d8afaf0a8d70d2caa8a0dd21b70e2b Mon Sep 17 00:00:00 2001 From: Vladless Date: Wed, 15 Apr 2026 11:10:57 +0000 Subject: [PATCH] Notification photo cache, bulk traffic, blocked preload filter / Fix discount datetime aware --- handlers/admin/clusters/cluster_sync.py | 2 +- handlers/keys/key_mode/key_discount_mode.py | 10 ++- .../notifications/general_notifications.py | 45 ++++++++++- handlers/notifications/notify_utils.py | 77 ++++++++++++++++--- .../notifications/special_notifications.py | 66 ++++++++-------- handlers/tariffs/addons/key_addons_main.py | 2 +- panels/remnawave_runtime.py | 45 +++++++++++ 7 files changed, 199 insertions(+), 48 deletions(-) diff --git a/handlers/admin/clusters/cluster_sync.py b/handlers/admin/clusters/cluster_sync.py index 89765059..139fa96c 100644 --- a/handlers/admin/clusters/cluster_sync.py +++ b/handlers/admin/clusters/cluster_sync.py @@ -510,7 +510,7 @@ async def handle_sync_cluster( tariffs_cache = {t.id: dict(t.__dict__) for t in tariffs_list} if only_remnawave: - batch_size = 50 + batch_size = 250 total_keys = len(keys_to_sync) processed_count = 0 diff --git a/handlers/keys/key_mode/key_discount_mode.py b/handlers/keys/key_mode/key_discount_mode.py index 1043b133..1eb74805 100644 --- a/handlers/keys/key_mode/key_discount_mode.py +++ b/handlers/keys/key_mode/key_discount_mode.py @@ -1,4 +1,4 @@ -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from aiogram import F, Router from aiogram.fsm.context import FSMContext @@ -46,7 +46,9 @@ async def handle_discount_entry(callback: CallbackQuery, session: AsyncSession): discount_active_hours = int(NOTIFICATIONS_CONFIG.get("DISCOUNT_ACTIVE_HOURS", DISCOUNT_ACTIVE_HOURS)) - now = datetime.utcnow() + now = datetime.now(timezone.utc) + if last_time.tzinfo is None: + last_time = last_time.replace(tzinfo=timezone.utc) if now - last_time > timedelta(hours=discount_active_hours): await callback.message.edit_text("⏳ Срок действия скидки истёк.") return @@ -133,7 +135,9 @@ async def handle_ultra_discount(callback: CallbackQuery, session: AsyncSession): discount_active_hours = int(NOTIFICATIONS_CONFIG.get("DISCOUNT_ACTIVE_HOURS", DISCOUNT_ACTIVE_HOURS)) - now = datetime.utcnow() + now = datetime.now(timezone.utc) + if last_time.tzinfo is None: + last_time = last_time.replace(tzinfo=timezone.utc) if now - last_time > timedelta(hours=discount_active_hours): await callback.message.edit_text("⏳ Срок действия финальной скидки истёк.") return diff --git a/handlers/notifications/general_notifications.py b/handlers/notifications/general_notifications.py index 3d219b9d..18124677 100644 --- a/handlers/notifications/general_notifications.py +++ b/handlers/notifications/general_notifications.py @@ -7,7 +7,7 @@ from typing import Any, Optional import pytz from aiogram import Bot, Router -from sqlalchemy import select, text, update +from sqlalchemy import exists, or_, select, text, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from config import ( @@ -44,6 +44,7 @@ from database import ( update_key_tariff, ) from database.models import Key, Tariff, User +from database.models.users import BlockedUser, ManualBan from database.tariffs import ( check_tariff_exists, get_tariff_by_id, @@ -112,7 +113,14 @@ async def preload_notification_data(session: AsyncSession) -> dict[str, Any]: ) .outerjoin(Tariff, Key.tariff_id == Tariff.id) .outerjoin(User, Key.user_id == User.id) - .where(Key.is_frozen.is_(False)) + .where( + Key.is_frozen.is_(False), + ~exists().where(BlockedUser.tg_id == Key.tg_id), + ~exists().where( + ManualBan.tg_id == Key.tg_id, + or_(ManualBan.until.is_(None), ManualBan.until > datetime.utcnow()), + ), + ) ) result = await session.execute(stmt) @@ -589,10 +597,43 @@ async def notify_expiring_keys( await asyncio.sleep(1) +async def _get_blocked_expired_keys(session: AsyncSession, current_time: int) -> list: + """Получает истекшие ключи заблокированных/забаненных пользователей.""" + stmt = ( + select(Key) + .where( + Key.is_frozen.is_(False), + Key.expiry_time.isnot(None), + Key.expiry_time < current_time, + or_( + exists().where(BlockedUser.tg_id == Key.tg_id), + exists().where( + ManualBan.tg_id == Key.tg_id, + or_(ManualBan.until.is_(None), ManualBan.until > datetime.utcnow()), + ), + ), + ) + ) + result = await session.execute(stmt) + return list(result.scalars().all()) + + async def handle_expired_keys(ctx: NotificationContext, keys: list): logger.info("Начало обработки истекших ключей.") expired_keys = [key for key in keys if key.expiry_time and key.expiry_time < ctx.current_time] + + try: + blocked_expired = await _get_blocked_expired_keys(ctx.session, ctx.current_time) + if blocked_expired: + logger.info(f"Дополнительно найдено {len(blocked_expired)} истекших ключей заблокированных пользователей.") + existing_ids = {key.client_id for key in expired_keys} + for bk in blocked_expired: + if bk.client_id not in existing_ids: + expired_keys.append(bk) + except Exception as error: + logger.error(f"Ошибка получения ключей заблокированных пользователей: {error}") + logger.info(f"Найдено {len(expired_keys)} истекших ключей.") tg_ids = [key.tg_id for key in expired_keys] diff --git a/handlers/notifications/notify_utils.py b/handlers/notifications/notify_utils.py index cd67d6e9..51094bc3 100644 --- a/handlers/notifications/notify_utils.py +++ b/handlers/notifications/notify_utils.py @@ -2,7 +2,7 @@ import asyncio import os import time -from collections import deque +from collections import OrderedDict, deque from datetime import datetime import aiofiles @@ -25,6 +25,40 @@ from services.tariffs.tariff_display import get_key_tariff_display moscow_tz = pytz.timezone("Europe/Moscow") +_photo_cache: OrderedDict[str, str] = OrderedDict() +_photo_cache_lock = asyncio.Lock() +_PHOTO_CACHE_MAX = 64 + + +async def _get_cached_file_id(photo_path: str) -> str | None: + async with _photo_cache_lock: + fid = _photo_cache.get(photo_path) + if fid: + _photo_cache.move_to_end(photo_path) + return fid + + +async def _set_cached_file_id(photo_path: str, file_id: str) -> None: + async with _photo_cache_lock: + if photo_path not in _photo_cache: + _photo_cache[photo_path] = file_id + if len(_photo_cache) > _PHOTO_CACHE_MAX: + _photo_cache.popitem(last=False) + + +_SUPPORTED_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp", ".gif") + + +def _find_photo_file(photo_path: str) -> str | None: + if os.path.isfile(photo_path): + return photo_path + base_name = os.path.splitext(photo_path)[0] + for ext in _SUPPORTED_EXTENSIONS: + candidate = base_name + ext + if os.path.isfile(candidate): + return candidate + return None + class NotificationRateLimiter: def __init__(self, max_rate: int = 35, window: float = 1.0) -> None: @@ -80,15 +114,25 @@ class FastNotificationSender: if msg.photo: photo_path = os.path.join("img", msg.photo) - if os.path.isfile(photo_path): - async with aiofiles.open(photo_path, "rb") as f: - image_data = await f.read() - buffered_photo = BufferedInputFile(image_data, filename=msg.photo) + cached_id = await _get_cached_file_id(photo_path) + + if cached_id: await self.bot.send_photo( - chat_id=msg.tg_id, photo=buffered_photo, caption=msg.text, reply_markup=msg.keyboard + chat_id=msg.tg_id, photo=cached_id, caption=msg.text, reply_markup=msg.keyboard ) else: - await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) + actual_path = _find_photo_file(photo_path) + if actual_path: + async with aiofiles.open(actual_path, "rb") as f: + image_data = await f.read() + buffered_photo = BufferedInputFile(image_data, filename=os.path.basename(actual_path)) + result = await self.bot.send_photo( + chat_id=msg.tg_id, photo=buffered_photo, caption=msg.text, reply_markup=msg.keyboard + ) + if result and hasattr(result, "photo") and result.photo: + await _set_cached_file_id(photo_path, result.photo[-1].file_id) + else: + await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) else: await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) return True @@ -259,8 +303,13 @@ async def send_notification( return await _send_text_notification(bot, tg_id, caption, keyboard) photo_path = os.path.join("img", image_filename) - if os.path.isfile(photo_path): - return await _send_photo_notification(bot, tg_id, photo_path, image_filename, caption, keyboard) + cached_id = await _get_cached_file_id(photo_path) + if cached_id: + return await _send_photo_notification(bot, tg_id, photo_path, image_filename, caption, keyboard, cached_id) + + actual_path = _find_photo_file(photo_path) + if actual_path: + return await _send_photo_notification(bot, tg_id, actual_path, image_filename, caption, keyboard) else: logger.warning(f"Файл с изображением не найден: {photo_path}") return await _send_text_notification(bot, tg_id, caption, keyboard) @@ -274,12 +323,20 @@ async def _send_photo_notification( image_filename: str, caption: str, keyboard: InlineKeyboardMarkup | None = None, + cached_file_id: str | None = None, ) -> bool: try: + if cached_file_id: + await bot.send_photo(tg_id, cached_file_id, caption=caption, reply_markup=keyboard) + return True async with aiofiles.open(photo_path, "rb") as image_file: image_data = await image_file.read() buffered_photo = BufferedInputFile(image_data, filename=image_filename) - await bot.send_photo(tg_id, buffered_photo, caption=caption, reply_markup=keyboard) + result = await bot.send_photo(tg_id, buffered_photo, caption=caption, reply_markup=keyboard) + if result and hasattr(result, "photo") and result.photo: + await _set_cached_file_id( + os.path.join("img", image_filename), result.photo[-1].file_id + ) return True except (TelegramForbiddenError, TelegramBadRequest): return False diff --git a/handlers/notifications/special_notifications.py b/handlers/notifications/special_notifications.py index 85b43535..8e847b5c 100644 --- a/handlers/notifications/special_notifications.py +++ b/handlers/notifications/special_notifications.py @@ -22,6 +22,7 @@ from database.models import Key, User from database.tariffs import get_tariffs from handlers.buttons import CONNECT_DEVICE, MAIN_MENU, SUPPORT, TRIAL_BONUS from handlers.keys.utils import build_key_callback +from panels.remnawave_runtime import fetch_all_remnawave_traffic from handlers.notifications.notify_utils import send_messages_with_limit from handlers.texts import ( TRIAL_INACTIVE_BONUS_MSG, @@ -153,50 +154,53 @@ async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time: remnawave_webapp_enabled = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_ENABLED", REMNAWAVE_WEBAPP)) open_in_browser = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_OPEN_IN_BROWSER", REMNAWAVE_WEBAPP_OPEN_IN_BROWSER)) + candidate_keys = [] + for key in keys: + if key.tariff_id not in trial_tariff_ids: + continue + if key.created_at is None or key.notified: + continue + created_at_dt = pytz.utc.localize(datetime.fromtimestamp(key.created_at / 1000)).astimezone(moscow_tz) + if current_dt < created_at_dt + timedelta(hours=inactive_traffic_hours): + continue + if key.expiry_time: + expiry_dt = pytz.utc.localize(datetime.fromtimestamp(key.expiry_time / 1000)).astimezone(moscow_tz) + if current_dt > expiry_dt: + continue + candidate_keys.append(key) + + if not candidate_keys: + logger.info("Нет кандидатов для проверки нулевого трафика.") + return + + needed_uuids = {key.client_id for key in candidate_keys if key.client_id} + logger.info(f"[Zero Traffic] Кандидатов: {len(candidate_keys)}, уникальных UUID: {len(needed_uuids)}") + + try: + traffic_map = await fetch_all_remnawave_traffic(session, needed_uuids=needed_uuids) + except Exception as error: + logger.error(f"[Zero Traffic] Ошибка bulk-получения трафика: {error}") + return + messages = [] keys_to_mark_notified = [] - for key in keys: + for key in candidate_keys: tg_id = key.tg_id email = key.email - created_at = key.created_at client_id = key.client_id - expiry_time = key.expiry_time - notified = key.notified - tariff_id = key.tariff_id - - if tariff_id not in trial_tariff_ids: - continue - - if created_at is None or notified: - continue - - created_at_dt = pytz.utc.localize(datetime.fromtimestamp(created_at / 1000)).astimezone(moscow_tz) - if current_dt < created_at_dt + timedelta(hours=inactive_traffic_hours): - continue - - if expiry_time: - expiry_dt = pytz.utc.localize(datetime.fromtimestamp(expiry_time / 1000)).astimezone(moscow_tz) - if current_dt > expiry_dt: - continue keys_to_mark_notified.append(client_id) - try: - traffic_data = await get_user_traffic(session, tg_id, email) - except Exception as error: - logger.error(f"Ошибка получения трафика для {email}: {error}") + used_bytes = traffic_map.get(client_id) + if used_bytes is None: + logger.warning(f"[Zero Traffic] UUID {client_id} ({email}) не найден в bulk-данных, пропуск") continue - if traffic_data.get("status") != "success": - logger.warning(f"Ошибка при получении трафика для {email}: {traffic_data.get('message')}") + if used_bytes > 0: continue - total_traffic = sum( - value if isinstance(value, int | float) else 0 for value in traffic_data.get("traffic", {}).values() - ) - - if total_traffic == 0: + if used_bytes == 0: logger.info(f"У пользователя {tg_id} ({email}) 0 ГБ трафика. Отправляем уведомление.") builder = InlineKeyboardBuilder() diff --git a/handlers/tariffs/addons/key_addons_main.py b/handlers/tariffs/addons/key_addons_main.py index 4477c037..eea0d974 100644 --- a/handlers/tariffs/addons/key_addons_main.py +++ b/handlers/tariffs/addons/key_addons_main.py @@ -834,7 +834,7 @@ async def handle_addons_confirm(callback: CallbackQuery, state: FSMContext, sess server_id = record["server_id"] selected_traffic_gb_for_effective = ( - int(selected_traffic_gb) if selected_traffic_gb is not None and has_traffic_option else 0 + int(selected_traffic_gb) if selected_traffic_gb is not None and has_traffic_option else None ) current_subgroup = None diff --git a/panels/remnawave_runtime.py b/panels/remnawave_runtime.py index 189f888d..0090a980 100644 --- a/panels/remnawave_runtime.py +++ b/panels/remnawave_runtime.py @@ -204,6 +204,51 @@ async def invalidate_remnawave_profile( await invalidate_remnawave_profile_cache(client_id=client_id) +async def fetch_all_remnawave_traffic( + session: AsyncSession, + needed_uuids: set[str] | None = None, +) -> dict[str, int]: + """Bulk-запрос: возвращает {uuid: usedTrafficBytes} для всех (или нужных) юзеров Remnawave.""" + servers = await get_servers(session) + api_url = None + for cluster in servers.values(): + for srv in cluster: + if srv.get("panel_type") == "remnawave": + api_url = srv.get("api_url") + break + if api_url: + break + + if not api_url: + logger.warning("[Bulk Traffic] Нет доступных Remnawave-серверов") + return {} + + api = RemnawaveAPI(api_url) + try: + all_users = await api.get_all_users_time( + username=REMNAWAVE_LOGIN, + password=REMNAWAVE_PASSWORD, + ) + finally: + await api.aclose() + + if not all_users: + return {} + + result: dict[str, int] = {} + for user in all_users: + uuid = user.get("uuid") + if not uuid: + continue + if needed_uuids and uuid not in needed_uuids: + continue + traffic = user.get("userTraffic") or {} + result[uuid] = traffic.get("usedTrafficBytes", 0) + + logger.info(f"[Bulk Traffic] Получено {len(result)} профилей трафика из {len(all_users)} юзеров Remnawave") + return result + + async def with_remnawave_api( session: AsyncSession, server_ref: str,