62 lines
3.2 KiB
Python
62 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Хендлер для сбора и хранения метрик работы бота.
|
||
Счётчики: сообщения, обработанные файлы, AI-запросы, ошибки.
|
||
Предоставляет методы increment, get, reset.
|
||
"""
|
||
|
||
import logging
|
||
from typing import Dict, Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MetricsHandler:
|
||
"""Управляет статистикой работы бота."""
|
||
|
||
def __init__(self):
|
||
# Внутренний словарь со счётчиками (все инициализируются нулями)
|
||
self._metrics: Dict[str, int] = {
|
||
"messages_total": 0, # общее количество входящих сообщений
|
||
"files_processed": 0, # количество файлов, загруженных и обработанных
|
||
"ai_requests": 0, # количество вызовов AI (генерация, классификация и т.д.)
|
||
"errors_total": 0 # общее количество ошибок (исключений)
|
||
}
|
||
|
||
def increment(self, metric_name: str, value: int = 1) -> None:
|
||
"""
|
||
Увеличивает указанный счётчик на заданное значение (по умолчанию 1).
|
||
|
||
Аргументы:
|
||
metric_name (str): имя счётчика (должно быть одним из ключей _metrics).
|
||
value (int): на сколько увеличить.
|
||
|
||
Логирует изменение на уровне DEBUG, если счётчик существует.
|
||
Если передан неизвестный счётчик, логирует предупреждение.
|
||
"""
|
||
if metric_name in self._metrics:
|
||
self._metrics[metric_name] += value
|
||
logger.debug(f"Метрика {metric_name} увеличена до {self._metrics[metric_name]}")
|
||
else:
|
||
logger.warning(f"Попытка инкремента неизвестной метрики: {metric_name}")
|
||
|
||
def get(self, metric_name: str = None) -> int | Dict[str, int]:
|
||
"""
|
||
Возвращает либо значение конкретной метрики, либо копию всех метрик.
|
||
|
||
Аргументы:
|
||
metric_name (str, optional): имя счётчика. Если None, возвращается словарь всех метрик.
|
||
|
||
Возвращает:
|
||
int или Dict[str, int]: запрошенное значение или копия словаря.
|
||
"""
|
||
if metric_name:
|
||
return self._metrics.get(metric_name, 0)
|
||
# Возвращаем копию, чтобы внешний код не мог изменить внутренний словарь напрямую
|
||
return self._metrics.copy()
|
||
|
||
def reset(self) -> None:
|
||
"""Сбрасывает все счётчики в 0. Логирует действие."""
|
||
for key in self._metrics:
|
||
self._metrics[key] = 0
|
||
logger.info("Все метрики сброшены") |