1054 lines
38 KiB
Python
1054 lines
38 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import time
|
|
import weakref
|
|
from collections import OrderedDict
|
|
from collections.abc import Awaitable, Callable, Iterable
|
|
from contextvars import ContextVar, Token
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from astrbot.api import logger
|
|
|
|
from .sqlite import AsyncSQLiteDB
|
|
|
|
API_ENDPOINTS: tuple[str, ...] = (
|
|
"/active/calendar",
|
|
"/active/celebs",
|
|
"/arena/awesome",
|
|
"/arena/recent",
|
|
"/arena/schools",
|
|
"/auction/records",
|
|
"/battle/records",
|
|
"/card/cached",
|
|
"/card/random",
|
|
"/card/records",
|
|
"/castle/status",
|
|
"/chat/records",
|
|
"/chitu/records",
|
|
"/chitu/week/records",
|
|
"/duowan/statistics",
|
|
"/event/collect",
|
|
"/event/missing",
|
|
"/event/recent",
|
|
"/event/records",
|
|
"/event/statistics",
|
|
"/exam/search",
|
|
"/fenxian/records",
|
|
"/firework/records",
|
|
"/food/list",
|
|
"/fraud/detail",
|
|
"/home/flower",
|
|
"/home/furniture",
|
|
"/home/travel",
|
|
"/mech/decrypt",
|
|
"/mentor/search",
|
|
"/monster/records",
|
|
"/monster/weekly",
|
|
"/news/announce",
|
|
"/news/records",
|
|
"/raid/records",
|
|
"/ranch/chat",
|
|
"/ranch/records",
|
|
"/rank/arena",
|
|
"/rank/championship",
|
|
"/rank/constable",
|
|
"/rank/outlaw",
|
|
"/rank/statistics",
|
|
"/rank/trials",
|
|
"/rank/wanted",
|
|
"/recruit/search",
|
|
"/reward/statistics",
|
|
"/role/achievement",
|
|
"/role/detail",
|
|
"/sand/records",
|
|
"/saohua/answer",
|
|
"/saohua/content",
|
|
"/saohua/context",
|
|
"/saohua/drink",
|
|
"/saohua/eat",
|
|
"/saohua/random",
|
|
"/saohua/zhanan",
|
|
"/school/matrix",
|
|
"/school/seniority",
|
|
"/school/skills",
|
|
"/school/talent",
|
|
"/server/status/check",
|
|
"/skill/rework",
|
|
"/steed/records",
|
|
"/tieba/item/records",
|
|
"/tieba/random",
|
|
"/trade/demon",
|
|
"/trade/manufacture",
|
|
"/trade/records",
|
|
"/trade/wanbaolou",
|
|
"/tuilan/achievement",
|
|
"/wicked/records",
|
|
)
|
|
|
|
|
|
class CacheService:
|
|
"""持久化接口 JSON 与 HTML 渲染图片，并提供 WebUI 配置。"""
|
|
|
|
DEFAULT_API_TTL = 300
|
|
DEFAULT_IMAGE_TTL = 600
|
|
MAX_TTL_SECONDS = 30 * 24 * 60 * 60
|
|
DEFAULT_MAX_MEMORY_BYTES = 16 * 1024 * 1024
|
|
DEFAULT_MAX_API_ENTRIES = 256
|
|
DEFAULT_MAX_IMAGE_BYTES = 512 * 1024 * 1024
|
|
MAX_MEMORY_MB_LIMIT = 1024
|
|
MAX_API_ENTRIES_LIMIT = 100_000
|
|
MAX_IMAGE_MB_LIMIT = 10_240
|
|
CLEANUP_INTERVAL_SECONDS = 60
|
|
_SENSITIVE_KEYS = frozenset(
|
|
{"token", "ticket", "authorization", "access_token", "jx3api_token"}
|
|
)
|
|
_NO_CACHE_API_DEFAULTS = frozenset(
|
|
{
|
|
"/card/random",
|
|
"/saohua/answer",
|
|
"/saohua/content",
|
|
"/saohua/context",
|
|
"/saohua/drink",
|
|
"/saohua/eat",
|
|
"/saohua/random",
|
|
"/saohua/zhanan",
|
|
"/tieba/random",
|
|
}
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
sqlite: AsyncSQLiteDB,
|
|
image_dir: Path,
|
|
asset_roots: Iterable[Path] = (),
|
|
):
|
|
self._sqlite = sqlite
|
|
self.image_dir = Path(image_dir)
|
|
self._settings: dict[tuple[str, str], int] = {}
|
|
self.max_memory_bytes = self.DEFAULT_MAX_MEMORY_BYTES
|
|
self.max_api_entries = self.DEFAULT_MAX_API_ENTRIES
|
|
self.max_image_bytes = self.DEFAULT_MAX_IMAGE_BYTES
|
|
self._memory: OrderedDict[str, tuple[int, int, bytes, int]] = OrderedDict()
|
|
self._memory_size_bytes = 0
|
|
self._api_storage_lock = asyncio.Lock()
|
|
self._cleanup_task: asyncio.Task | None = None
|
|
self._api_locks: weakref.WeakValueDictionary[str, asyncio.Lock] = (
|
|
weakref.WeakValueDictionary()
|
|
)
|
|
self._image_locks: weakref.WeakValueDictionary[str, asyncio.Lock] = (
|
|
weakref.WeakValueDictionary()
|
|
)
|
|
self._command_context: ContextVar[tuple[str, str]] = ContextVar(
|
|
"jx3_cache_command_context",
|
|
default=("", ""),
|
|
)
|
|
self._image_names: set[str] = set()
|
|
self._asset_signature = self._build_asset_signature(asset_roots)
|
|
|
|
@staticmethod
|
|
def _build_asset_signature(roots: Iterable[Path]) -> str:
|
|
signature = hashlib.sha256()
|
|
for root_index, root in enumerate(roots):
|
|
path = Path(root)
|
|
if not path.exists():
|
|
continue
|
|
for item in sorted(
|
|
(candidate for candidate in path.rglob("*") if candidate.is_file()),
|
|
key=lambda candidate: candidate.relative_to(path).as_posix(),
|
|
):
|
|
try:
|
|
content_hash = hashlib.sha256()
|
|
with item.open("rb") as file:
|
|
while chunk := file.read(1024 * 1024):
|
|
content_hash.update(chunk)
|
|
except OSError:
|
|
continue
|
|
relative_path = item.relative_to(path).as_posix()
|
|
signature.update(str(root_index).encode("ascii"))
|
|
signature.update(b"\0")
|
|
signature.update(relative_path.encode("utf-8"))
|
|
signature.update(b"\0")
|
|
signature.update(content_hash.digest())
|
|
return signature.hexdigest()
|
|
|
|
async def initialize(self):
|
|
self.image_dir.mkdir(parents=True, exist_ok=True)
|
|
await self._sqlite.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS cache_settings(
|
|
cache_type TEXT NOT NULL,
|
|
cache_name TEXT NOT NULL,
|
|
ttl_seconds INTEGER NOT NULL,
|
|
PRIMARY KEY(cache_type, cache_name)
|
|
)
|
|
"""
|
|
)
|
|
await self._sqlite.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS cache_limits(
|
|
limit_name TEXT PRIMARY KEY,
|
|
limit_value INTEGER NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
await self._sqlite.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS api_response_cache(
|
|
cache_key TEXT PRIMARY KEY,
|
|
endpoint TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
expires_at INTEGER NOT NULL,
|
|
last_accessed_at INTEGER NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
await self._sqlite.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS image_render_cache(
|
|
cache_key TEXT PRIMARY KEY,
|
|
cache_name TEXT NOT NULL,
|
|
file_name TEXT NOT NULL,
|
|
size_bytes INTEGER NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
expires_at INTEGER NOT NULL,
|
|
last_accessed_at INTEGER NOT NULL,
|
|
message_text TEXT NOT NULL DEFAULT ''
|
|
)
|
|
"""
|
|
)
|
|
image_columns = await self._sqlite.fetch_all(
|
|
"PRAGMA table_info(image_render_cache)"
|
|
)
|
|
if "message_text" not in {str(row.get("name")) for row in image_columns}:
|
|
await self._sqlite.execute(
|
|
"""
|
|
ALTER TABLE image_render_cache
|
|
ADD COLUMN message_text TEXT NOT NULL DEFAULT ''
|
|
"""
|
|
)
|
|
await self._sqlite.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_api_cache_endpoint ON api_response_cache(endpoint)"
|
|
)
|
|
await self._sqlite.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_api_cache_lru ON api_response_cache(last_accessed_at)"
|
|
)
|
|
await self._sqlite.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_image_cache_name ON image_render_cache(cache_name)"
|
|
)
|
|
await self._load_settings()
|
|
await self._load_limits()
|
|
async with self._api_storage_lock:
|
|
await self._enforce_api_limit()
|
|
await self.cleanup_expired()
|
|
await self._enforce_image_limit()
|
|
if self._cleanup_task is None or self._cleanup_task.done():
|
|
self._cleanup_task = asyncio.create_task(
|
|
self._cleanup_loop(), name="jx3-cache-cleanup"
|
|
)
|
|
|
|
async def stop(self):
|
|
if self._cleanup_task is not None:
|
|
self._cleanup_task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._cleanup_task
|
|
self._cleanup_task = None
|
|
|
|
async def _cleanup_loop(self):
|
|
while True:
|
|
await asyncio.sleep(self.CLEANUP_INTERVAL_SECONDS)
|
|
try:
|
|
await self.cleanup_expired()
|
|
except Exception as exc:
|
|
logger.warning(f"自动清理查询缓存失败：{exc}")
|
|
|
|
async def _load_settings(self):
|
|
rows = await self._sqlite.select_all("cache_settings")
|
|
self._settings = {
|
|
(str(row["cache_type"]), str(row["cache_name"])): int(row["ttl_seconds"])
|
|
for row in rows
|
|
}
|
|
|
|
async def _load_limits(self):
|
|
rows = await self._sqlite.select_all("cache_limits")
|
|
limits = {str(row["limit_name"]): int(row["limit_value"]) for row in rows}
|
|
memory_limit_mb = self._validated_memory_limit_mb(
|
|
limits.get(
|
|
"api_memory_max_mb",
|
|
self.DEFAULT_MAX_MEMORY_BYTES // 1024 // 1024,
|
|
)
|
|
)
|
|
self.max_memory_bytes = memory_limit_mb * 1024 * 1024
|
|
self.max_api_entries = self._validated_api_entry_limit(
|
|
limits.get(
|
|
"api_max_entries",
|
|
limits.get("api_memory_entries", self.DEFAULT_MAX_API_ENTRIES),
|
|
)
|
|
)
|
|
image_limit_mb = self._validated_image_limit_mb(
|
|
limits.get("image_max_mb", self.DEFAULT_MAX_IMAGE_BYTES // 1024 // 1024)
|
|
)
|
|
self.max_image_bytes = image_limit_mb * 1024 * 1024
|
|
|
|
@classmethod
|
|
def _validated_memory_limit_mb(cls, value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
raise ValueError("接口内存缓存容量必须是整数 MB")
|
|
try:
|
|
limit = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("接口内存缓存容量必须是整数 MB") from exc
|
|
if limit < 1 or limit > cls.MAX_MEMORY_MB_LIMIT:
|
|
raise ValueError("接口内存缓存容量必须在 1 到 1024 MB 之间")
|
|
return limit
|
|
|
|
@classmethod
|
|
def _validated_api_entry_limit(cls, value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
raise ValueError("SQLite 接口缓存条数必须是整数")
|
|
try:
|
|
limit = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("SQLite 接口缓存条数必须是整数") from exc
|
|
if limit < 1 or limit > cls.MAX_API_ENTRIES_LIMIT:
|
|
raise ValueError("SQLite 接口缓存条数必须在 1 到 100000 之间")
|
|
return limit
|
|
|
|
@classmethod
|
|
def _validated_image_limit_mb(cls, value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
raise ValueError("图片缓存容量必须是整数 MB")
|
|
try:
|
|
limit = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("图片缓存容量必须是整数 MB") from exc
|
|
if limit < 1 or limit > cls.MAX_IMAGE_MB_LIMIT:
|
|
raise ValueError("图片缓存容量必须在 1 到 10240 MB 之间")
|
|
return limit
|
|
|
|
async def set_limits(
|
|
self,
|
|
api_memory_max_mb: Any,
|
|
api_max_entries: Any,
|
|
image_max_mb: Any,
|
|
):
|
|
memory_limit_mb = self._validated_memory_limit_mb(api_memory_max_mb)
|
|
api_entry_limit = self._validated_api_entry_limit(api_max_entries)
|
|
image_limit_mb = self._validated_image_limit_mb(image_max_mb)
|
|
for limit_name, limit_value in (
|
|
("api_memory_max_mb", memory_limit_mb),
|
|
("api_max_entries", api_entry_limit),
|
|
("image_max_mb", image_limit_mb),
|
|
):
|
|
await self._sqlite.execute(
|
|
"""
|
|
INSERT INTO cache_limits(limit_name, limit_value)
|
|
VALUES(?, ?)
|
|
ON CONFLICT(limit_name) DO UPDATE SET
|
|
limit_value=excluded.limit_value
|
|
""",
|
|
(limit_name, limit_value),
|
|
)
|
|
self.max_image_bytes = image_limit_mb * 1024 * 1024
|
|
async with self._api_storage_lock:
|
|
self.max_memory_bytes = memory_limit_mb * 1024 * 1024
|
|
self.max_api_entries = api_entry_limit
|
|
await self._enforce_api_limit()
|
|
await self._enforce_image_limit()
|
|
|
|
def register_image_names(self, names: Iterable[str]):
|
|
self._image_names.update(
|
|
str(name).strip() for name in names if str(name).strip()
|
|
)
|
|
|
|
def enter_command(self, command_name: str, args: Iterable[Any] = ()) -> Token:
|
|
argument_signature = hashlib.sha256(
|
|
self._json(list(args)).encode("utf-8")
|
|
).hexdigest()
|
|
return self._command_context.set(
|
|
(str(command_name or "").strip(), argument_signature)
|
|
)
|
|
|
|
def leave_command(self, token: Token):
|
|
self._command_context.reset(token)
|
|
|
|
def current_command(self) -> str:
|
|
return self._command_context.get()[0]
|
|
|
|
def current_command_signature(self) -> str:
|
|
return self._command_context.get()[1]
|
|
|
|
def _base_ttl(self, cache_type: str, cache_name: str) -> int:
|
|
if cache_type == "api" and cache_name in self._NO_CACHE_API_DEFAULTS:
|
|
return 0
|
|
# 会话避雷图片默认不缓存，避免修改记录后仍展示旧图；仍可在 WebUI 单独开启。
|
|
if cache_type == "image" and cache_name in {"避雷查看", "避雷查询"}:
|
|
return 0
|
|
return self.DEFAULT_API_TTL if cache_type == "api" else self.DEFAULT_IMAGE_TTL
|
|
|
|
def get_ttl(self, cache_type: str, cache_name: str) -> int:
|
|
specific = self._settings.get((cache_type, cache_name))
|
|
if specific is not None:
|
|
return specific
|
|
if cache_type == "api" and cache_name in self._NO_CACHE_API_DEFAULTS:
|
|
return 0
|
|
if cache_type == "image" and cache_name in {"避雷查看", "避雷查询"}:
|
|
return 0
|
|
default = self._settings.get((cache_type, "*"))
|
|
if default is not None:
|
|
return default
|
|
return self._base_ttl(cache_type, cache_name)
|
|
|
|
@classmethod
|
|
def _validate_ttl(cls, value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
raise ValueError("缓存时间必须是整数秒")
|
|
try:
|
|
ttl = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("缓存时间必须是整数秒") from exc
|
|
if ttl < 0 or ttl > cls.MAX_TTL_SECONDS:
|
|
raise ValueError("缓存时间必须在 0 到 2592000 秒之间")
|
|
return ttl
|
|
|
|
async def set_ttl(
|
|
self,
|
|
cache_type: str,
|
|
cache_name: str,
|
|
ttl_seconds: Any = None,
|
|
inherit: bool = False,
|
|
):
|
|
if cache_type not in {"api", "image"}:
|
|
raise ValueError("缓存类型仅支持 api 或 image")
|
|
cache_name = str(cache_name or "").strip()
|
|
if not cache_name:
|
|
raise ValueError("缓存项目不能为空")
|
|
if cache_name == "*" and inherit:
|
|
raise ValueError("默认缓存时间不能继承")
|
|
|
|
if inherit:
|
|
await self._sqlite.delete(
|
|
"cache_settings",
|
|
"cache_type=? AND cache_name=?",
|
|
(cache_type, cache_name),
|
|
)
|
|
self._settings.pop((cache_type, cache_name), None)
|
|
return
|
|
|
|
ttl = self._validate_ttl(ttl_seconds)
|
|
await self._sqlite.execute(
|
|
"""
|
|
INSERT INTO cache_settings(cache_type, cache_name, ttl_seconds)
|
|
VALUES(?, ?, ?)
|
|
ON CONFLICT(cache_type, cache_name) DO UPDATE SET
|
|
ttl_seconds=excluded.ttl_seconds
|
|
""",
|
|
(cache_type, cache_name, ttl),
|
|
)
|
|
self._settings[(cache_type, cache_name)] = ttl
|
|
|
|
@classmethod
|
|
def _normalized(cls, value: Any, strip_sensitive: bool = False) -> Any:
|
|
if isinstance(value, dict):
|
|
return {
|
|
str(key): cls._normalized(item, strip_sensitive)
|
|
for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
|
|
if not strip_sensitive or str(key).lower() not in cls._SENSITIVE_KEYS
|
|
}
|
|
if isinstance(value, (list, tuple)):
|
|
return [cls._normalized(item, strip_sensitive) for item in value]
|
|
if isinstance(value, (str, int, float, bool)) or value is None:
|
|
return value
|
|
return str(value)
|
|
|
|
@classmethod
|
|
def _json(cls, value: Any, strip_sensitive: bool = False) -> str:
|
|
return json.dumps(
|
|
cls._normalized(value, strip_sensitive),
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
@classmethod
|
|
def build_api_key(cls, endpoint: str, params: dict[str, Any]) -> str:
|
|
source = f"api:v1|{endpoint}|{cls._json(params, strip_sensitive=True)}"
|
|
return hashlib.sha256(source.encode("utf-8")).hexdigest()
|
|
|
|
async def _read_api_payload(
|
|
self,
|
|
cache_key: str,
|
|
endpoint: str,
|
|
) -> tuple[Any | None, int | None, int | None]:
|
|
async with self._api_storage_lock:
|
|
now = time.time()
|
|
ttl = self.get_ttl("api", endpoint)
|
|
memory = self._memory.get(cache_key)
|
|
if memory is not None:
|
|
created_at, expires_at, payload, _ = memory
|
|
else:
|
|
row = await self._sqlite.fetch_one(
|
|
"""
|
|
SELECT payload, created_at, expires_at
|
|
FROM api_response_cache
|
|
WHERE cache_key=? AND endpoint=?
|
|
""",
|
|
(cache_key, endpoint),
|
|
)
|
|
if not row:
|
|
return None, None, None
|
|
created_at = int(row["created_at"])
|
|
expires_at = int(row["expires_at"])
|
|
payload = str(row["payload"])
|
|
effective_expiry = min(expires_at, created_at + ttl)
|
|
try:
|
|
data = (
|
|
json.loads(payload) if ttl > 0 and effective_expiry > now else None
|
|
)
|
|
except json.JSONDecodeError:
|
|
data = None
|
|
if data is None:
|
|
self._forget_memory(cache_key)
|
|
await self._sqlite.delete(
|
|
"api_response_cache", "cache_key=?", (cache_key,)
|
|
)
|
|
return None, effective_expiry, created_at
|
|
|
|
self._remember(cache_key, created_at, expires_at, payload)
|
|
await self._sqlite.execute(
|
|
"UPDATE api_response_cache SET last_accessed_at=? WHERE cache_key=?",
|
|
(now, cache_key),
|
|
)
|
|
return data, effective_expiry, created_at
|
|
|
|
def _remember(
|
|
self,
|
|
cache_key: str,
|
|
created_at: int,
|
|
expires_at: int,
|
|
payload: str | bytes,
|
|
):
|
|
self._forget_memory(cache_key)
|
|
encoded = payload if isinstance(payload, bytes) else payload.encode("utf-8")
|
|
size_bytes = len(encoded)
|
|
if size_bytes > self.max_memory_bytes:
|
|
return
|
|
self._memory[cache_key] = (created_at, expires_at, encoded, size_bytes)
|
|
self._memory_size_bytes += size_bytes
|
|
self._memory.move_to_end(cache_key)
|
|
self._enforce_memory_limit()
|
|
|
|
def _forget_memory(self, cache_key: str):
|
|
removed = self._memory.pop(cache_key, None)
|
|
if removed is not None:
|
|
self._memory_size_bytes -= removed[3]
|
|
|
|
def _clear_memory(self):
|
|
self._memory.clear()
|
|
self._memory_size_bytes = 0
|
|
|
|
def _retain_memory_keys(self, retained: set[str]):
|
|
self._memory = OrderedDict(
|
|
(key, value) for key, value in self._memory.items() if key in retained
|
|
)
|
|
self._memory_size_bytes = sum(value[3] for value in self._memory.values())
|
|
|
|
def _enforce_memory_limit(self):
|
|
while self._memory_size_bytes > self.max_memory_bytes and self._memory:
|
|
_, removed = self._memory.popitem(last=False)
|
|
self._memory_size_bytes -= removed[3]
|
|
|
|
async def _enforce_api_limit(self):
|
|
"""Enforce memory bytes and SQLite entries while holding the storage lock."""
|
|
self._enforce_memory_limit()
|
|
row = await self._sqlite.fetch_one(
|
|
"SELECT COUNT(*) AS count FROM api_response_cache"
|
|
)
|
|
if int((row or {}).get("count") or 0) <= self.max_api_entries:
|
|
return
|
|
await self._sqlite.execute(
|
|
"""
|
|
DELETE FROM api_response_cache WHERE cache_key IN (
|
|
SELECT cache_key FROM api_response_cache
|
|
ORDER BY last_accessed_at DESC, rowid DESC
|
|
LIMIT -1 OFFSET ?
|
|
)
|
|
""",
|
|
(self.max_api_entries,),
|
|
)
|
|
remaining = await self._sqlite.fetch_all(
|
|
"SELECT cache_key FROM api_response_cache"
|
|
)
|
|
retained = {row["cache_key"] for row in remaining}
|
|
self._retain_memory_keys(retained)
|
|
|
|
async def _save_api_payload(
|
|
self,
|
|
cache_key: str,
|
|
endpoint: str,
|
|
data: Any,
|
|
ttl_seconds: int,
|
|
) -> int:
|
|
payload = self._json(data)
|
|
now = int(time.time())
|
|
expires_at = now + ttl_seconds
|
|
async with self._api_storage_lock:
|
|
await self._sqlite.execute(
|
|
"""
|
|
INSERT INTO api_response_cache(
|
|
cache_key, endpoint, payload, created_at, expires_at, last_accessed_at
|
|
) VALUES(?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(cache_key) DO UPDATE SET
|
|
endpoint=excluded.endpoint,
|
|
payload=excluded.payload,
|
|
created_at=excluded.created_at,
|
|
expires_at=excluded.expires_at,
|
|
last_accessed_at=excluded.last_accessed_at
|
|
""",
|
|
(cache_key, endpoint, payload, now, expires_at, time.time()),
|
|
)
|
|
self._remember(cache_key, now, expires_at, payload)
|
|
await self._enforce_api_limit()
|
|
return now
|
|
|
|
async def request_api(
|
|
self,
|
|
endpoint: str,
|
|
params: dict[str, Any],
|
|
requester: Callable[[], Awaitable[Any]],
|
|
is_cacheable: Callable[[Any], bool],
|
|
force_refresh: bool = False,
|
|
) -> tuple[Any, dict[str, Any]]:
|
|
ttl = self.get_ttl("api", endpoint)
|
|
cache_key = self.build_api_key(endpoint, params)
|
|
metadata = {
|
|
"endpoint": endpoint,
|
|
"cache_key": cache_key,
|
|
"hit": False,
|
|
"stale": False,
|
|
"ttl_seconds": ttl,
|
|
"data_hash": "",
|
|
"created_at": None,
|
|
}
|
|
if ttl <= 0:
|
|
data = await requester()
|
|
metadata["created_at"] = int(time.time())
|
|
if is_cacheable(data):
|
|
metadata["data_hash"] = hashlib.sha256(
|
|
self._json(data).encode("utf-8")
|
|
).hexdigest()
|
|
return data, metadata
|
|
|
|
if not force_refresh:
|
|
cached, _, created_at = await self._read_api_payload(cache_key, endpoint)
|
|
if cached is not None:
|
|
metadata["hit"] = True
|
|
metadata["created_at"] = created_at
|
|
metadata["data_hash"] = hashlib.sha256(
|
|
self._json(cached).encode("utf-8")
|
|
).hexdigest()
|
|
return cached, metadata
|
|
|
|
lock = self._api_locks.setdefault(cache_key, asyncio.Lock())
|
|
async with lock:
|
|
if not force_refresh:
|
|
cached, _, created_at = await self._read_api_payload(
|
|
cache_key, endpoint
|
|
)
|
|
if cached is not None:
|
|
metadata["hit"] = True
|
|
metadata["created_at"] = created_at
|
|
metadata["data_hash"] = hashlib.sha256(
|
|
self._json(cached).encode("utf-8")
|
|
).hexdigest()
|
|
return cached, metadata
|
|
|
|
data = None
|
|
try:
|
|
data = await requester()
|
|
finally:
|
|
if not is_cacheable(data):
|
|
async with self._api_storage_lock:
|
|
self._forget_memory(cache_key)
|
|
await self._sqlite.delete(
|
|
"api_response_cache", "cache_key=?", (cache_key,)
|
|
)
|
|
if is_cacheable(data):
|
|
metadata["data_hash"] = hashlib.sha256(
|
|
self._json(data).encode("utf-8")
|
|
).hexdigest()
|
|
try:
|
|
metadata["created_at"] = await self._save_api_payload(
|
|
cache_key,
|
|
endpoint,
|
|
data,
|
|
ttl,
|
|
)
|
|
except Exception as exc:
|
|
metadata["created_at"] = int(time.time())
|
|
logger.warning(f"写入接口缓存失败 endpoint={endpoint}: {exc}")
|
|
return data, metadata
|
|
return data, metadata
|
|
|
|
def build_image_key(
|
|
self,
|
|
cache_name: str,
|
|
template: str,
|
|
data: dict[str, Any],
|
|
render_options: dict[str, Any],
|
|
source_signature: str = "",
|
|
variant_signature: str = "",
|
|
) -> str:
|
|
source = "|".join(
|
|
(
|
|
"image:v2",
|
|
cache_name,
|
|
hashlib.sha256(template.encode("utf-8")).hexdigest(),
|
|
source_signature
|
|
or hashlib.sha256(self._json(data).encode("utf-8")).hexdigest(),
|
|
variant_signature,
|
|
self._json(render_options),
|
|
self._asset_signature,
|
|
)
|
|
)
|
|
return hashlib.sha256(source.encode("utf-8")).hexdigest()
|
|
|
|
@classmethod
|
|
def value_signature(cls, value: Any) -> str:
|
|
return hashlib.sha256(cls._json(value).encode("utf-8")).hexdigest()
|
|
|
|
def build_image_request_key(
|
|
self,
|
|
cache_name: str,
|
|
render_options: dict[str, Any],
|
|
variant_signature: str,
|
|
scope_signature: str = "",
|
|
) -> str:
|
|
"""生成可在请求接口前计算的最终图片缓存键。"""
|
|
source = "|".join(
|
|
(
|
|
"image-request:v2",
|
|
cache_name,
|
|
variant_signature,
|
|
scope_signature,
|
|
self._json(render_options),
|
|
self._asset_signature,
|
|
)
|
|
)
|
|
return hashlib.sha256(source.encode("utf-8")).hexdigest()
|
|
|
|
def image_lock(self, cache_key: str) -> asyncio.Lock:
|
|
return self._image_locks.setdefault(cache_key, asyncio.Lock())
|
|
|
|
async def get_image_entry(
|
|
self,
|
|
cache_key: str,
|
|
cache_name: str,
|
|
) -> tuple[Path, str] | None:
|
|
try:
|
|
ttl = self.get_ttl("image", cache_name)
|
|
if ttl <= 0:
|
|
return None
|
|
now = int(time.time())
|
|
row = await self._sqlite.fetch_one(
|
|
"""
|
|
SELECT file_name, created_at, expires_at, message_text
|
|
FROM image_render_cache
|
|
WHERE cache_key=? AND cache_name=?
|
|
""",
|
|
(cache_key, cache_name),
|
|
)
|
|
if not row:
|
|
return None
|
|
effective_expiry = min(int(row["expires_at"]), int(row["created_at"]) + ttl)
|
|
path = self.image_dir / str(row["file_name"])
|
|
if effective_expiry <= now or not path.is_file():
|
|
await self._delete_image_record(cache_key, path)
|
|
return None
|
|
await self._sqlite.execute(
|
|
"UPDATE image_render_cache SET last_accessed_at=? WHERE cache_key=?",
|
|
(now, cache_key),
|
|
)
|
|
return path, str(row.get("message_text") or "")
|
|
except Exception as exc:
|
|
logger.warning(f"读取图片缓存失败 cache={cache_name}: {exc}")
|
|
return None
|
|
|
|
async def get_image(self, cache_key: str, cache_name: str) -> Path | None:
|
|
entry = await self.get_image_entry(cache_key, cache_name)
|
|
return entry[0] if entry else None
|
|
|
|
async def save_image(
|
|
self,
|
|
cache_key: str,
|
|
cache_name: str,
|
|
source_path: str,
|
|
image_format: str,
|
|
message_text: str = "",
|
|
) -> Path | None:
|
|
ttl = self.get_ttl("image", cache_name)
|
|
source = Path(source_path)
|
|
if ttl <= 0 or not source.is_file():
|
|
return None
|
|
extension = "jpg" if image_format == "jpeg" else "png"
|
|
file_name = f"{cache_key}.{extension}"
|
|
target = self.image_dir / file_name
|
|
temporary = self.image_dir / f".{file_name}.tmp"
|
|
try:
|
|
await asyncio.to_thread(shutil.copy2, source, temporary)
|
|
await asyncio.to_thread(os.replace, temporary, target)
|
|
size_bytes = target.stat().st_size
|
|
now = int(time.time())
|
|
await self._sqlite.execute(
|
|
"""
|
|
INSERT INTO image_render_cache(
|
|
cache_key, cache_name, file_name, size_bytes,
|
|
created_at, expires_at, last_accessed_at, message_text
|
|
) VALUES(?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(cache_key) DO UPDATE SET
|
|
cache_name=excluded.cache_name,
|
|
file_name=excluded.file_name,
|
|
size_bytes=excluded.size_bytes,
|
|
created_at=excluded.created_at,
|
|
expires_at=excluded.expires_at,
|
|
last_accessed_at=excluded.last_accessed_at,
|
|
message_text=excluded.message_text
|
|
""",
|
|
(
|
|
cache_key,
|
|
cache_name,
|
|
file_name,
|
|
size_bytes,
|
|
now,
|
|
now + ttl,
|
|
now,
|
|
str(message_text or ""),
|
|
),
|
|
)
|
|
await self._enforce_image_limit()
|
|
return target
|
|
except Exception as exc:
|
|
logger.warning(f"保存图片缓存失败 cache={cache_name}: {exc}")
|
|
for path in (temporary, target):
|
|
try:
|
|
if path.exists():
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
async def _delete_image_record(self, cache_key: str, path: Path):
|
|
await self._sqlite.delete("image_render_cache", "cache_key=?", (cache_key,))
|
|
try:
|
|
if path.is_file() and path.parent.resolve() == self.image_dir.resolve():
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
async def _enforce_image_limit(self):
|
|
row = await self._sqlite.fetch_one(
|
|
"SELECT COALESCE(SUM(size_bytes), 0) AS total FROM image_render_cache"
|
|
)
|
|
total = int((row or {}).get("total") or 0)
|
|
if total <= self.max_image_bytes:
|
|
return
|
|
rows = await self._sqlite.fetch_all(
|
|
"""
|
|
SELECT cache_key, file_name, size_bytes
|
|
FROM image_render_cache
|
|
ORDER BY last_accessed_at ASC
|
|
"""
|
|
)
|
|
for item in rows:
|
|
if total <= self.max_image_bytes:
|
|
break
|
|
await self._delete_image_record(
|
|
str(item["cache_key"]),
|
|
self.image_dir / str(item["file_name"]),
|
|
)
|
|
total -= int(item["size_bytes"])
|
|
|
|
async def cleanup_expired(self):
|
|
now = int(time.time())
|
|
async with self._api_storage_lock:
|
|
uncached_defaults = sorted(self._NO_CACHE_API_DEFAULTS)
|
|
placeholders = ",".join("?" for _ in uncached_defaults)
|
|
await self._sqlite.execute(
|
|
f"""
|
|
DELETE FROM api_response_cache
|
|
WHERE expires_at<=? OR created_at + COALESCE(
|
|
(SELECT ttl_seconds FROM cache_settings
|
|
WHERE cache_type='api' AND cache_name=api_response_cache.endpoint),
|
|
CASE WHEN endpoint IN ({placeholders}) THEN 0 ELSE ? END
|
|
)<=?
|
|
""",
|
|
(now, *uncached_defaults, self.get_ttl("api", "*"), now),
|
|
)
|
|
remaining = await self._sqlite.fetch_all(
|
|
"SELECT cache_key FROM api_response_cache"
|
|
)
|
|
retained = {row["cache_key"] for row in remaining}
|
|
self._retain_memory_keys(retained)
|
|
rows = await self._sqlite.fetch_all(
|
|
"SELECT cache_key, file_name FROM image_render_cache WHERE expires_at<=?",
|
|
(now,),
|
|
)
|
|
for row in rows:
|
|
cache_key = str(row["cache_key"])
|
|
async with self.image_lock(cache_key):
|
|
expired = await self._sqlite.fetch_one(
|
|
"""
|
|
SELECT file_name FROM image_render_cache
|
|
WHERE cache_key=? AND expires_at<=?
|
|
""",
|
|
(cache_key, now),
|
|
)
|
|
if expired:
|
|
await self._delete_image_record(
|
|
cache_key, self.image_dir / str(expired["file_name"])
|
|
)
|
|
|
|
async def clear(self, cache_type: str) -> dict[str, int]:
|
|
if cache_type not in {"api", "image", "all"}:
|
|
raise ValueError("清理类型仅支持 api、image 或 all")
|
|
removed = {"api": 0, "image": 0}
|
|
if cache_type in {"api", "all"}:
|
|
async with self._api_storage_lock:
|
|
row = await self._sqlite.fetch_one(
|
|
"SELECT COUNT(*) AS count FROM api_response_cache"
|
|
)
|
|
removed["api"] = int((row or {}).get("count") or 0)
|
|
await self._sqlite.execute("DELETE FROM api_response_cache")
|
|
self._clear_memory()
|
|
if cache_type in {"image", "all"}:
|
|
rows = await self._sqlite.fetch_all(
|
|
"SELECT cache_key, file_name FROM image_render_cache"
|
|
)
|
|
removed["image"] = len(rows)
|
|
for row in rows:
|
|
await self._delete_image_record(
|
|
str(row["cache_key"]),
|
|
self.image_dir / str(row["file_name"]),
|
|
)
|
|
return removed
|
|
|
|
async def clear_item(self, cache_type: str, cache_name: str) -> int:
|
|
if cache_type not in {"api", "image"}:
|
|
raise ValueError("缓存类型仅支持 api 或 image")
|
|
cache_name = str(cache_name or "").strip()
|
|
if not cache_name:
|
|
raise ValueError("缓存项目不能为空")
|
|
|
|
if cache_type == "api":
|
|
async with self._api_storage_lock:
|
|
rows = await self._sqlite.fetch_all(
|
|
"SELECT cache_key FROM api_response_cache WHERE endpoint=?",
|
|
(cache_name,),
|
|
)
|
|
await self._sqlite.delete(
|
|
"api_response_cache",
|
|
"endpoint=?",
|
|
(cache_name,),
|
|
)
|
|
for row in rows:
|
|
self._forget_memory(str(row["cache_key"]))
|
|
return len(rows)
|
|
|
|
rows = await self._sqlite.fetch_all(
|
|
"""
|
|
SELECT cache_key, file_name
|
|
FROM image_render_cache
|
|
WHERE cache_name=?
|
|
""",
|
|
(cache_name,),
|
|
)
|
|
for row in rows:
|
|
await self._delete_image_record(
|
|
str(row["cache_key"]),
|
|
self.image_dir / str(row["file_name"]),
|
|
)
|
|
return len(rows)
|
|
|
|
def _setting_item(self, cache_type: str, cache_name: str) -> dict[str, Any]:
|
|
return {
|
|
"name": cache_name,
|
|
"ttl_seconds": self.get_ttl(cache_type, cache_name),
|
|
"overridden": (cache_type, cache_name) in self._settings,
|
|
"safe_default": (
|
|
(
|
|
(cache_type == "api" and cache_name in self._NO_CACHE_API_DEFAULTS)
|
|
or (
|
|
cache_type == "image" and cache_name in {"避雷查看", "避雷查询"}
|
|
)
|
|
)
|
|
and (cache_type, cache_name) not in self._settings
|
|
),
|
|
}
|
|
|
|
async def dashboard(self) -> dict[str, Any]:
|
|
await self.cleanup_expired()
|
|
api_row = await self._sqlite.fetch_one(
|
|
"""
|
|
SELECT COUNT(*) AS count,
|
|
COALESCE(SUM(LENGTH(CAST(payload AS BLOB))), 0) AS size_bytes
|
|
FROM api_response_cache
|
|
"""
|
|
)
|
|
image_row = await self._sqlite.fetch_one(
|
|
"""
|
|
SELECT COUNT(*) AS count,
|
|
COALESCE(SUM(size_bytes), 0) AS size_bytes
|
|
FROM image_render_cache
|
|
"""
|
|
)
|
|
known_api_names = set(API_ENDPOINTS)
|
|
known_api_names.update(
|
|
name
|
|
for cache_type, name in self._settings
|
|
if cache_type == "api" and name != "*"
|
|
)
|
|
known_image_names = set(self._image_names)
|
|
known_image_names.update(
|
|
name
|
|
for cache_type, name in self._settings
|
|
if cache_type == "image" and name != "*"
|
|
)
|
|
return {
|
|
"defaults": {
|
|
"api": self.get_ttl("api", "*"),
|
|
"image": self.get_ttl("image", "*"),
|
|
},
|
|
"limits": {
|
|
"api_memory_max_mb": self.max_memory_bytes // 1024 // 1024,
|
|
"api_max_entries": self.max_api_entries,
|
|
"image_max_mb": self.max_image_bytes // 1024 // 1024,
|
|
},
|
|
"api": [
|
|
self._setting_item("api", name) for name in sorted(known_api_names)
|
|
],
|
|
"images": [
|
|
self._setting_item("image", name)
|
|
for name in sorted(
|
|
known_image_names, key=lambda value: value.encode("utf-8")
|
|
)
|
|
],
|
|
"stats": {
|
|
"api_count": int((api_row or {}).get("count") or 0),
|
|
"api_size_bytes": int((api_row or {}).get("size_bytes") or 0),
|
|
"api_memory_count": len(self._memory),
|
|
"api_memory_size_bytes": self._memory_size_bytes,
|
|
"api_memory_limit_bytes": self.max_memory_bytes,
|
|
"api_entry_limit": self.max_api_entries,
|
|
"image_count": int((image_row or {}).get("count") or 0),
|
|
"image_size_bytes": int((image_row or {}).get("size_bytes") or 0),
|
|
"image_limit_bytes": self.max_image_bytes,
|
|
},
|
|
}
|