64 lines
1.7 KiB
Python
64 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_CACHE_TTL_SECONDS = max(3600, int(os.environ.get("YOUTUBE_ENCODER_CACHE_TTL", str(24 * 3600))))
|
|
|
|
|
|
def _cache_path(base_dir: str | Path | None = None) -> Path:
|
|
root = Path(base_dir or os.environ.get("LIVE_PROJECT_ROOT") or Path(__file__).resolve().parent)
|
|
path = root / "runtime"
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path / "encoder_cache.json"
|
|
|
|
|
|
def load_encoder_cache(base_dir: str | Path | None = None) -> dict[str, Any]:
|
|
path = _cache_path(base_dir)
|
|
if not path.is_file():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def get_cached_encoder(
|
|
machine: str,
|
|
ffmpeg_bin: str,
|
|
*,
|
|
base_dir: str | Path | None = None,
|
|
) -> str | None:
|
|
blob = load_encoder_cache(base_dir)
|
|
key = f"{machine}|{ffmpeg_bin}"
|
|
row = blob.get(key)
|
|
if not isinstance(row, dict):
|
|
return None
|
|
if time.time() - float(row.get("ts") or 0) > _CACHE_TTL_SECONDS:
|
|
return None
|
|
name = str(row.get("encoder") or "").strip()
|
|
return name or None
|
|
|
|
|
|
def store_cached_encoder(
|
|
machine: str,
|
|
ffmpeg_bin: str,
|
|
encoder: str,
|
|
*,
|
|
base_dir: str | Path | None = None,
|
|
) -> None:
|
|
if not encoder:
|
|
return
|
|
path = _cache_path(base_dir)
|
|
blob = load_encoder_cache(base_dir)
|
|
key = f"{machine}|{ffmpeg_bin}"
|
|
blob[key] = {"encoder": encoder, "ts": time.time()}
|
|
try:
|
|
path.write_text(json.dumps(blob, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
|
|
except OSError:
|
|
pass
|