iptv/M3U8/scrapers/utils/cache.py

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.parent.mkdir(parents=True, exist_ok=True)

    file.write_text(json.dumps(data, indent=2), encoding="utf-8")