import time from collections import deque from dataclasses import dataclass, field from typing import Deque, Tuple WINDOWS = { "5min": 5 * 60, "1h": 60 * 60, } @dataclass class IncrementalCounter: """Счётчик, который умеет: • `+=` – увеличивает внутренний счётчик на 1 • `last_5min`, `last_hour`, `total` – сколько было увеличений за последние 5 минут, 1 час и за всё время соответственно """ # Внутренний счётчик (сумма всех увеличений) _total: int = 0 # История – deque из timestamps (float) когда происходил инкремент _history: Deque[float] = field(default_factory=deque, init=False) # ---------- Оператор += ---------- def __iadd__(self, other): """ При любом `+=` увеличиваем счётчик на 1. Возвращаем self, чтобы поддерживать цепочку выражений. """ # Счётчик всегда +1, игнорируем `other` self._total += 1 # Храним только время события self._history.append(time.monotonic()) # Удаляем слишком старые элементы (самый длинный интервал = 1h) self._purge_old_entries() return self # ---------- Свойства для статистики ---------- @property def total(self) -> int: """Общее количество прибавлений.""" return self._total @property def last_5min(self) -> int: """Сколько прибавлений было за последние 5 минут.""" return self._count_in_window(WINDOWS["5min"]) @property def last_hour(self) -> int: """Сколько прибавлений было за последний час.""" return self._count_in_window(WINDOWS["1h"]) # ---------- Вспомогательные методы ---------- def _purge_old_entries(self) -> None: """Удаляем из deque все записи старше 1 часа.""" cutoff = time.monotonic() - WINDOWS["1h"] while self._history and self._history[0] < cutoff: self._history.popleft() def _count_in_window(self, seconds: float) -> int: """Подсчёт, сколько событий попадает в заданный интервал.""" cutoff = time.monotonic() - seconds # Удаляем старые элементы, которые уже не нужны while self._history and self._history[0] < cutoff: self._history.popleft() return len(self._history) # ---------- Пользовательский интерфейс ---------- def __repr__(self): return ( f"" )