Capybara-z
432 lines
16 KiB
Python
432 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import time
|
|
|
|
from collections import OrderedDict, deque
|
|
|
|
import aiofiles
|
|
|
|
from aiogram import Bot
|
|
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter
|
|
from aiogram.types import BufferedInputFile, InlineKeyboardMarkup
|
|
|
|
from datetime import datetime
|
|
|
|
import pytz
|
|
|
|
from database import async_session_maker
|
|
from database.bans import save_blocked_user_ids
|
|
from handlers.utils import format_hours, format_minutes, get_russian_month
|
|
from logger import logger
|
|
from services.tariffs.tariff_display import get_key_tariff_display
|
|
from utils.custom_emojis import _process_text
|
|
|
|
moscow_tz = pytz.timezone("Europe/Moscow")
|
|
|
|
|
|
_photo_cache: OrderedDict[str, str] = OrderedDict()
|
|
_photo_cache_lock = asyncio.Lock()
|
|
_PHOTO_CACHE_MAX = 64
|
|
_SUPPORTED_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp", ".gif")
|
|
|
|
|
|
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)
|
|
|
|
|
|
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:
|
|
self.max_rate = max_rate
|
|
self.window = window
|
|
self.send_times: deque = deque()
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def acquire(self):
|
|
async with self.lock:
|
|
while True:
|
|
now = time.time()
|
|
cutoff = now - self.window
|
|
while self.send_times and self.send_times[0] <= cutoff:
|
|
self.send_times.popleft()
|
|
if len(self.send_times) < self.max_rate:
|
|
self.send_times.append(now)
|
|
return
|
|
time_to_wait = (self.send_times[0] + self.window) - now
|
|
if time_to_wait > 0:
|
|
await asyncio.sleep(time_to_wait + 0.001)
|
|
|
|
|
|
def rate_limited_send(func):
|
|
async def wrapper(*args, **kwargs):
|
|
while True:
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except TelegramRetryAfter as e:
|
|
await asyncio.sleep(int(e.retry_after) + 1)
|
|
except TelegramForbiddenError:
|
|
return False
|
|
except TelegramBadRequest:
|
|
return False
|
|
except Exception as e:
|
|
tg_id = kwargs.get("tg_id") or (args[1] if len(args) > 1 else "?")
|
|
logger.error(f"Ошибка отправки пользователю {tg_id}: {e}")
|
|
return False
|
|
return wrapper
|
|
|
|
|
|
async def send_notification(
|
|
bot: Bot,
|
|
tg_id: int,
|
|
image_filename: str | None,
|
|
caption: str,
|
|
keyboard: InlineKeyboardMarkup | None = None,
|
|
) -> bool:
|
|
if image_filename is None:
|
|
return await _send_text(bot, tg_id, caption, keyboard)
|
|
|
|
photo_path = os.path.join("img", image_filename)
|
|
cached_id = await _get_cached_file_id(photo_path)
|
|
if cached_id:
|
|
return await _send_photo(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(bot, tg_id, actual_path, image_filename, caption, keyboard)
|
|
else:
|
|
logger.warning(f"Файл изображения не найден: {photo_path}")
|
|
return await _send_text(bot, tg_id, caption, keyboard)
|
|
|
|
|
|
@rate_limited_send
|
|
async def _send_photo(
|
|
bot: Bot,
|
|
tg_id: int,
|
|
photo_path: str,
|
|
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 f:
|
|
image_data = await f.read()
|
|
buffered = BufferedInputFile(image_data, filename=image_filename)
|
|
result = await bot.send_photo(tg_id, buffered, 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
|
|
except Exception as e:
|
|
logger.error(f"Ошибка отправки фото пользователю {tg_id}: {e}")
|
|
return await _send_text(bot, tg_id, caption, keyboard)
|
|
|
|
|
|
@rate_limited_send
|
|
async def _send_text(
|
|
bot: Bot,
|
|
tg_id: int,
|
|
caption: str,
|
|
keyboard: InlineKeyboardMarkup | None = None,
|
|
) -> bool:
|
|
try:
|
|
processed, entities = await _process_text(caption)
|
|
kwargs = {"reply_markup": keyboard}
|
|
if entities:
|
|
kwargs["entities"] = entities
|
|
kwargs["parse_mode"] = None
|
|
await bot.send_message(tg_id, processed, **kwargs)
|
|
return True
|
|
except (TelegramForbiddenError, TelegramBadRequest):
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Ошибка отправки текста пользователю {tg_id}: {e}")
|
|
return False
|
|
|
|
|
|
_NOTIFY_MAX_ATTEMPTS = 5
|
|
_NOTIFY_MAX_RETRY_AFTER = 120.0
|
|
|
|
|
|
class FastNotificationSender:
|
|
def __init__(self, bot: Bot, messages_per_second: int = 35, max_attempts: int = _NOTIFY_MAX_ATTEMPTS) -> None:
|
|
self.bot = bot
|
|
self.rate_limiter = NotificationRateLimiter(max_rate=messages_per_second)
|
|
self.max_attempts = max_attempts
|
|
self.blocked_users: set[int] = set()
|
|
self.queue: asyncio.Queue = asyncio.Queue()
|
|
self.results: list[bool] = []
|
|
self.total_sent = 0
|
|
self.pending_retries = 0
|
|
self.is_running = False
|
|
|
|
async def _send_one(self, msg: dict) -> str:
|
|
tg_id = msg["tg_id"]
|
|
try:
|
|
await self.rate_limiter.acquire()
|
|
|
|
processed_text, entities = await _process_text(msg["text"])
|
|
emoji_kwargs = {}
|
|
if entities:
|
|
emoji_kwargs["parse_mode"] = None
|
|
|
|
if msg.get("photo"):
|
|
photo_path = os.path.join("img", msg["photo"])
|
|
cached_id = await _get_cached_file_id(photo_path)
|
|
|
|
caption_kwargs = {"caption_entities": entities} if entities else {}
|
|
|
|
if cached_id:
|
|
await self.bot.send_photo(
|
|
chat_id=tg_id, photo=cached_id,
|
|
caption=processed_text, reply_markup=msg.get("keyboard"),
|
|
**emoji_kwargs, **caption_kwargs,
|
|
)
|
|
else:
|
|
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 = BufferedInputFile(image_data, filename=os.path.basename(actual_path))
|
|
result = await self.bot.send_photo(
|
|
chat_id=tg_id, photo=buffered,
|
|
caption=processed_text, reply_markup=msg.get("keyboard"),
|
|
**emoji_kwargs, **caption_kwargs,
|
|
)
|
|
if result and hasattr(result, "photo") and result.photo:
|
|
await _set_cached_file_id(photo_path, result.photo[-1].file_id)
|
|
else:
|
|
text_kwargs = {"entities": entities} if entities else {}
|
|
await self.bot.send_message(
|
|
chat_id=tg_id, text=processed_text, reply_markup=msg.get("keyboard"),
|
|
**emoji_kwargs, **text_kwargs,
|
|
)
|
|
else:
|
|
text_kwargs = {"entities": entities} if entities else {}
|
|
await self.bot.send_message(
|
|
chat_id=tg_id, text=processed_text, reply_markup=msg.get("keyboard"),
|
|
**emoji_kwargs, **text_kwargs,
|
|
)
|
|
return "ok"
|
|
|
|
except TelegramRetryAfter as e:
|
|
wait_seconds = min(float(e.retry_after), _NOTIFY_MAX_RETRY_AFTER)
|
|
msg["_attempts"] = msg.get("_attempts", 0) + 1
|
|
msg["_retry_at"] = time.time() + wait_seconds
|
|
return "retry"
|
|
except TelegramForbiddenError:
|
|
self.blocked_users.add(tg_id)
|
|
return "fail"
|
|
except TelegramBadRequest as e:
|
|
if "chat not found" in str(e).lower():
|
|
self.blocked_users.add(tg_id)
|
|
return "fail"
|
|
except Exception:
|
|
return "fail"
|
|
|
|
async def _schedule_retry(self, msg: dict) -> None:
|
|
try:
|
|
wait = msg.get("_retry_at", 0.0) - time.time()
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
await self.queue.put(msg)
|
|
except asyncio.CancelledError:
|
|
self.results.append(False)
|
|
raise
|
|
except Exception:
|
|
self.results.append(False)
|
|
finally:
|
|
self.pending_retries -= 1
|
|
|
|
async def _worker(self):
|
|
while True:
|
|
try:
|
|
msg = await asyncio.wait_for(self.queue.get(), timeout=0.5)
|
|
except (TimeoutError, asyncio.TimeoutError):
|
|
if not self.is_running:
|
|
return
|
|
continue
|
|
|
|
try:
|
|
result = await self._send_one(msg)
|
|
if result == "ok":
|
|
self.total_sent += 1
|
|
self.results.append(True)
|
|
elif result == "retry":
|
|
if msg.get("_attempts", 0) < self.max_attempts:
|
|
try:
|
|
asyncio.create_task(self._schedule_retry(msg))
|
|
self.pending_retries += 1
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Не удалось запланировать повтор для {msg.get('tg_id')}: {e}"
|
|
)
|
|
self.results.append(False)
|
|
else:
|
|
self.results.append(False)
|
|
else:
|
|
self.results.append(False)
|
|
except Exception as e:
|
|
logger.error(f"Ошибка в воркере уведомлений: {e}")
|
|
self.results.append(False)
|
|
finally:
|
|
self.queue.task_done()
|
|
|
|
async def _save_blocked_users(self):
|
|
if not self.blocked_users:
|
|
return
|
|
try:
|
|
async with async_session_maker() as session:
|
|
await save_blocked_user_ids(session, list(self.blocked_users))
|
|
await session.commit()
|
|
logger.info(f"Добавлено до {len(self.blocked_users)} пользователей в blocked_users")
|
|
except Exception as e:
|
|
logger.error(f"Ошибка сохранения заблокированных: {e}")
|
|
|
|
async def send_all(self, messages: list[dict], workers: int = 15) -> list[bool]:
|
|
if not messages:
|
|
return []
|
|
|
|
self.is_running = True
|
|
self.results = []
|
|
self.total_sent = 0
|
|
self.blocked_users = set()
|
|
self.pending_retries = 0
|
|
start = time.time()
|
|
|
|
for msg in messages:
|
|
await self.queue.put(msg)
|
|
|
|
worker_tasks = [asyncio.create_task(self._worker()) for _ in range(workers)]
|
|
|
|
while True:
|
|
await self.queue.join()
|
|
if self.pending_retries <= 0:
|
|
break
|
|
await asyncio.sleep(0.5)
|
|
|
|
self.is_running = False
|
|
for task in worker_tasks:
|
|
task.cancel()
|
|
await asyncio.gather(*worker_tasks, return_exceptions=True)
|
|
await self._save_blocked_users()
|
|
|
|
duration = time.time() - start
|
|
speed = self.total_sent / duration if duration > 0 else 0
|
|
logger.info(f"Уведомления: {self.total_sent}/{len(messages)} за {duration:.1f}s ({speed:.1f} msg/s)")
|
|
|
|
return self.results
|
|
|
|
|
|
async def send_messages_with_limit(
|
|
bot: Bot,
|
|
messages: list[dict],
|
|
messages_per_second: int = 35,
|
|
) -> list[bool]:
|
|
sender = FastNotificationSender(bot, messages_per_second)
|
|
return await sender.send_all(messages)
|
|
|
|
|
|
async def prepare_key_expiry_data(key, session, current_time: int) -> dict:
|
|
if isinstance(key, dict):
|
|
expiry_timestamp = key.get("expiry_time")
|
|
email = key.get("email") or ""
|
|
record = dict(key)
|
|
else:
|
|
expiry_timestamp = getattr(key, "expiry_time", None)
|
|
email = getattr(key, "email", "") or ""
|
|
record = {
|
|
"tariff_id": getattr(key, "tariff_id", None),
|
|
"server_id": getattr(key, "server_id", None),
|
|
"client_id": getattr(key, "client_id", None),
|
|
"selected_device_limit": getattr(key, "selected_device_limit", None),
|
|
"selected_traffic_limit": getattr(key, "selected_traffic_limit", None),
|
|
}
|
|
|
|
if not expiry_timestamp:
|
|
return {
|
|
"hours_left_formatted": "",
|
|
"formatted_expiry_date": "",
|
|
"tariff_name": "—",
|
|
"tariff_details": "",
|
|
}
|
|
|
|
delta_ms = max(0, expiry_timestamp - current_time)
|
|
total_minutes = delta_ms // (60 * 1000)
|
|
hours_left = total_minutes // 60
|
|
minutes_left = total_minutes % 60
|
|
|
|
if hours_left > 0 or minutes_left > 0:
|
|
parts = []
|
|
if hours_left > 0:
|
|
parts.append(format_hours(hours_left))
|
|
if minutes_left > 0:
|
|
parts.append(format_minutes(minutes_left))
|
|
hours_left_formatted = f"⏳ Осталось времени: {' '.join(parts)}"
|
|
else:
|
|
hours_left_formatted = "⏳ Последний день подписки!"
|
|
|
|
expiry_datetime = datetime.fromtimestamp(expiry_timestamp / 1000, tz=moscow_tz)
|
|
month_name = get_russian_month(expiry_datetime)
|
|
formatted_expiry_date = expiry_datetime.strftime(f"%d {month_name} %Y, %H:%M (МСК)")
|
|
|
|
tariff_name = "—"
|
|
subgroup_title = ""
|
|
traffic_limit_gb = 0
|
|
device_limit = 0
|
|
|
|
try:
|
|
name, subgroup_title, traffic_limit_gb, device_limit, _, _ = await get_key_tariff_display(
|
|
session=session,
|
|
key_record=record,
|
|
)
|
|
if name:
|
|
tariff_name = name
|
|
except Exception as error:
|
|
logger.warning(f"[NOTIFY] Ошибка тарифных лимитов для {email}: {error}")
|
|
|
|
traffic_text = "безлимит" if traffic_limit_gb == 0 else f"{traffic_limit_gb} ГБ"
|
|
devices_text = "безлимит" if device_limit == 0 else str(device_limit)
|
|
|
|
lines = []
|
|
if subgroup_title:
|
|
lines.append(subgroup_title)
|
|
lines.append(f"Трафик: {traffic_text}")
|
|
lines.append(f"Устройств: {devices_text}")
|
|
tariff_details = "\n" + "\n".join(lines) if lines else ""
|
|
|
|
return {
|
|
"hours_left_formatted": hours_left_formatted,
|
|
"formatted_expiry_date": formatted_expiry_date,
|
|
"tariff_name": tariff_name,
|
|
"tariff_details": tariff_details,
|
|
}
|