94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
|
"""Управление историей отправленных новостей"""
|
|||
|
|
|
|||
|
|
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()
|