"""Управление историей отправленных новостей""" import json from datetime import datetime, timedelta from typing import Dict, Optional from pathlib import Path import logging logger = logging.getLogger(__name__) class HistoryManager: """Менеджер истории отправленных новостей""" def __init__(self, history_file: str, history_days: int, max_history_size: int): self.history_file = Path(history_file) self.history_days = history_days self.max_history_size = max_history_size self.sent_history: Dict[str, dict] = {} # Создаем директорию для файла истории self.history_file.parent.mkdir(parents=True, exist_ok=True) self._load() def _load(self) -> None: """Загружает историю из файла""" try: if self.history_file.exists(): with open(self.history_file, 'r', encoding='utf-8') as f: self.sent_history = json.load(f) logger.info(f"Загружено {len(self.sent_history)} записей из истории") else: logger.info("Файл с историей не найден, создаю новый") self.sent_history = {} except json.JSONDecodeError: logger.warning("Файл истории повреждён, создаю новый") self.sent_history = {} def save(self) -> None: """Сохраняет историю в файл""" try: with open(self.history_file, 'w', encoding='utf-8') as f: json.dump(self.sent_history, f, indent=2, ensure_ascii=False, default=str) except Exception as e: logger.error(f"Ошибка сохранения истории: {e}") def add(self, url: str, title: str) -> None: """Добавляет ссылку в историю""" self.sent_history[url] = { "date": datetime.now().isoformat(), "title": title[:100] } def is_already_sent(self, url: str) -> bool: """Проверяет, была ли ссылка уже отправлена""" return url in self.sent_history def clean_old(self, force: bool = False) -> None: """Удаляет старые записи из истории""" if not self.sent_history: return original_size = len(self.sent_history) now = datetime.now() cutoff_date = now - timedelta(days=self.history_days) new_history = {} for url, data in self.sent_history.items(): try: if isinstance(data, str): timestamp = datetime.fromisoformat(data) if timestamp > cutoff_date: new_history[url] = {"date": data, "title": "unknown"} else: timestamp = datetime.fromisoformat(data['date']) if timestamp > cutoff_date: new_history[url] = data except (ValueError, KeyError): pass self.sent_history = new_history # Ограничиваем размер if len(self.sent_history) > self.max_history_size: sorted_items = sorted( self.sent_history.items(), key=lambda x: x[1]['date'] if isinstance(x[1], dict) else x[1], reverse=True ) self.sent_history = dict(sorted_items[:self.max_history_size]) if original_size != len(self.sent_history) or force: self.save()