src/webui/services/content.py
Rama main · réplica de lectura del repositorio original.
"""Public DiceFrame content fetcher with bounded online-first caching."""
from __future__ import annotations
import asyncio
import hashlib
import json
import os
import re
import time
from pathlib import Path
from typing import Any
import aiohttp
GITHUB_CONTENT_BASE_URL = "https://raw.githubusercontent.com/diceframe/diceframe-content/main/content"
DEFAULT_BASE_URL = GITHUB_CONTENT_BASE_URL
_SUCCESS_TTL = 600
_FAILURE_TTL = 60
_TOTAL_TIMEOUT = 3
_PER_SOURCE_TIMEOUT = 1.5
_MAX_BYTES = 512 * 1024
_SAFE_PATH = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
_CACHE: dict[str, dict[str, Any]] = {}
_FAILURE_UNTIL: dict[str, float] = {}
_INFLIGHT: dict[str, asyncio.Task[str]] = {}
def base_urls() -> tuple[str, ...]:
primary = os.getenv("DICEFRAME_CONTENT_BASE_URL", DEFAULT_BASE_URL).strip().rstrip("/")
fallback = GITHUB_CONTENT_BASE_URL.rstrip("/")
urls: list[str] = []
for url in (primary, fallback):
if url and url not in urls:
urls.append(url)
return tuple(urls)
def _validate_path(path: str) -> str:
value = (path or "").strip().lstrip("/")
if not _SAFE_PATH.fullmatch(value) or ".." in value.split("/"):
raise ValueError("invalid public content path")
return value
def _cache_file(cache_dir: Path | None, path: str) -> Path | None:
if cache_dir is None:
return None
digest = hashlib.sha256(path.encode("utf-8")).hexdigest()
return cache_dir / f"{digest}.cache"
def _read_disk_cache(cache_dir: Path | None, path: str) -> str:
target = _cache_file(cache_dir, path)
if target is None:
return ""
try:
return target.read_text(encoding="utf-8")
except OSError:
return ""
def _write_disk_cache(cache_dir: Path | None, path: str, text: str) -> None:
target = _cache_file(cache_dir, path)
if target is None:
return
try:
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(text, encoding="utf-8")
except OSError:
return
def disk_cache_age_seconds(cache_dir: Path | None, path: str) -> float | None:
"""公共内容磁盘缓存文件的年龄（秒）；无缓存或不可读时返回 None。
供公告等服务判断返回正文是否来自缓存，而不是刚同步的在线内容。
"""
target = _cache_file(cache_dir, path)
if target is None:
return None
try:
return time.time() - target.stat().st_mtime
except OSError:
return None
async def fetch_text(
cache_dir: Path | None,
path: str,
*,
force_refresh: bool = False,
allow_cached: bool = True,
mirror_manager: Any | None = None,
) -> str:
"""Fetch a public content path, with optional cache bypass for legal originals."""
key = _validate_path(path)
now = time.monotonic()
cached = _CACHE.get(key)
if allow_cached and not force_refresh and cached and now - float(cached["fetched_at"]) < _SUCCESS_TTL:
return str(cached["text"])
if allow_cached and not force_refresh and now < _FAILURE_UNTIL.get(key, 0):
return str(cached["text"]) if cached else _read_disk_cache(cache_dir, key)
task = _INFLIGHT.get(key)
if task is None:
task = asyncio.create_task(
_fetch_remote(key, mirror_manager=mirror_manager)
if mirror_manager is not None
else _fetch_remote(key)
)
_INFLIGHT[key] = task
def _cleanup(done: asyncio.Task[str]) -> None:
if _INFLIGHT.get(key) is done:
_INFLIGHT.pop(key, None)
task.add_done_callback(_cleanup)
try:
text = await asyncio.shield(task)
except asyncio.CancelledError:
raise
except Exception:
text = ""
now = time.monotonic()
if text:
_CACHE[key] = {"text": text, "fetched_at": now}
_FAILURE_UNTIL.pop(key, None)
_write_disk_cache(cache_dir, key, text)
return text
_FAILURE_UNTIL[key] = now + _FAILURE_TTL
if allow_cached:
return str(cached["text"]) if cached else _read_disk_cache(cache_dir, key)
return ""
async def fetch_json(
cache_dir: Path | None,
path: str,
*,
force_refresh: bool = False,
allow_cached: bool = True,
mirror_manager: Any | None = None,
) -> dict[str, Any] | None:
text = await fetch_text(
cache_dir,
path,
force_refresh=force_refresh,
allow_cached=allow_cached,
mirror_manager=mirror_manager,
)
if not text:
return None
try:
value = json.loads(text)
except json.JSONDecodeError:
return None
return value if isinstance(value, dict) else None
async def _fetch_remote(path: str, *, mirror_manager: Any | None = None) -> str:
if mirror_manager is not None:
primary = os.getenv("DICEFRAME_CONTENT_BASE_URL", DEFAULT_BASE_URL).strip().rstrip("/")
if primary == DEFAULT_BASE_URL:
try:
result = await mirror_manager.fetch_github_url(
f"{DEFAULT_BASE_URL}/{path}",
binary=False,
max_bytes=_MAX_BYTES,
resolve=False,
)
if result.ok and isinstance(result.data, str) and result.data.strip():
return result.data
except Exception:
# Mirror configuration is an availability aid; retain the
# existing direct-source and cache fallback when it fails.
pass
timeout = aiohttp.ClientTimeout(total=_PER_SOURCE_TIMEOUT)
headers = {"User-Agent": "DiceFrame content"}
async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session:
for base in base_urls():
text = await _fetch_one(session, f"{base}/{path}")
if text:
return text
return ""
async def _fetch_one(session: aiohttp.ClientSession, url: str) -> str:
try:
async with session.get(url) as response:
if response.status != 200:
return ""
if response.content_length is not None and response.content_length > _MAX_BYTES:
return ""
chunks = bytearray()
async for chunk in response.content.iter_chunked(16 * 1024):
chunks.extend(chunk)
if len(chunks) > _MAX_BYTES:
return ""
text = bytes(chunks).decode("utf-8")
return text if text.strip() else ""
except (aiohttp.ClientError, asyncio.TimeoutError, UnicodeDecodeError):
return ""
class PublicContentService:
"""Bounded public-content fetcher rooted at one optional disk cache."""
def __init__(self, cache_dir: Path | None = None, *, mirror_manager: Any | None = None) -> None:
self._cache_dir = cache_dir
self._mirror_manager = mirror_manager
async def fetch_text(
self,
path: str,
*,
force_refresh: bool = False,
allow_cached: bool = True,
) -> str:
return await fetch_text(
self._cache_dir,
path,
force_refresh=force_refresh,
allow_cached=allow_cached,
mirror_manager=self._mirror_manager,
)
async def fetch_json(
self,
path: str,
*,
force_refresh: bool = False,
allow_cached: bool = True,
) -> dict[str, Any] | None:
return await fetch_json(
self._cache_dir,
path,
force_refresh=force_refresh,
allow_cached=allow_cached,
mirror_manager=self._mirror_manager,
)
def disk_cache_age(self, path: str) -> float | None:
return disk_cache_age_seconds(self._cache_dir, path)
def cache_file(self, path: str) -> Path | None:
return _cache_file(self._cache_dir, path)