"""Bounded on-disk cache eviction so the container's disk footprint stays lean. The app caches Emby source images, stamped posters, generated thumbnails and uploaded backgrounds under ``cache/``. Every entry is regenerated on demand, so none of it is precious — it just needs an upper bound. :func:`prune` enforces two limits, oldest-first: 1. **age** — anything older than ``max_age_days`` is removed; 2. **size** — if the cache is still over ``max_total_mb``, the oldest files are removed until it fits. All work is confined to the given cache directory and never raises (best-effort). """ from __future__ import annotations import logging import os import time from pathlib import Path logger = logging.getLogger("homelabtoolkit.cache") def _int_env(name: str, default: int) -> int: try: return int(os.environ.get(name, default)) except (TypeError, ValueError): return default # Tunable via env so the deployment can trade disk for cache warmth. MAX_AGE_DAYS = _int_env("CACHE_MAX_AGE_DAYS", 14) MAX_TOTAL_MB = _int_env("CACHE_MAX_MB", 512) SWEEP_INTERVAL_MIN = _int_env("CACHE_SWEEP_INTERVAL_MIN", 60) def prune(cache_dir: Path, *, max_age_days: int = MAX_AGE_DAYS, max_total_mb: int = MAX_TOTAL_MB) -> dict: """Evict cache files by age, then by total size. Returns a small summary.""" if not cache_dir.exists(): return {"removed": 0, "freed_mb": 0.0, "kept_mb": 0.0} now = time.time() entries: list[tuple[Path, float, int]] = [] for path in cache_dir.rglob("*"): if not path.is_file(): continue try: stat = path.stat() except OSError: continue entries.append((path, stat.st_mtime, stat.st_size)) removed = 0 freed = 0 # 1) age pass age_cutoff = now - max_age_days * 86400 survivors: list[tuple[Path, float, int]] = [] for path, mtime, size in entries: if mtime < age_cutoff: if _unlink(path): removed += 1 freed += size else: survivors.append((path, mtime, size)) # 2) size pass — drop oldest first until under budget budget = max_total_mb * 1024 * 1024 total = sum(size for _, _, size in survivors) if total > budget: survivors.sort(key=lambda entry: entry[1]) # oldest mtime first for path, _mtime, size in survivors: if total <= budget: break if _unlink(path): total -= size removed += 1 freed += size return { "removed": removed, "freed_mb": round(freed / 1024 / 1024, 1), "kept_mb": round(total / 1024 / 1024, 1), } def _unlink(path: Path) -> bool: try: path.unlink() return True except OSError: return False