import os
import json
import logging
from typing import Dict, Any, Optional
from youtube_mcp_server.config import Config
logger = logging.getLogger("youtube-mcp-server")
class CacheService:
def __init__(self):
pass
def _get_path(self, base_dir: str, video_id: str, language: str) -> str:
return os.path.join(base_dir, language, f"{video_id}.json")
def _ensure_dir(self, path: str):
os.makedirs(os.path.dirname(path), exist_ok=True)
def _load_json(self, path: str) -> Optional[Any]:
if os.path.exists(path):
logger.info(f"Loading from cache: {path}")
try:
with open(path, "r") as f:
return json.load(f)
except Exception as e:
logger.warning(f"Failed to load cache {path}: {e}")
return None
def _save_json(self, path: str, data: Any):
self._ensure_dir(path)
logger.info(f"Saving to cache: {path}")
try:
with open(path, "w") as f:
json.dump(data, f, indent=2)
except Exception as e:
logger.error(f"Failed to save cache {path}: {e}")
# Transcription Cache
def load_transcription(self, video_id: str, language: str = "en") -> Optional[Dict[str, Any]]:
path = self._get_path(Config.TRANSCRIPTIONS_DIR, video_id, language)
return self._load_json(path)
def save_transcription(self, video_id: str, data: Dict[str, Any], language: str = "en"):
path = self._get_path(Config.TRANSCRIPTIONS_DIR, video_id, language)
self._save_json(path, data)
