147 lines
5.2 KiB
Python
147 lines
5.2 KiB
Python
"""
|
|
Cache manager for tracking processed articles to avoid reprocessing.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
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."""
|
|
empty_cache = {
|
|
"cache_version": "1.0",
|
|
"created": datetime.now().isoformat(),
|
|
"processed_files": {},
|
|
}
|
|
|
|
try:
|
|
if not os.path.exists(self.cache_file):
|
|
logger.info(
|
|
f"Cache file does not exist: {self.cache_file}. Creating new empty cache."
|
|
)
|
|
return empty_cache
|
|
|
|
# Check if the file is empty (0 bytes)
|
|
file_size = os.path.getsize(self.cache_file)
|
|
if file_size == 0:
|
|
logger.warning(
|
|
f"Cache file is empty (0 bytes): {self.cache_file}. Treating as fresh cache."
|
|
)
|
|
return empty_cache
|
|
|
|
with open(self.cache_file, "r", encoding="utf-8") as f:
|
|
content = f.read().strip()
|
|
|
|
# Check if content is empty or just whitespace
|
|
if not content:
|
|
logger.warning(
|
|
f"Cache file contains only whitespace: {self.cache_file}. Treating as fresh cache."
|
|
)
|
|
return empty_cache
|
|
|
|
cache_data = json.loads(content)
|
|
|
|
# Handle case where JSON loaded as None, list, or other non-dict type
|
|
if not isinstance(cache_data, dict):
|
|
logger.warning(
|
|
f"Cache file contains invalid data type ({type(cache_data).__name__}). Treating as fresh cache."
|
|
)
|
|
return empty_cache
|
|
|
|
# Ensure the cache has the correct structure
|
|
if (
|
|
"processed_files" not in cache_data
|
|
or cache_data["processed_files"] is None
|
|
):
|
|
cache_data["processed_files"] = {}
|
|
|
|
# Validate that processed_files is a dict
|
|
if not isinstance(cache_data["processed_files"], dict):
|
|
logger.warning(
|
|
f"processed_files is not a dict ({type(cache_data['processed_files']).__name__}). Resetting to empty."
|
|
)
|
|
cache_data["processed_files"] = {}
|
|
|
|
logger.info(
|
|
f"Loaded cache with {len(cache_data['processed_files'])} processed files"
|
|
)
|
|
return cache_data
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.warning(
|
|
f"Cache file contains invalid JSON: {e}. Treating as fresh cache."
|
|
)
|
|
return empty_cache
|
|
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
|
|
|
|
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."""
|
|
processed_files = self.cache.get("processed_files")
|
|
|
|
# Handle case where processed_files is None or not a dict
|
|
if not isinstance(processed_files, dict):
|
|
logger.warning(
|
|
f"processed_files is not a valid dict (got {type(processed_files).__name__}). Returning False."
|
|
)
|
|
return False
|
|
|
|
return file_path in 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")
|