e

M3U8/scrapers/utils/caching.py

@@ -0,0 +1,65 @@
+import json
+from datetime import datetime
+from pathlib import Path
+
+from .config import Time
+
+
+class Cache:
+    def __init__(self, file: Path, exp: int | float) -> None:
+        self.file = file
+        self.exp = exp
+
+    @staticmethod
+    def near_hr(dt: datetime) -> float:
+        return dt.replace(minute=0, second=0, microsecond=0).timestamp()
+
+    def is_fresh(
+        self,
+        entry: dict,
+        nearest_hr: bool,
+    ) -> bool:
+        ts: float | int = entry.get("timestamp", 31496400)
+
+        if nearest_hr:
+            ts = self.near_hr(Time.from_ts(ts))
+
+        return Time.now().timestamp() - ts < self.exp
+
+    def load(
+        self,
+        nearest_hr: bool = False,
+        per_entry: bool = True,
+    ) -> dict[str, dict[str, str | float]]:
+        try:
+            data: dict = json.loads(self.file.read_text(encoding="utf-8"))
+        except (FileNotFoundError, json.JSONDecodeError):
+            return {}
+
+        if not data:
+            return {}
+
+        if per_entry:
+            return {k: v for k, v in data.items() if self.is_fresh(v, nearest_hr)}
+
+        ts: float | int = data.get("timestamp", 31496400)
+
+        if nearest_hr:
+            ts = self.near_hr(Time.from_ts(ts))
+
+        return data if self.is_fresh({"timestamp": ts}, False) else {}
+
+    def write(self, data: dict) -> None:
+        self.file.parent.mkdir(parents=True, exist_ok=True)
+
+        self.file.write_text(
+            json.dumps(
+                data,
+                indent=2,
+                ensure_ascii=False,
+            ),
+            encoding="utf-8",
+        )
+
+
+__all__ = ["Cache"]