iptv/M3U8/scrapers/utils/caching.py
2025-10-01 18:12:29 -04:00

62 lines
1.6 KiB
Python

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 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"]