""" 静态资源磁盘缓存 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