Something went wrong. Try again.
A local, collaborative music player (main-branch-only mirror of forge.ejuarezg.com/ejuarezg/jam-session)
Something went wrong. Try again.
4.0 kB · 132 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133import jsonimport osimport shutilimport timefrom datetime import datetime, timezonefrom pathlib import Path
from jam_session.models import Track
class AudioCacheManager: DEFAULT_CACHE_DIR = Path.home() / ".cache" / "jam-session"
def __init__( self, max_size_gb: float = 1.0, cache_dir: Path | None = None, prefetch_count: int = 5, ): self.cache_dir = cache_dir or self.DEFAULT_CACHE_DIR self.max_size_bytes = int(max_size_gb * 1024**3) self.prefetch_count = prefetch_count self.cache_dir.mkdir(parents=True, exist_ok=True)
self._index_path = self.cache_dir / "cache_index.json" self._index: dict[str, dict] = self._load_index()
def _load_index(self) -> dict[str, dict]: if self._index_path.exists(): try: return json.loads(self._index_path.read_text()) except (json.JSONDecodeError, OSError): pass return {}
def _save_index(self): self._index_path.write_text(json.dumps(self._index, default=str))
def get_cached(self, video_id: str) -> Path | None: entry = self._index.get(video_id) if entry is None: return None
file_path = Path(entry["file_path"]) if not file_path.exists(): del self._index[video_id] self._save_index() return None
entry["last_accessed"] = datetime.now(timezone.utc).isoformat() self._save_index() return file_path
def add_to_cache( self, video_id: str, url: str, title: str, duration: int | None ) -> Path: import yt_dlp
output_template = str(self.cache_dir / f"{video_id}.%(ext)s")
opts = { "quiet": True, "format": "bestaudio/best", "outtmpl": output_template, "postprocessors": [ {"key": "FFmpegExtractAudio", "preferredcodec": "opus"} ], }
with yt_dlp.YoutubeDL(opts) as ydl: ydl.download([url])
opus_path = self.cache_dir / f"{video_id}.opus" if not opus_path.exists(): candidates = sorted(self.cache_dir.glob(f"{video_id}.*")) audio_files = [p for p in candidates if p.suffix != ".json"] if audio_files: opus_path = audio_files[0] else: raise FileNotFoundError(f"No audio file found for {video_id}")
file_size = opus_path.stat().st_size
self._index[video_id] = { "title": title, "duration": duration, "file_path": str(opus_path), "file_size": file_size, "last_accessed": datetime.now(timezone.utc).isoformat(), }
self._evict_lru() self._save_index()
return opus_path
def _evict_lru(self): total_size = sum(e["file_size"] for e in self._index.values()) if total_size <= self.max_size_bytes: return
entries = sorted( self._index.items(), key=lambda x: x[1].get("last_accessed", "") )
for video_id, entry in entries: if total_size <= self.max_size_bytes: break
file_path = Path(entry["file_path"]) if file_path.exists(): file_path.unlink()
total_size -= entry["file_size"] del self._index[video_id]
async def pre_fetch_upcoming(self, tracks: list[Track]): for track in tracks[: self.prefetch_count]: if track.source_id not in self._index: try: self.add_to_cache( track.source_id, track.webpage_url, track.title, track.duration, ) except Exception: continue
@property def size_bytes(self) -> int: return sum(e["file_size"] for e in self._index.values())