71 lines
2.9 KiB
Python
71 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
|