# bot/library_cache.py import json import os import time from typing import List from plex_client import PlexClient from difflib import SequenceMatcher CACHE_FILE = 'bot/library_cache.json' MAX_CACHE_AGE_HOURS = 6 class LibraryCache: def __init__(self): if self.is_cache_stale(): print("📦 Plex cache is stale or missing — refreshing...") self.refresh_cache() else: print("✅ Using cached Plex library data.") self.load_cache() def is_cache_stale(self) -> bool: if not os.path.exists(CACHE_FILE): return True age_hours = (time.time() - os.path.getmtime(CACHE_FILE)) / 3600 return age_hours > MAX_CACHE_AGE_HOURS def refresh_cache(self): plex = PlexClient() media = plex.server.library.sections() new_data = [] for section in media: section_type = section.type.lower() # Normalize to 'show' if section_type == 'show': section_type = 'tv' if section_type in ['movie', 'tv']: for item in section.all(): new_data.append({ 'title': item.title.strip(), 'type': section_type, 'genres': [g.tag for g in getattr(item, 'genres', [])], 'year': getattr(item, 'year', None) }) self.data = new_data self.save_cache() def save_cache(self): with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(self.data, f, ensure_ascii=False, indent=2) def load_cache(self): with open(CACHE_FILE, 'r', encoding='utf-8') as f: self.data = json.load(f) def get_titles_by_type(self, media_type: str) -> List[str]: return [item['title'] for item in self.data if item['type'] == media_type] def search(self, title: str, media_type: str) -> bool: def is_match(a: str, b: str) -> bool: ratio = SequenceMatcher(None, a.lower(), b.lower()).ratio() return ratio > 0.8 # tweak as needed for item in self.data: if item["type"] != media_type: continue if is_match(title, item["title"]): return True return False