kerrad/telegram-helper-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement audio record management features in AsyncBotDB and AudioRepository

Browse Source 
- Added methods to delete audio moderation records and retrieve all audio records in async_db.py.
- Enhanced AudioRepository with functionality to delete audio records by file name and retrieve all audio message records.
- Improved logging for audio record operations to enhance monitoring and debugging capabilities.
- Updated related handlers to ensure proper integration of new audio management features.
This commit is contained in:
Andrey
2025-09-05 01:31:50 +03:00
parent fc0517c011
commit 5f6882d348
32 changed files with 2661 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
helper_bot/utils/auto_unban_scheduler.py
View File

@@ -8,6 +8,11 @@ from apscheduler.triggers.cron import CronTrigger
from helper_bot.utils.base_dependency_factory import get_global_instance
from logs.custom_logger import logger
from .metrics import (
track_time,
track_errors,
db_query_time
)
class AutoUnbanScheduler:
"""
@@ -24,7 +29,10 @@ class AutoUnbanScheduler:
def set_bot(self, bot):
"""Устанавливает экземпляр бота для отправки уведомлений"""
self.bot = bot
@track_time("auto_unban_users", "auto_unban_scheduler")
@track_errors("auto_unban_scheduler", "auto_unban_users")
@db_query_time("auto_unban_users", "users", "mixed")
async def auto_unban_users(self):
"""
Основная функция автоматического разбана пользователей.
@@ -104,6 +112,8 @@ class AutoUnbanScheduler:
return report
@track_time("send_report", "auto_unban_scheduler")
@track_errors("auto_unban_scheduler", "send_report")
async def _send_report(self, report: str):
"""Отправляет отчет в лог-канал"""
try:
@@ -117,6 +127,8 @@ class AutoUnbanScheduler:
except Exception as e:
logger.error(f"Ошибка при отправке отчета: {e}")
@track_time("send_error_report", "auto_unban_scheduler")
@track_errors("auto_unban_scheduler", "send_error_report")
async def _send_error_report(self, error_msg: str):
"""Отправляет отчет об ошибке в важный лог-канал"""
try:

148
helper_bot/utils/helper_func.py
View File

@@ -22,10 +22,11 @@ from database.models import TelegramPost
# Local imports - metrics
from .metrics import (
metrics,
track_time,
track_errors,
db_query_time
db_query_time,
track_media_processing,
track_file_operations,
)
bdf = get_global_instance()
@@ -115,7 +116,9 @@ def get_text_message(post_text: str, first_name: str, username: str = None):
else:
return f'Пост из ТГ:\n{safe_post_text}\n\nАвтор поста: {author_info}'
@track_time("download_file", "helper_func")
@track_errors("helper_func", "download_file")
@track_file_operations("unknown")
async def download_file(message: types.Message, file_id: str, content_type: str = None) -> Optional[str]:
"""
Скачивает файл по file_id из Telegram и сохраняет в соответствующую папку.
@@ -180,18 +183,16 @@ async def download_file(message: types.Message, file_id: str, content_type: str
logger.info(f"download_file: Файл успешно скачан - {file_path}, размер: {file_size} байт, время: {download_time:.2f}с")
# Записываем метрики
metrics.record_file_download(content_type or 'unknown', file_size, download_time)
return file_path
except Exception as e:
download_time = time.time() - start_time
logger.error(f"download_file: Ошибка скачивания файла {file_id}: {e}, время: {download_time:.2f}с")
metrics.record_file_download_error(content_type or 'unknown', str(e))
return None
@track_time("prepare_media_group_from_middlewares", "helper_func")
@track_errors("helper_func", "prepare_media_group_from_middlewares")
@track_media_processing("media_group")
async def prepare_media_group_from_middlewares(album, post_caption: str = ''):
"""
Создает MediaGroup согласно best practices aiogram 3.x.
@@ -243,7 +244,10 @@ async def prepare_media_group_from_middlewares(album, post_caption: str = ''):
return media_group
@track_time("add_in_db_media_mediagroup", "helper_func")
@track_errors("helper_func", "add_in_db_media_mediagroup")
@track_media_processing("media_group")
@db_query_time("add_in_db_media_mediagroup", "posts", "insert")
async def add_in_db_media_mediagroup(sent_message: List[types.Message], bot_db: Any, main_post_id: Optional[int] = None) -> bool:
"""
Добавляет контент медиа-группы в базу данных
@@ -340,7 +344,6 @@ async def add_in_db_media_mediagroup(sent_message: List[types.Message], bot_db:
if processed_count == 0:
logger.error(f"add_in_db_media_mediagroup: Не удалось обработать ни одного сообщения из медиагруппы {post_id}")
metrics.record_media_processing('media_group', processing_time, False)
return False
if failed_count > 0:
@@ -348,18 +351,18 @@ async def add_in_db_media_mediagroup(sent_message: List[types.Message], bot_db:
else:
logger.info(f"add_in_db_media_mediagroup: Успешно обработана медиагруппа {post_id} - {processed_count} сообщений, время: {processing_time:.2f}с")
# Записываем метрики
metrics.record_media_processing('media_group', processing_time, failed_count == 0)
return failed_count == 0
except Exception as e:
processing_time = time.time() - start_time
logger.error(f"add_in_db_media_mediagroup: Критическая ошибка обработки медиагруппы: {e}, время: {processing_time:.2f}с")
metrics.record_media_processing('media_group', processing_time, False)
return False
@track_time("add_in_db_media", "helper_func")
@track_errors("helper_func", "add_in_db_media")
@track_media_processing("media_group")
@db_query_time("add_in_db_media", "posts", "insert")
@track_file_operations("media")
async def add_in_db_media(sent_message: types.Message, bot_db: Any) -> bool:
"""
Добавляет контент одиночного сообщения в базу данных
@@ -430,18 +433,17 @@ async def add_in_db_media(sent_message: types.Message, bot_db: Any) -> bool:
processing_time = time.time() - start_time
logger.info(f"add_in_db_media: Контент успешно добавлен для сообщения {post_id}, тип: {content_type}, время: {processing_time:.2f}с")
# Записываем метрики
metrics.record_media_processing(content_type, processing_time, True)
return True
except Exception as e:
processing_time = time.time() - start_time
logger.error(f"add_in_db_media: Ошибка обработки медиа для сообщения {post_id}: {e}, время: {processing_time:.2f}с")
metrics.record_media_processing(content_type or 'unknown', processing_time, False)
return False
@track_time("send_media_group_message_to_private_chat", "helper_func")
@track_errors("helper_func", "send_media_group_message_to_private_chat")
@track_media_processing("media_group")
@db_query_time("send_media_group_message_to_private_chat", "posts", "insert")
async def send_media_group_message_to_private_chat(chat_id: int, message: types.Message,
media_group: List, bot_db: Any, main_post_id: Optional[int] = None) -> int:
sent_message = await message.bot.send_media_group(
@@ -461,7 +463,9 @@ async def send_media_group_message_to_private_chat(chat_id: int, message: types.
message_id = sent_message[-1].message_id
return message_id
@track_time("send_media_group_to_channel", "helper_func")
@track_errors("helper_func", "send_media_group_to_channel")
@track_media_processing("media_group")
async def send_media_group_to_channel(bot, chat_id: int, post_content: List, post_text: str):
"""
Отправляет медиа-группу с подписью к последнему файлу.
@@ -510,28 +514,32 @@ async def send_media_group_to_channel(bot, chat_id: int, post_content: List, pos
logger.error(f"Ошибка при отправке медиа-группы в чат {chat_id}: {e}")
raise
@track_time("send_text_message", "helper_func")
@track_errors("helper_func", "send_text_message")
async def send_text_message(chat_id, message: types.Message, post_text: str, markup: types.ReplyKeyboardMarkup = None):
from .rate_limiter import send_with_rate_limit
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
if markup is None:
sent_message = await message.bot.send_message(
chat_id=chat_id,
text=safe_post_text
)
message_id = sent_message.message_id
return message_id
else:
sent_message = await message.bot.send_message(
chat_id=chat_id,
text=safe_post_text,
reply_markup=markup
)
message_id = sent_message.message_id
return message_id
async def _send_message():
if markup is None:
return await message.bot.send_message(
chat_id=chat_id,
text=safe_post_text
)
else:
return await message.bot.send_message(
chat_id=chat_id,
text=safe_post_text,
reply_markup=markup
)
sent_message = await send_with_rate_limit(_send_message, chat_id)
return sent_message.message_id
@track_time("send_photo_message", "helper_func")
@track_errors("helper_func", "send_photo_message")
async def send_photo_message(chat_id, message: types.Message, photo: str, post_text: str,
markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
@@ -552,7 +560,8 @@ async def send_photo_message(chat_id, message: types.Message, photo: str, post_t
)
return sent_message
@track_time("send_video_message", "helper_func")
@track_errors("helper_func", "send_video_message")
async def send_video_message(chat_id, message: types.Message, video: str, post_text: str = "",
markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
@@ -573,7 +582,8 @@ async def send_video_message(chat_id, message: types.Message, video: str, post_t
)
return sent_message
@track_time("send_video_note_message", "helper_func")
@track_errors("helper_func", "send_video_note_message")
async def send_video_note_message(chat_id, message: types.Message, video_note: str,
markup: types.ReplyKeyboardMarkup = None):
if markup is None:
@@ -589,7 +599,8 @@ async def send_video_note_message(chat_id, message: types.Message, video_note: s
)
return sent_message
@track_time("send_audio_message", "helper_func")
@track_errors("helper_func", "send_audio_message")
async def send_audio_message(chat_id, message: types.Message, audio: str, post_text: str,
markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
@@ -611,22 +622,30 @@ async def send_audio_message(chat_id, message: types.Message, audio: str, post_t
return sent_message
@track_time("send_voice_message", "helper_func")
@track_errors("helper_func", "send_voice_message")
async def send_voice_message(chat_id, message: types.Message, voice: str,
markup: types.ReplyKeyboardMarkup = None):
if markup is None:
sent_message = await message.bot.send_voice(
chat_id=chat_id,
voice=voice
)
else:
sent_message = await message.bot.send_voice(
chat_id=chat_id,
voice=voice,
reply_markup=markup
)
return sent_message
from .rate_limiter import send_with_rate_limit
async def _send_voice():
if markup is None:
return await message.bot.send_voice(
chat_id=chat_id,
voice=voice
)
else:
return await message.bot.send_voice(
chat_id=chat_id,
voice=voice,
reply_markup=markup
)
return await send_with_rate_limit(_send_voice, chat_id)
@track_time("check_access", "helper_func")
@track_errors("helper_func", "check_access")
@db_query_time("check_access", "users", "select")
async def check_access(user_id: int, bot_db):
"""Проверка прав на совершение действий"""
from logs.custom_logger import logger
@@ -641,7 +660,9 @@ def add_days_to_date(days: int):
future_date = current_date + timedelta(days=days)
return int(future_date.timestamp())
@track_time("get_banned_users_list", "helper_func")
@track_errors("helper_func", "get_banned_users_list")
@db_query_time("get_banned_users_list", "users", "select")
async def get_banned_users_list(offset: int, bot_db):
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
@@ -689,7 +710,9 @@ async def get_banned_users_list(offset: int, bot_db):
message += f"**Дата разбана:** {safe_unban_date}\n\n"
return message
@track_time("get_banned_users_buttons", "helper_func")
@track_errors("helper_func", "get_banned_users_buttons")
@db_query_time("get_banned_users_buttons", "users", "select")
async def get_banned_users_buttons(bot_db):
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
@@ -716,7 +739,9 @@ async def get_banned_users_buttons(bot_db):
user_ids.append((safe_user_name, user_id))
return user_ids
@track_time("delete_user_blacklist", "helper_func")
@track_errors("helper_func", "delete_user_blacklist")
@db_query_time("delete_user_blacklist", "users", "delete")
async def delete_user_blacklist(user_id: int, bot_db):
return await bot_db.delete_user_blacklist(user_id=user_id)
@@ -734,7 +759,9 @@ async def check_username_and_full_name(user_id: int, username: str, full_name: s
logger.error(f"Ошибка при проверке username и full_name: {e}")
return False
@track_time("unban_notifier", "helper_func")
@track_errors("helper_func", "unban_notifier")
@db_query_time("unban_notifier", "users", "select")
async def unban_notifier(bot, BotDB, GROUP_FOR_MESSAGE):
# Получение текущего UNIX timestamp
current_date = datetime.now()
@@ -757,6 +784,7 @@ async def unban_notifier(bot, BotDB, GROUP_FOR_MESSAGE):
@track_time("update_user_info", "helper_func")
@track_errors("helper_func", "update_user_info")
@db_query_time("update_user_info", "users", "update")
async def update_user_info(source: str, message: types.Message):
# Собираем данные
full_name = message.from_user.full_name
@@ -787,12 +815,10 @@ async def update_user_info(source: str, message: types.Message):
voice_bot_welcome_received=False
)
await BotDB.add_user(user)
metrics.record_db_query("add_user", 0.0, "users", "insert")
else:
is_need_update = await check_username_and_full_name(user_id, username, full_name, BotDB)
if is_need_update:
await BotDB.update_user_info(user_id, username, full_name)
metrics.record_db_query("update_user_info", 0.0, "users", "update")
if source != 'voice':
await message.answer(
f"Давно не виделись! Вижу что ты изменился;) Теперь буду звать тебя: {full_name}")
@@ -800,7 +826,6 @@ async def update_user_info(source: str, message: types.Message):
text=f'Для пользователя: {user_id} обновлены данные в БД.\nНовое имя: {full_name}\nНовый ник:{username}. Новый эмодзи:{user_emoji}')
sleep(1)
await BotDB.update_user_date(user_id)
metrics.record_db_query("update_user_date", 0.0, "users", "update")
@track_time("check_user_emoji", "helper_func")
@@ -812,7 +837,6 @@ async def check_user_emoji(message: types.Message):
if user_emoji is None or user_emoji in ("Смайл еще не определен", "Эмоджи не определен", ""):
user_emoji = await get_random_emoji()
await BotDB.update_user_emoji(user_id=user_id, emoji=user_emoji)
metrics.record_db_query("update_user_emoji", 0.0, "users", "update")
return user_emoji

253
helper_bot/utils/metrics.py
View File

@@ -7,10 +7,13 @@ from typing import Dict, Any, Optional
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from prometheus_client.core import CollectorRegistry
import time
import os
from functools import wraps
import asyncio
from contextlib import asynccontextmanager
# Метрики rate limiter теперь создаются в основном классе
class BotMetrics:
"""Central class for managing all bot metrics."""
@@ -18,6 +21,9 @@ class BotMetrics:
def __init__(self):
self.registry = CollectorRegistry()
# Создаем метрики rate limiter в том же registry
self._create_rate_limit_metrics()
# Bot commands counter
self.bot_commands_total = Counter(
'bot_commands_total',
@@ -158,6 +164,78 @@ class BotMetrics:
registry=self.registry
)
def _create_rate_limit_metrics(self):
"""Создает метрики rate limiter в основном registry"""
try:
# Создаем метрики rate limiter в том же registry
self.rate_limit_requests_total = Counter(
'rate_limit_requests_total',
'Total number of rate limited requests',
['chat_id', 'status', 'error_type'],
registry=self.registry
)
self.rate_limit_errors_total = Counter(
'rate_limit_errors_total',
'Total number of rate limit errors',
['error_type', 'chat_id'],
registry=self.registry
)
self.rate_limit_wait_duration_seconds = Histogram(
'rate_limit_wait_duration_seconds',
'Time spent waiting due to rate limiting',
['chat_id'],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0],
registry=self.registry
)
self.rate_limit_active_chats = Gauge(
'rate_limit_active_chats',
'Number of active chats with rate limiting',
registry=self.registry
)
self.rate_limit_success_rate = Gauge(
'rate_limit_success_rate',
'Success rate of rate limited requests',
['chat_id'],
registry=self.registry
)
self.rate_limit_requests_per_minute = Gauge(
'rate_limit_requests_per_minute',
'Requests per minute',
['chat_id'],
registry=self.registry
)
self.rate_limit_total_requests = Gauge(
'rate_limit_total_requests',
'Total number of requests',
['chat_id'],
registry=self.registry
)
self.rate_limit_total_errors = Gauge(
'rate_limit_total_errors',
'Total number of errors',
['chat_id', 'error_type'],
registry=self.registry
)
self.rate_limit_avg_wait_time_seconds = Gauge(
'rate_limit_avg_wait_time_seconds',
'Average wait time in seconds',
['chat_id'],
registry=self.registry
)
except Exception as e:
# Логируем ошибку, но не прерываем инициализацию
import logging
logging.warning(f"Failed to create rate limit metrics: {e}")
def record_command(self, command_type: str, handler_type: str = "unknown", user_type: str = "unknown", status: str = "success"):
"""Record a bot command execution."""
self.bot_commands_total.labels(
@@ -267,8 +345,97 @@ class BotMetrics:
method_name="add_in_db_media"
).inc()
def record_db_error(self, error_type: str, query_type: str = "unknown", table_name: str = "unknown", operation: str = "unknown"):
"""Record database error occurrence."""
self.db_errors_total.labels(
error_type=error_type,
query_type=query_type,
table_name=table_name,
operation=operation
).inc()
def record_rate_limit_request(self, chat_id: int, success: bool, wait_time: float = 0.0, error_type: str = None):
"""Record rate limit request metrics."""
try:
# Определяем статус
status = "success" if success else "error"
# Записываем счетчик запросов
self.rate_limit_requests_total.labels(
chat_id=str(chat_id),
status=status,
error_type=error_type or "none"
).inc()
# Записываем время ожидания
if wait_time > 0:
self.rate_limit_wait_duration_seconds.labels(
chat_id=str(chat_id)
).observe(wait_time)
# Записываем ошибки
if not success and error_type:
self.rate_limit_errors_total.labels(
error_type=error_type,
chat_id=str(chat_id)
).inc()
except Exception as e:
import logging
logging.warning(f"Failed to record rate limit request: {e}")
def update_rate_limit_gauges(self):
"""Update rate limit gauge metrics."""
try:
from .rate_limit_monitor import rate_limit_monitor
# Обновляем количество активных чатов
self.rate_limit_active_chats.set(len(rate_limit_monitor.stats))
# Обновляем метрики для каждого чата
for chat_id, chat_stats in rate_limit_monitor.stats.items():
chat_id_str = str(chat_id)
# Процент успеха
self.rate_limit_success_rate.labels(
chat_id=chat_id_str
).set(chat_stats.success_rate)
# Запросов в минуту
self.rate_limit_requests_per_minute.labels(
chat_id=chat_id_str
).set(chat_stats.requests_per_minute)
# Общее количество запросов
self.rate_limit_total_requests.labels(
chat_id=chat_id_str
).set(chat_stats.total_requests)
# Среднее время ожидания
self.rate_limit_avg_wait_time_seconds.labels(
chat_id=chat_id_str
).set(chat_stats.average_wait_time)
# Количество ошибок по типам
if chat_stats.retry_after_errors > 0:
self.rate_limit_total_errors.labels(
chat_id=chat_id_str,
error_type="RetryAfter"
).set(chat_stats.retry_after_errors)
if chat_stats.other_errors > 0:
self.rate_limit_total_errors.labels(
chat_id=chat_id_str,
error_type="Other"
).set(chat_stats.other_errors)
except Exception as e:
import logging
logging.warning(f"Failed to update rate limit gauges: {e}")
def get_metrics(self) -> bytes:
"""Generate metrics in Prometheus format."""
# Обновляем gauge метрики rate limiter перед генерацией
self.update_rate_limit_gauges()
return generate_latest(self.registry)
@@ -449,3 +616,89 @@ async def track_middleware(middleware_name: str):
middleware_name
)
raise
def track_media_processing(content_type: str = "unknown"):
"""Decorator to track media processing operations."""
def decorator(func):
@wraps(func)
async def async_wrapper(*args, **kwargs):
start_time = time.time()
try:
result = await func(*args, **kwargs)
duration = time.time() - start_time
metrics.record_media_processing(content_type, duration, True)
return result
except Exception as e:
duration = time.time() - start_time
metrics.record_media_processing(content_type, duration, False)
raise
@wraps(func)
def sync_wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
metrics.record_media_processing(content_type, duration, True)
return result
except Exception as e:
duration = time.time() - start_time
metrics.record_media_processing(content_type, duration, False)
raise
if asyncio.iscoroutinefunction(func):
return async_wrapper
return sync_wrapper
return decorator
def track_file_operations(content_type: str = "unknown"):
"""Decorator to track file download/upload operations."""
def decorator(func):
@wraps(func)
async def async_wrapper(*args, **kwargs):
start_time = time.time()
try:
result = await func(*args, **kwargs)
duration = time.time() - start_time
# Получаем размер файла из результата
file_size = 0
if result and isinstance(result, str) and os.path.exists(result):
file_size = os.path.getsize(result)
# Записываем метрики
metrics.record_file_download(content_type, file_size, duration)
return result
except Exception as e:
duration = time.time() - start_time
metrics.record_file_download_error(content_type, str(e))
raise
@wraps(func)
def sync_wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
# Получаем размер файла из результата
file_size = 0
if result and isinstance(result, str) and os.path.exists(result):
file_size = os.path.getsize(result)
# Записываем метрики
metrics.record_file_download(content_type, file_size, duration)
return result
except Exception as e:
duration = time.time() - start_time
metrics.record_file_download_error(content_type, str(e))
raise
if asyncio.iscoroutinefunction(func):
return async_wrapper
return sync_wrapper
return decorator

220
helper_bot/utils/rate_limit_monitor.py Normal file
View File

@@ -0,0 +1,220 @@
"""
Мониторинг и статистика rate limiting
"""
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import defaultdict, deque
from logs.custom_logger import logger
@dataclass
class RateLimitStats:
"""Статистика rate limiting для чата"""
chat_id: int
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
retry_after_errors: int = 0
other_errors: int = 0
total_wait_time: float = 0.0
last_request_time: float = 0.0
request_times: deque = field(default_factory=lambda: deque(maxlen=100))
@property
def success_rate(self) -> float:
"""Процент успешных запросов"""
if self.total_requests == 0:
return 1.0
return self.successful_requests / self.total_requests
@property
def error_rate(self) -> float:
"""Процент ошибок"""
return 1.0 - self.success_rate
@property
def average_wait_time(self) -> float:
"""Среднее время ожидания"""
if self.total_requests == 0:
return 0.0
return self.total_wait_time / self.total_requests
@property
def requests_per_minute(self) -> float:
"""Запросов в минуту"""
if not self.request_times:
return 0.0
current_time = time.time()
minute_ago = current_time - 60
# Подсчитываем запросы за последнюю минуту
recent_requests = sum(1 for req_time in self.request_times if req_time > minute_ago)
return recent_requests
class RateLimitMonitor:
"""Монитор для отслеживания статистики rate limiting"""
def __init__(self, max_history_size: int = 1000):
self.stats: Dict[int, RateLimitStats] = defaultdict(lambda: RateLimitStats(0))
self.global_stats = RateLimitStats(0)
self.max_history_size = max_history_size
self.error_history: deque = deque(maxlen=max_history_size)
def record_request(self, chat_id: int, success: bool, wait_time: float = 0.0, error_type: Optional[str] = None):
"""Записывает информацию о запросе"""
current_time = time.time()
# Обновляем статистику для чата
chat_stats = self.stats[chat_id]
chat_stats.chat_id = chat_id
chat_stats.total_requests += 1
chat_stats.total_wait_time += wait_time
chat_stats.last_request_time = current_time
chat_stats.request_times.append(current_time)
if success:
chat_stats.successful_requests += 1
else:
chat_stats.failed_requests += 1
if error_type == "RetryAfter":
chat_stats.retry_after_errors += 1
else:
chat_stats.other_errors += 1
# Записываем ошибку в историю
self.error_history.append({
'chat_id': chat_id,
'error_type': error_type,
'timestamp': current_time,
'wait_time': wait_time
})
# Обновляем глобальную статистику
self.global_stats.total_requests += 1
self.global_stats.total_wait_time += wait_time
self.global_stats.last_request_time = current_time
self.global_stats.request_times.append(current_time)
if success:
self.global_stats.successful_requests += 1
else:
self.global_stats.failed_requests += 1
if error_type == "RetryAfter":
self.global_stats.retry_after_errors += 1
else:
self.global_stats.other_errors += 1
def get_chat_stats(self, chat_id: int) -> Optional[RateLimitStats]:
"""Получает статистику для конкретного чата"""
return self.stats.get(chat_id)
def get_global_stats(self) -> RateLimitStats:
"""Получает глобальную статистику"""
return self.global_stats
def get_top_chats_by_requests(self, limit: int = 10) -> List[tuple]:
"""Получает топ чатов по количеству запросов"""
sorted_chats = sorted(
self.stats.items(),
key=lambda x: x[1].total_requests,
reverse=True
)
return sorted_chats[:limit]
def get_chats_with_high_error_rate(self, threshold: float = 0.1) -> List[tuple]:
"""Получает чаты с высоким процентом ошибок"""
high_error_chats = [
(chat_id, stats) for chat_id, stats in self.stats.items()
if stats.error_rate > threshold and stats.total_requests > 5
]
return sorted(high_error_chats, key=lambda x: x[1].error_rate, reverse=True)
def get_recent_errors(self, minutes: int = 60) -> List[dict]:
"""Получает недавние ошибки"""
current_time = time.time()
cutoff_time = current_time - (minutes * 60)
return [
error for error in self.error_history
if error['timestamp'] > cutoff_time
]
def get_error_summary(self, minutes: int = 60) -> Dict[str, int]:
"""Получает сводку ошибок за указанный период"""
recent_errors = self.get_recent_errors(minutes)
error_summary = defaultdict(int)
for error in recent_errors:
error_summary[error['error_type']] += 1
return dict(error_summary)
def log_statistics(self, log_level: str = "info"):
"""Логирует текущую статистику"""
global_stats = self.get_global_stats()
log_message = (
f"Rate Limit Statistics:\n"
f" Total requests: {global_stats.total_requests}\n"
f" Success rate: {global_stats.success_rate:.2%}\n"
f" Error rate: {global_stats.error_rate:.2%}\n"
f" RetryAfter errors: {global_stats.retry_after_errors}\n"
f" Other errors: {global_stats.other_errors}\n"
f" Average wait time: {global_stats.average_wait_time:.2f}s\n"
f" Requests per minute: {global_stats.requests_per_minute:.1f}\n"
f" Active chats: {len(self.stats)}"
)
if log_level == "error":
logger.error(log_message)
elif log_level == "warning":
logger.warning(log_message)
else:
logger.info(log_message)
# Логируем чаты с высоким процентом ошибок
high_error_chats = self.get_chats_with_high_error_rate(0.2)
if high_error_chats:
logger.warning(f"Chats with high error rate (>20%): {len(high_error_chats)}")
for chat_id, stats in high_error_chats[:5]: # Показываем только первые 5
logger.warning(f" Chat {chat_id}: {stats.error_rate:.2%} error rate ({stats.failed_requests}/{stats.total_requests})")
def reset_stats(self, chat_id: Optional[int] = None):
"""Сбрасывает статистику"""
if chat_id is None:
# Сбрасываем всю статистику
self.stats.clear()
self.global_stats = RateLimitStats(0)
self.error_history.clear()
else:
# Сбрасываем статистику для конкретного чата
if chat_id in self.stats:
del self.stats[chat_id]
# Глобальный экземпляр монитора
rate_limit_monitor = RateLimitMonitor()
def record_rate_limit_request(chat_id: int, success: bool, wait_time: float = 0.0, error_type: Optional[str] = None):
"""Удобная функция для записи информации о запросе"""
rate_limit_monitor.record_request(chat_id, success, wait_time, error_type)
def get_rate_limit_summary() -> Dict:
"""Получает краткую сводку по rate limiting"""
global_stats = rate_limit_monitor.get_global_stats()
recent_errors = rate_limit_monitor.get_recent_errors(60) # За последний час
return {
'total_requests': global_stats.total_requests,
'success_rate': global_stats.success_rate,
'error_rate': global_stats.error_rate,
'recent_errors_count': len(recent_errors),
'active_chats': len(rate_limit_monitor.stats),
'requests_per_minute': global_stats.requests_per_minute,
'average_wait_time': global_stats.average_wait_time
}

215
helper_bot/utils/rate_limiter.py Normal file
View File

@@ -0,0 +1,215 @@
"""
Rate limiter для предотвращения Flood control ошибок в Telegram Bot API
"""
import asyncio
import time
from typing import Dict, Optional, Any, Callable
from dataclasses import dataclass
from aiogram.exceptions import TelegramRetryAfter, TelegramAPIError
from logs.custom_logger import logger
from .metrics import metrics
@dataclass
class RateLimitConfig:
"""Конфигурация для rate limiting"""
messages_per_second: float = 0.5 # Максимум 0.5 сообщений в секунду на чат
burst_limit: int = 3 # Максимум 3 сообщения подряд
retry_after_multiplier: float = 1.2 # Множитель для увеличения задержки при retry
max_retry_delay: float = 60.0 # Максимальная задержка между попытками
class ChatRateLimiter:
"""Rate limiter для конкретного чата"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.last_send_time = 0.0
self.burst_count = 0
self.burst_reset_time = 0.0
self.retry_delay = 1.0
async def wait_if_needed(self) -> None:
"""Ждет если необходимо для соблюдения rate limit"""
current_time = time.time()
# Сбрасываем счетчик burst если прошло достаточно времени
if current_time >= self.burst_reset_time:
self.burst_count = 0
self.burst_reset_time = current_time + 1.0
# Проверяем burst limit
if self.burst_count >= self.config.burst_limit:
wait_time = self.burst_reset_time - current_time
if wait_time > 0:
logger.info(f"Burst limit reached, waiting {wait_time:.2f}s")
await asyncio.sleep(wait_time)
current_time = time.time()
self.burst_count = 0
self.burst_reset_time = current_time + 1.0
# Проверяем минимальный интервал между сообщениями
time_since_last = current_time - self.last_send_time
min_interval = 1.0 / self.config.messages_per_second
if time_since_last < min_interval:
wait_time = min_interval - time_since_last
logger.debug(f"Rate limiting: waiting {wait_time:.2f}s")
await asyncio.sleep(wait_time)
# Обновляем время последней отправки
self.last_send_time = time.time()
self.burst_count += 1
class GlobalRateLimiter:
"""Глобальный rate limiter для всех чатов"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.chat_limiters: Dict[int, ChatRateLimiter] = {}
self.global_last_send = 0.0
self.global_min_interval = 0.1 # Минимум 100ms между любыми сообщениями
def get_chat_limiter(self, chat_id: int) -> ChatRateLimiter:
"""Получает rate limiter для конкретного чата"""
if chat_id not in self.chat_limiters:
self.chat_limiters[chat_id] = ChatRateLimiter(self.config)
return self.chat_limiters[chat_id]
async def wait_if_needed(self, chat_id: int) -> None:
"""Ждет если необходимо для соблюдения глобального и чат-специфичного rate limit"""
current_time = time.time()
# Глобальный rate limit
time_since_global = current_time - self.global_last_send
if time_since_global < self.global_min_interval:
wait_time = self.global_min_interval - time_since_global
await asyncio.sleep(wait_time)
current_time = time.time()
# Чат-специфичный rate limit
chat_limiter = self.get_chat_limiter(chat_id)
await chat_limiter.wait_if_needed()
self.global_last_send = time.time()
class RetryHandler:
"""Обработчик повторных попыток с экспоненциальной задержкой"""
def __init__(self, config: RateLimitConfig):
self.config = config
async def execute_with_retry(
self,
func: Callable,
chat_id: int,
*args,
max_retries: int = 3,
**kwargs
) -> Any:
"""Выполняет функцию с повторными попытками при ошибках"""
retry_count = 0
current_delay = self.config.retry_after_multiplier
total_wait_time = 0.0
while retry_count <= max_retries:
try:
result = await func(*args, **kwargs)
# Записываем успешный запрос
metrics.record_rate_limit_request(chat_id, True, total_wait_time)
return result
except TelegramRetryAfter as e:
retry_count += 1
if retry_count > max_retries:
logger.error(f"Max retries exceeded for RetryAfter: {e}")
metrics.record_rate_limit_request(chat_id, False, total_wait_time, "RetryAfter")
raise
# Используем время ожидания от Telegram или наше увеличенное
wait_time = max(e.retry_after, current_delay)
wait_time = min(wait_time, self.config.max_retry_delay)
total_wait_time += wait_time
logger.warning(f"RetryAfter error, waiting {wait_time:.2f}s (attempt {retry_count}/{max_retries})")
await asyncio.sleep(wait_time)
current_delay *= self.config.retry_after_multiplier
except TelegramAPIError as e:
retry_count += 1
if retry_count > max_retries:
logger.error(f"Max retries exceeded for TelegramAPIError: {e}")
metrics.record_rate_limit_request(chat_id, False, total_wait_time, "TelegramAPIError")
raise
wait_time = min(current_delay, self.config.max_retry_delay)
total_wait_time += wait_time
logger.warning(f"TelegramAPIError, waiting {wait_time:.2f}s (attempt {retry_count}/{max_retries}): {e}")
await asyncio.sleep(wait_time)
current_delay *= self.config.retry_after_multiplier
except Exception as e:
# Для других ошибок не делаем retry
logger.error(f"Non-retryable error: {e}")
metrics.record_rate_limit_request(chat_id, False, total_wait_time, "Other")
raise
class TelegramRateLimiter:
"""Основной класс для rate limiting в Telegram боте"""
def __init__(self, config: Optional[RateLimitConfig] = None):
self.config = config or RateLimitConfig()
self.global_limiter = GlobalRateLimiter(self.config)
self.retry_handler = RetryHandler(self.config)
async def send_with_rate_limit(
self,
send_func: Callable,
chat_id: int,
*args,
**kwargs
) -> Any:
"""Отправляет сообщение с соблюдением rate limit и retry логики"""
async def _send():
await self.global_limiter.wait_if_needed(chat_id)
return await send_func(*args, **kwargs)
return await self.retry_handler.execute_with_retry(_send, chat_id)
# Глобальный экземпляр rate limiter
from helper_bot.config.rate_limit_config import get_rate_limit_config, RateLimitSettings
def _create_rate_limit_config(settings: RateLimitSettings) -> RateLimitConfig:
"""Создает RateLimitConfig из RateLimitSettings"""
return RateLimitConfig(
messages_per_second=settings.messages_per_second,
burst_limit=settings.burst_limit,
retry_after_multiplier=settings.retry_after_multiplier,
max_retry_delay=settings.max_retry_delay
)
# Получаем конфигурацию из настроек
_rate_limit_settings = get_rate_limit_config("production")
_default_config = _create_rate_limit_config(_rate_limit_settings)
telegram_rate_limiter = TelegramRateLimiter(_default_config)
async def send_with_rate_limit(send_func: Callable, chat_id: int, *args, **kwargs) -> Any:
"""
Удобная функция для отправки сообщений с rate limiting
Args:
send_func: Функция отправки (например, bot.send_message)
chat_id: ID чата
*args, **kwargs: Аргументы для функции отправки
Returns:
Результат выполнения функции отправки
"""
return await telegram_rate_limiter.send_with_rate_limit(send_func, chat_id, *args, **kwargs)