climperor/web/refresh_cache.py

"""Restore/save durable Web refresh state around ephemeral Actions checkouts."""

from __future__ import annotations

import argparse
import json
import os
import shutil
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from shared.http_utils import write_json_atomic
from shared.paths import DATA

ROOT = Path(__file__).resolve().parent
CACHE_DIR = ROOT / ".refresh-cache"
STATE_DIR = Path(
    os.environ.get("CLIMPEROR_REFRESH_STATE")
    or (Path.home() / ".cache" / "climperor-web-refresh")
)
CACHE_FILES = (
    "patches.json",
    "hero_stats.json",
    "hero_matches.json",
    "pro_matches.json",
    "stratz_hero_meta.json",
    "stratz_matchup_tops.json",
    "item_counter_stats.json",
    "leaderboards.json",
    "hero_items.json",
    "items_meta.json",
    "hero_item_fears.json",
    "hero_abilities.json",
    "item_shop.json",
)
STREAMER_RUNTIME_KEYS = (
    "nickname",
    "unique_id",
    "signature",
    "following_count",
    "follower_count",
    "total_favorited",
    "avatar",
    "profile_fetched_at",
    "is_live",
    "live_probed_at",
)


def _copy(src: Path, dst: Path) -> None:
    dst.parent.mkdir(parents=True, exist_ok=True)
    tmp = dst.with_name(f".{dst.name}.restore.tmp")
    shutil.copy2(src, tmp)
    tmp.replace(dst)


def _payload_time(path: Path) -> datetime | None:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        raw = payload.get("fetched_at") or (payload.get("meta") or {}).get("fetched_at")
        if not raw:
            return None
        return datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
    except (OSError, ValueError, TypeError, json.JSONDecodeError):
        return None


def _cache_is_at_least_as_new(cached: Path, checkout: Path) -> bool:
    if not checkout.is_file():
        return True
    cached_at = _payload_time(cached)
    checkout_at = _payload_time(checkout)
    if cached_at is not None and checkout_at is not None:
        return cached_at >= checkout_at
    # Ignored runtime files have no checkout copy; for tracked generated files,
    # prefer Git when freshness cannot be established.
    return cached_at is not None


def restore_streamers(cached_path: Path, checkout_path: Path) -> None:
    """Merge runtime fields without replacing newly committed manual rows."""
    if not cached_path.is_file() or not checkout_path.is_file():
        return
    if not _cache_is_at_least_as_new(cached_path, checkout_path):
        return
    try:
        cached = json.loads(cached_path.read_text(encoding="utf-8"))
        checkout = json.loads(checkout_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return
    cached_by_id = {
        str(row.get("id")): row
        for row in (cached.get("streamers") or [])
        if isinstance(row, dict) and row.get("id")
    }
    for row in checkout.get("streamers") or []:
        if not isinstance(row, dict):
            continue
        old = cached_by_id.get(str(row.get("id")))
        if not old:
            continue
        for key in STREAMER_RUNTIME_KEYS:
            if key in old:
                row[key] = old[key]
    if cached.get("fetched_at"):
        checkout["fetched_at"] = cached["fetched_at"]
    write_json_atomic(checkout_path, checkout)


def restore_patches(cached_path: Path, checkout_path: Path) -> None:
    """Merge fetched patch details without reverting newer committed entries."""
    if not cached_path.is_file():
        return
    try:
        cached = json.loads(cached_path.read_text(encoding="utf-8"))
        checkout = (
            json.loads(checkout_path.read_text(encoding="utf-8"))
            if checkout_path.is_file()
            else {}
        )
    except (OSError, json.JSONDecodeError):
        return

    patches_by_version: dict[str, dict] = {}
    for source in (cached, checkout):
        for row in source.get("patches") or []:
            if isinstance(row, dict) and row.get("version"):
                patches_by_version[str(row["version"])] = row
    patches = sorted(
        patches_by_version.values(),
        key=lambda row: int(row.get("timestamp") or 0),
        reverse=True,
    )
    details = dict(cached.get("details") or {})
    details.update(checkout.get("details") or {})
    lookup: dict[str, dict] = {}
    lookup_groups = set(cached.get("lookup") or {}) | set(
        checkout.get("lookup") or {}
    )
    for group in lookup_groups:
        merged = dict((cached.get("lookup") or {}).get(group) or {})
        merged.update((checkout.get("lookup") or {}).get(group) or {})
        lookup[group] = merged
    meta = dict(cached.get("meta") or {})
    meta.update(checkout.get("meta") or {})
    meta["count"] = len(patches)
    meta["details_count"] = len(details)
    write_json_atomic(
        checkout_path,
        {"meta": meta, "patches": patches, "lookup": lookup, "details": details},
    )


def restore() -> None:
    source_dir = (
        STATE_DIR
        if STATE_DIR.is_dir() and any(STATE_DIR.glob("*.json"))
        else CACHE_DIR
    )
    restored = 0
    for name in CACHE_FILES:
        if name == "patches.json":
            continue
        src = source_dir / name
        if src.is_file() and _cache_is_at_least_as_new(src, DATA / name):
            _copy(src, DATA / name)
            restored += 1
    restore_patches(source_dir / "patches.json", DATA / "patches.json")
    restore_streamers(source_dir / "streamers.json", DATA / "streamers.json")
    print(f"refresh cache restored: {restored} generated files from {source_dir}")


def save() -> None:
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    saved = 0
    for name in (*CACHE_FILES, "streamers.json"):
        src = DATA / name
        if src.is_file():
            _copy(src, CACHE_DIR / name)
            _copy(src, STATE_DIR / name)
            saved += 1
    print(f"refresh cache saved: {saved} files to {STATE_DIR}")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=("restore", "save"))
    args = parser.parse_args()
    restore() if args.action == "restore" else save()


if __name__ == "__main__":
    main()