96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""
|
|
Cache manager for tracking processed articles to avoid reprocessing.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional
|
|
|
|
from config import CACHE_FILE
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class CacheManager:
|
|
"""Manages caching of processed articles to prevent duplicate processing."""
|
|
|
|
def __init__(self, cache_file: str = CACHE_FILE):
|
|
self.cache_file = cache_file
|
|
self.cache = self._load_cache()
|
|
|
|
def _load_cache(self) -> Dict:
|
|
"""Load cache from file."""
|
|
try:
|
|
if os.path.exists(self.cache_file):
|
|
with open(self.cache_file, 'r', encoding='utf-8') as f:
|
|
cache_data = json.load(f)
|
|
# Ensure the cache has the correct structure
|
|
if "processed_files" not in cache_data:
|
|
cache_data["processed_files"] = {}
|
|
return cache_data
|
|
else:
|
|
# Create empty cache file if it doesn't exist
|
|
cache_data = {
|
|
"cache_version": "1.0",
|
|
"created": datetime.now().isoformat(),
|
|
"processed_files": {}
|
|
}
|
|
# Set the cache attribute directly
|
|
self.cache = cache_data
|
|
self._save_cache()
|
|
return cache_data
|
|
except Exception as e:
|
|
logger.error(f"Error loading cache file {self.cache_file}: {e}")
|
|
logger.info("Creating new empty cache due to load error")
|
|
# Return empty cache on error
|
|
return {
|
|
"cache_version": "1.0",
|
|
"created": datetime.now().isoformat(),
|
|
"processed_files": {}
|
|
}
|
|
|
|
def _save_cache(self) -> None:
|
|
"""Save cache to file."""
|
|
try:
|
|
# Create directory if it doesn't exist
|
|
os.makedirs(os.path.dirname(self.cache_file), exist_ok=True)
|
|
|
|
with open(self.cache_file, 'w', encoding='utf-8') as f:
|
|
json.dump(self.cache, f, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
logger.error(f"Error saving cache file {self.cache_file}: {e}")
|
|
|
|
def is_processed(self, file_path: str) -> bool:
|
|
"""Check if a file has been processed."""
|
|
return file_path in self.cache.get("processed_files", {})
|
|
|
|
def mark_processed(self, file_path: str, status: str = "processed") -> None:
|
|
"""Mark a file as processed."""
|
|
# Ensure we're using the correct cache structure
|
|
if "processed_files" not in self.cache:
|
|
self.cache["processed_files"] = {}
|
|
self.cache["processed_files"][file_path] = {
|
|
"processed_date": datetime.now().isoformat(),
|
|
"status": status,
|
|
"last_updated": datetime.now().isoformat()
|
|
}
|
|
self._save_cache()
|
|
|
|
def get_processed_files(self) -> List[str]:
|
|
"""Get list of all processed files."""
|
|
return list(self.cache.get("processed_files", {}).keys())
|
|
|
|
def get_cache_stats(self) -> Dict:
|
|
"""Get cache statistics."""
|
|
return {
|
|
"total_files": len(self.cache.get("processed_files", {})),
|
|
"processed_files": len(self.cache.get("processed_files", {})),
|
|
"cache_file": self.cache_file
|
|
}
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Clear the entire cache."""
|
|
self.cache = {}
|
|
self._save_cache()
|
|
logger.info("Cache cleared") |