agento3/cache.py

"""
静态资源磁盘缓存 v5 — 对标 go3

只缓存明确是静态资源的路径（CSS/JS/图片/字体）。
动态 HTML / API / 登录流程 零缓存 — 直接透传。
缓存文件以 MD5(URL) 命名，存储在 cache/ 目录。

go3 参考: staticExtensions map + MD5 cache
"""
import hashlib
import logging
import os
import threading
import time
from typing import Optional

logger = logging.getLogger("ao3-cache")

CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cache")

# Only cache these extensions — everything else passes through
STATIC_EXTENSIONS = {
    ".css", ".js", ".jpg", ".jpeg", ".png", ".gif", ".ico",
    ".woff", ".woff2", ".ttf", ".svg", ".webp",
}

# Path prefixes that are always static
STATIC_PATH_PREFIXES = (
    "/stylesheets/", "/javascripts/", "/images/", "/media/",
    "/skins/", "/assets/", "/favicon.ico",
)

# Cache TTL: 7 days for static assets (like go3's static_cache_ttl_seconds)
STATIC_TTL = 604800  # 7 days


def is_static_path(path: str) -> bool:
    """Check if a URL path points to a static asset."""
    path_lower = path.lower()
    for ext in STATIC_EXTENSIONS:
        if path_lower.endswith(ext):
            return True
    for prefix in STATIC_PATH_PREFIXES:
        if path_lower.startswith(prefix):
            return True
    return False


class StaticCache:
    """Disk-based static file cache. No LRU, no TTL on dynamic content."""

    def __init__(self):
        os.makedirs(CACHE_DIR, exist_ok=True)
        self._lock = threading.Lock()
        self._hits = 0
        self._misses = 0

    def _cache_path(self, url: str) -> str:
        h = hashlib.md5(url.encode()).hexdigest()
        return os.path.join(CACHE_DIR, h)

    def get(self, url: str) -> Optional[tuple[bytes, dict, int]]:
        """Get cached static file. Returns (body, headers, status) or None."""
        if not is_static_path(url):
            return None
        path = self._cache_path(url)
        try:
            with self._lock:
                if os.path.exists(path):
                    mtime = os.path.getmtime(path)
                    if time.time() - mtime < STATIC_TTL:
                        with open(path, "rb") as f:
                            body = f.read()
                        self._hits += 1
                        # Minimal headers for static content
                        headers = {
                            "Content-Type": _guess_content_type(url),
                            "Cache-Control": f"public, max-age={STATIC_TTL}, immutable",
                        }
                        return (body, headers, 200)
                    else:
                        os.remove(path)
        except Exception:
            pass
        self._misses += 1
        return None

    def set(self, url: str, body: bytes, headers: dict, status: int):
        """Store static file on disk."""
        if not is_static_path(url):
            return
        if status != 200:
            return
        path = self._cache_path(url)
        try:
            with self._lock:
                with open(path, "wb") as f:
                    f.write(body)
        except Exception as e:
            logger.debug(f"Cache write failed: {e}")

    def get_stats(self) -> dict:
        with self._lock:
            try:
                files = os.listdir(CACHE_DIR)
                total_size = sum(
                    os.path.getsize(os.path.join(CACHE_DIR, f))
                    for f in files
                    if os.path.isfile(os.path.join(CACHE_DIR, f))
                )
            except Exception:
                files = []
                total_size = 0
        return {
            "size": len(files),
            "capacity": "unlimited",
            "usage_pct": round(total_size / (1024 * 1024), 1),
            "hits": self._hits,
            "misses": self._misses,
        }


def _guess_content_type(url: str) -> str:
    url_lower = url.lower()
    if url_lower.endswith(".css"):
        return "text/css; charset=utf-8"
    if url_lower.endswith(".js"):
        return "application/javascript; charset=utf-8"
    if url_lower.endswith(".png"):
        return "image/png"
    if url_lower.endswith(".jpg") or url_lower.endswith(".jpeg"):
        return "image/jpeg"
    if url_lower.endswith(".gif"):
        return "image/gif"
    if url_lower.endswith(".svg"):
        return "image/svg+xml"
    if url_lower.endswith(".ico"):
        return "image/x-icon"
    if url_lower.endswith(".woff"):
        return "font/woff"
    if url_lower.endswith(".woff2"):
        return "font/woff2"
    if url_lower.endswith(".ttf"):
        return "font/ttf"
    if url_lower.endswith(".webp"):
        return "image/webp"
    return "application/octet-stream"


def get_ttl_for_path(path: str) -> int:
    """Only used for static path TTL. Returns 0 for dynamic paths."""
    return STATIC_TTL if is_static_path(path) else 0


# ─── Singleton ──────────────────────────────────────────────────────────────

_cache: Optional[StaticCache] = None


def get_cache() -> StaticCache:
    global _cache
    if _cache is None:
        _cache = StaticCache()
    return _cache