72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
|
"""Модуль для хранения и управления статистикой."""
|
|||
|
|
import json
|
|||
|
|
import logging
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import Set
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Statistics:
|
|||
|
|
"""Класс для управления статистикой сервиса."""
|
|||
|
|
|
|||
|
|
def __init__(self, stats_file: Path):
|
|||
|
|
"""
|
|||
|
|
Args:
|
|||
|
|
stats_file: Путь к JSON файлу со статистикой
|
|||
|
|
"""
|
|||
|
|
self.stats_file = stats_file
|
|||
|
|
self.stats_file.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
self._users: Set[int] = set()
|
|||
|
|
self._processed_urls: int = 0
|
|||
|
|
self._load()
|
|||
|
|
|
|||
|
|
def _load(self):
|
|||
|
|
"""Загрузить статистику из файла."""
|
|||
|
|
try:
|
|||
|
|
if self.stats_file.exists():
|
|||
|
|
with open(self.stats_file, 'r', encoding='utf-8') as f:
|
|||
|
|
data = json.load(f)
|
|||
|
|
self._users = set(data.get('users', []))
|
|||
|
|
self._processed_urls = data.get('processed_urls', 0)
|
|||
|
|
logger.info(f"Loaded statistics: {len(self._users)} users, {self._processed_urls} processed URLs")
|
|||
|
|
else:
|
|||
|
|
logger.info("Statistics file not found, starting with empty stats")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Error loading statistics: {e}", exc_info=True)
|
|||
|
|
self._users = set()
|
|||
|
|
self._processed_urls = 0
|
|||
|
|
|
|||
|
|
def _save(self):
|
|||
|
|
"""Сохранить статистику в файл."""
|
|||
|
|
try:
|
|||
|
|
with open(self.stats_file, 'w', encoding='utf-8') as f:
|
|||
|
|
json.dump({
|
|||
|
|
'users': list(self._users),
|
|||
|
|
'processed_urls': self._processed_urls
|
|||
|
|
}, f, indent=2)
|
|||
|
|
logger.debug(f"Saved statistics: {len(self._users)} users, {self._processed_urls} processed URLs")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Error saving statistics: {e}", exc_info=True)
|
|||
|
|
|
|||
|
|
def add_user(self, user_id: int):
|
|||
|
|
"""Добавить пользователя в статистику."""
|
|||
|
|
if user_id not in self._users:
|
|||
|
|
self._users.add(user_id)
|
|||
|
|
self._save()
|
|||
|
|
logger.debug(f"Added user to statistics: {user_id}")
|
|||
|
|
|
|||
|
|
def increment_processed_urls(self):
|
|||
|
|
"""Увеличить счётчик обработанных ссылок."""
|
|||
|
|
self._processed_urls += 1
|
|||
|
|
self._save()
|
|||
|
|
logger.debug(f"Incremented processed URLs counter: {self._processed_urls}")
|
|||
|
|
|
|||
|
|
def get_user_count(self) -> int:
|
|||
|
|
"""Получить количество уникальных пользователей."""
|
|||
|
|
return len(self._users)
|
|||
|
|
|
|||
|
|
def get_processed_urls_count(self) -> int:
|
|||
|
|
"""Получить количество обработанных ссылок."""
|
|||
|
|
return self._processed_urls
|