266 lines
9.5 KiB
Python
266 lines
9.5 KiB
Python
"""Disk cache for EPG and playlists, invalidated when channels.json changes."""
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from .config import CHANNELS_FILE, DATA_ROOT, MOVIES_ROOT, SERIES_ROOT
|
|
|
|
CACHE_DIR = DATA_ROOT / "cache"
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_channels_hash(channels_file: Path = CHANNELS_FILE) -> str | None:
|
|
"""
|
|
Compute SHA256 hash of channels.json content.
|
|
Returns None if file does not exist.
|
|
"""
|
|
if not channels_file.exists():
|
|
return None
|
|
try:
|
|
with open(channels_file, "rb") as f:
|
|
content = f.read()
|
|
return hashlib.sha256(content).hexdigest()
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
logger.warning("Could not hash channels file: %s", e)
|
|
return None
|
|
|
|
|
|
class CacheManager:
|
|
"""
|
|
Manages disk cache for EPG and playlists.
|
|
Cache is valid only while channels.json content stays the same.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
channels_file: Path = CHANNELS_FILE,
|
|
cache_dir: Path = CACHE_DIR,
|
|
schedule_builder=None,
|
|
):
|
|
self._channels_file = channels_file
|
|
self._cache_dir = cache_dir
|
|
self._schedule_builder = schedule_builder
|
|
|
|
def _ensure_cache_dir(self) -> bool:
|
|
"""Ensure cache dir exists. Returns False if creation fails."""
|
|
try:
|
|
self._cache_dir.mkdir(parents=True, exist_ok=True)
|
|
return True
|
|
except OSError as e:
|
|
logger.debug("Could not create cache dir %s: %s", self._cache_dir, e)
|
|
return False
|
|
|
|
def _hash_file(self) -> Path:
|
|
return self._cache_dir / "channels_hash.txt"
|
|
|
|
def _epg_cache_file(self) -> Path:
|
|
return self._cache_dir / "epg.xml"
|
|
|
|
def _get_stored_hash(self) -> str | None:
|
|
hf = self._hash_file()
|
|
if not hf.exists():
|
|
return None
|
|
try:
|
|
return hf.read_text().strip() or None
|
|
except OSError:
|
|
return None
|
|
|
|
def _write_hash(self, h: str) -> None:
|
|
self._hash_file().write_text(h)
|
|
|
|
def invalidate_if_needed(self) -> bool:
|
|
"""
|
|
If channels.json has changed, clear all caches and return True.
|
|
Otherwise return False.
|
|
Only clear when we have a stored hash that explicitly differs; when
|
|
stored is None (hash file missing), do not delete cache files — they
|
|
may be valid and we don't want to destroy them on every restart.
|
|
"""
|
|
current = get_channels_hash(self._channels_file)
|
|
stored = self._get_stored_hash()
|
|
|
|
if current is None:
|
|
return False
|
|
|
|
if stored is not None and stored != current:
|
|
logger.info("Channels changed (hash %s -> %s), invalidating cache", stored, current)
|
|
self._clear_cache()
|
|
if self._schedule_builder and hasattr(self._schedule_builder, "invalidate_cache"):
|
|
self._schedule_builder.invalidate_cache()
|
|
return True
|
|
return False
|
|
|
|
def _clear_cache(self) -> None:
|
|
"""Remove all cached files and hash."""
|
|
if not self._cache_dir.exists():
|
|
return
|
|
for f in self._cache_dir.iterdir():
|
|
try:
|
|
f.unlink()
|
|
except OSError as e:
|
|
logger.warning("Could not remove cache file %s: %s", f, e)
|
|
hf = self._hash_file()
|
|
if hf.exists():
|
|
try:
|
|
hf.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
def _ensure_hash_stored(self, current_hash: str) -> None:
|
|
"""Store current hash after a successful cache write."""
|
|
stored = self._get_stored_hash()
|
|
if stored != current_hash:
|
|
self._write_hash(current_hash)
|
|
|
|
def get_epg(self) -> str | None:
|
|
"""Return cached EPG XML if valid, else None."""
|
|
self.invalidate_if_needed()
|
|
current = get_channels_hash(self._channels_file)
|
|
if current is None:
|
|
return None
|
|
epg_path = self._epg_cache_file()
|
|
if not epg_path.exists():
|
|
return None
|
|
stored = self._get_stored_hash()
|
|
if stored != current:
|
|
return None
|
|
try:
|
|
return epg_path.read_text(encoding="utf-8")
|
|
except OSError as e:
|
|
logger.warning("Could not read EPG cache: %s", e)
|
|
return None
|
|
|
|
def set_epg(self, xml: str) -> None:
|
|
"""Save EPG XML to cache and update stored hash."""
|
|
current = get_channels_hash(self._channels_file)
|
|
if current is None or not self._ensure_cache_dir():
|
|
return
|
|
try:
|
|
self._epg_cache_file().write_text(xml, encoding="utf-8")
|
|
self._ensure_hash_stored(current)
|
|
except OSError as e:
|
|
logger.warning("Could not write EPG cache: %s", e)
|
|
|
|
def _playlist_cache_path(self, channel_id: str) -> Path:
|
|
safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in channel_id)
|
|
return self._cache_dir / f"playlist_{safe_id}.json"
|
|
|
|
def _schedule_cache_path(self, channel_id: str) -> Path:
|
|
safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in channel_id)
|
|
return self._cache_dir / f"schedule_{safe_id}.json"
|
|
|
|
def get_playlist(self, channel_id: str) -> list[tuple[str, str]] | None:
|
|
"""
|
|
Return cached playlist data as [(display_name, path_str), ...] if valid.
|
|
path_str is the path for stream?path= (e.g. movies/foo or series/bar).
|
|
"""
|
|
self.invalidate_if_needed()
|
|
current = get_channels_hash(self._channels_file)
|
|
if current is None:
|
|
return None
|
|
path = self._playlist_cache_path(channel_id)
|
|
if not path.exists():
|
|
return None
|
|
stored = self._get_stored_hash()
|
|
if stored != current:
|
|
return None
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return [tuple(item) for item in data]
|
|
except (OSError, json.JSONDecodeError, TypeError) as e:
|
|
logger.warning("Could not read playlist cache for %s: %s", channel_id, e)
|
|
return None
|
|
|
|
def set_playlist(self, channel_id: str, data: list[tuple[str, str]]) -> None:
|
|
"""Save playlist data to cache."""
|
|
current = get_channels_hash(self._channels_file)
|
|
if current is None or not self._ensure_cache_dir():
|
|
return
|
|
path = self._playlist_cache_path(channel_id)
|
|
try:
|
|
path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
self._ensure_hash_stored(current)
|
|
except OSError as e:
|
|
logger.warning("Could not write playlist cache for %s: %s", channel_id, e)
|
|
|
|
def get_schedule(self, channel_id: str) -> tuple[list[Path], list[float], list[dict]] | None:
|
|
"""
|
|
Return cached schedule (paths, durations, path_configs) if valid.
|
|
"""
|
|
self.invalidate_if_needed()
|
|
current = get_channels_hash(self._channels_file)
|
|
if current is None:
|
|
return None
|
|
path = self._schedule_cache_path(channel_id)
|
|
if not path.exists():
|
|
return None
|
|
stored = self._get_stored_hash()
|
|
if stored != current:
|
|
return None
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError, TypeError) as e:
|
|
logger.warning("Could not read schedule cache for %s: %s", channel_id, e)
|
|
return None
|
|
path_strs = data.get("path_strs")
|
|
durations = data.get("durations")
|
|
path_configs = data.get("path_configs")
|
|
if not path_strs or not durations or len(path_strs) != len(durations):
|
|
return None
|
|
paths = []
|
|
for ps in path_strs:
|
|
if not isinstance(ps, str) or ".." in ps:
|
|
return None
|
|
if ps.startswith("movies/"):
|
|
p = MOVIES_ROOT / ps[7:]
|
|
elif ps.startswith("series/"):
|
|
p = SERIES_ROOT / ps[7:]
|
|
else:
|
|
return None
|
|
if not p.exists():
|
|
return None
|
|
paths.append(p)
|
|
if not path_configs or len(path_configs) != len(paths):
|
|
return None
|
|
return (paths, [float(d) for d in durations], path_configs)
|
|
|
|
def set_schedule(
|
|
self,
|
|
channel_id: str,
|
|
paths: list[Path],
|
|
durations: list[float],
|
|
path_configs: list[dict],
|
|
) -> None:
|
|
"""Save schedule to cache."""
|
|
current = get_channels_hash(self._channels_file)
|
|
if current is None or not self._ensure_cache_dir():
|
|
return
|
|
path_strs = []
|
|
for p in paths:
|
|
try:
|
|
if MOVIES_ROOT in p.parents:
|
|
path_strs.append("movies/" + str(p.relative_to(MOVIES_ROOT)).replace("\\", "/"))
|
|
else:
|
|
path_strs.append("series/" + str(p.relative_to(SERIES_ROOT)).replace("\\", "/"))
|
|
except ValueError:
|
|
path_strs.append("movies/" + p.name)
|
|
path = self._schedule_cache_path(channel_id)
|
|
try:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"path_strs": path_strs,
|
|
"durations": durations,
|
|
"path_configs": path_configs,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
self._ensure_hash_stored(current)
|
|
except OSError as e:
|
|
logger.warning("Could not write schedule cache for %s: %s", channel_id, e)
|