""" Cache management utilities for download plans. Handles caching, loading, and cleanup of download plan data. """ import hashlib import json from datetime import datetime, timedelta from pathlib import Path from karaoke_downloader.data_path_manager import get_data_path_manager # Constants DEFAULT_CACHE_EXPIRATION_DAYS = 1 DEFAULT_CACHE_FILENAME_LENGTH_LIMIT = 200 # Increased from 60 DEFAULT_CACHE_FILENAME_PREFIX_LENGTH = 100 # Increased from 40 def get_download_plan_cache_file(mode, **kwargs): """Generate a unique cache filename based on mode and key parameters.""" parts = [f"plan_{mode}"] # Handle parameters in a more readable way for k, v in sorted(kwargs.items()): if k == "channels_hash": # Use a shorter version of the hash for readability parts.append(f"hash{v[:8]}") else: parts.append(f"{k}{v}") base = "_".join(parts) # Hash for safety if string is still too long if len(base) > DEFAULT_CACHE_FILENAME_LENGTH_LIMIT: base = ( base[:DEFAULT_CACHE_FILENAME_PREFIX_LENGTH] + "_" + hashlib.md5(base.encode()).hexdigest()[:8] ) return get_data_path_manager().get_path(f"{base}.json") def load_cached_plan(cache_file, max_age_days=DEFAULT_CACHE_EXPIRATION_DAYS): """Load a cached download plan if it exists and is not expired.""" if not cache_file.exists(): return None, None try: with open(cache_file, "r", encoding="utf-8") as f: cache_data = json.load(f) cache_time = datetime.fromisoformat(cache_data.get("timestamp")) if datetime.now() - cache_time < timedelta(days=max_age_days): print( f"🗂️ Using cached download plan from {cache_time} ({cache_file.name})." ) return cache_data["download_plan"], cache_data["unmatched"] except Exception as e: print(f"⚠️ Could not load download plan cache: {e}") return None, None def save_plan_cache(cache_file, download_plan, unmatched): """Save a download plan to cache.""" if download_plan: cache_data = { "timestamp": datetime.now().isoformat(), "download_plan": download_plan, "unmatched": unmatched, } with open(cache_file, "w", encoding="utf-8") as f: json.dump(cache_data, f, indent=2, ensure_ascii=False) print(f"🗂️ Saved new download plan cache: {cache_file.name}") else: if cache_file.exists(): cache_file.unlink() print("🗂️ No matches found, not saving download plan cache.") def delete_plan_cache(cache_file): """Delete a download plan cache file.""" if cache_file.exists(): try: cache_file.unlink() print(f"🗑️ Deleted download plan cache: {cache_file.name}") except Exception as e: print(f"⚠️ Could not delete download plan cache: {e}")