's'

web2_youtube_encoder_cache.py

@@ -0,0 +1,63 @@
+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