e

M3U8/scrapers/utils/cache.py

@@ -0,0 +1,48 @@
+import json
+from datetime import datetime
+from pathlib import Path
+
+from .config import now
+
+
+def near_hr(dt: datetime) -> float:
+    return dt.replace(minute=0, second=0, microsecond=0).timestamp()
+
+
+def is_fresh(
+    entry: dict,
+    nearest_hr: bool,
+    exp: int,
+) -> bool:
+    ts: float | int = entry.get("timestamp", 31496400)
+
+    if nearest_hr:
+        ts = near_hr(datetime.fromtimestamp(ts))
+
+    return now.timestamp() - ts < exp
+
+
+def load_cache(
+    file: Path,
+    exp: int | float,
+    nearest_hr: bool = False,
+    per_entry: bool = True,
+) -> dict[str, dict[str, str | float]]:
+    try:
+        data: dict = json.loads(file.read_text(encoding="utf-8"))
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {}
+
+    if per_entry:
+        return {k: v for k, v in data.items() if is_fresh(v, nearest_hr, exp)}
+
+    ts: float | int = data.get("timestamp", 31496400)
+
+    if nearest_hr:
+        ts = near_hr(datetime.fromtimestamp(ts))
+
+    return data if now.timestamp() - ts < exp else {}
+
+
+def write_cache(file: Path, data: dict) -> None:
+    file.write_text(json.dumps(data, indent=2), encoding="utf-8")