112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
from collections import OrderedDict
|
|
from collections.abc import Callable
|
|
import gzip
|
|
import logging
|
|
from threading import RLock
|
|
|
|
from config.main import config
|
|
from api.core.Response import response
|
|
|
|
|
|
logger = logging.getLogger('dcalc.character_response')
|
|
|
|
CHARACTER_RESPONSE_CACHE_VERSION = 'v1'
|
|
CHARACTER_RESPONSE_LOCAL_CACHE_MAX_BYTES = 16 * 1024 * 1024
|
|
|
|
CharacterInfoLoader = Callable[[], dict]
|
|
CharacterResponseCacheKey = tuple[str, str, str]
|
|
|
|
_local_cache: OrderedDict[CharacterResponseCacheKey, bytes] = OrderedDict()
|
|
_local_cache_size = 0
|
|
_local_cache_lock = RLock()
|
|
|
|
|
|
def character_response_encoding(accept_encoding: str | None) -> str:
|
|
"""Match the application's GZipMiddleware negotiation behavior."""
|
|
return 'gzip' if 'gzip' in (accept_encoding or '').lower() else 'identity'
|
|
|
|
|
|
def character_response_cache_key(alter: str, equ_version: str, encoding: str) -> str:
|
|
return f'dcalc:character-response:{CHARACTER_RESPONSE_CACHE_VERSION}:{alter}:{equ_version}:{encoding}'
|
|
|
|
|
|
def serialize_character_response(info: dict, encoding: str) -> bytes:
|
|
content = response(data=info).model_dump_json().encode('utf-8')
|
|
if encoding == 'gzip':
|
|
return gzip.compress(content, compresslevel=9, mtime=0)
|
|
return content
|
|
|
|
|
|
def clear_character_response_local_cache() -> None:
|
|
global _local_cache_size
|
|
with _local_cache_lock:
|
|
_local_cache.clear()
|
|
_local_cache_size = 0
|
|
|
|
|
|
def _get_local_cache(key: CharacterResponseCacheKey) -> bytes | None:
|
|
with _local_cache_lock:
|
|
content = _local_cache.pop(key, None)
|
|
if content is not None:
|
|
_local_cache[key] = content
|
|
return content
|
|
|
|
|
|
def _set_local_cache(key: CharacterResponseCacheKey, content: bytes) -> None:
|
|
global _local_cache_size
|
|
if len(content) > CHARACTER_RESPONSE_LOCAL_CACHE_MAX_BYTES:
|
|
return
|
|
|
|
with _local_cache_lock:
|
|
previous = _local_cache.pop(key, None)
|
|
if previous is not None:
|
|
_local_cache_size -= len(previous)
|
|
_local_cache[key] = content
|
|
_local_cache_size += len(content)
|
|
|
|
while _local_cache_size > CHARACTER_RESPONSE_LOCAL_CACHE_MAX_BYTES:
|
|
_, evicted = _local_cache.popitem(last=False)
|
|
_local_cache_size -= len(evicted)
|
|
|
|
|
|
def _get_redis_cache(redis, key: str) -> bytes | None:
|
|
if config.DEBUG_MODE or redis is None:
|
|
return None
|
|
try:
|
|
content = redis.get(key)
|
|
return bytes(content) if content is not None else None
|
|
except Exception:
|
|
logger.warning('read character response cache failed key=%s', key, exc_info=True)
|
|
return None
|
|
|
|
|
|
def _set_redis_cache(redis, key: str, content: bytes) -> None:
|
|
if config.DEBUG_MODE or redis is None:
|
|
return
|
|
try:
|
|
redis.set(key, content)
|
|
except Exception:
|
|
logger.warning('write character response cache failed key=%s', key, exc_info=True)
|
|
|
|
|
|
def get_character_response_bytes(
|
|
redis,
|
|
alter: str,
|
|
equ_version: str,
|
|
encoding: str,
|
|
load_info: CharacterInfoLoader,
|
|
) -> bytes:
|
|
local_key = (alter, equ_version, encoding)
|
|
content = _get_local_cache(local_key)
|
|
if content is not None:
|
|
return content
|
|
|
|
redis_key = character_response_cache_key(alter, equ_version, encoding)
|
|
content = _get_redis_cache(redis, redis_key)
|
|
if content is None:
|
|
content = serialize_character_response(load_info(), encoding)
|
|
_set_redis_cache(redis, redis_key, content)
|
|
|
|
_set_local_cache(local_key, content)
|
|
return content
|