LitePan/cache/cache_manager.py

"""全局缓存管理器：LRU + TTL + 持久化 + 内存压力感知。"""

import asyncio
import time
import psutil
import contextvars
import json
import os
from collections import OrderedDict
from datetime import datetime, timedelta
from typing import Any, Dict, Optional, Union
from core.base import FileItem

from core.log_manager import get_writer, LogModule

from .cache_types import (
    CacheItem, CacheConfig, CacheStats, CacheType, CacheStatus,
    CacheConstants, AccountId, FileId, ParentId
)
from .cache_keys import CacheKeyGenerator, CacheKeyValidator


def get_cache_logger():
    # 延迟初始化避免与 log_manager 循环依赖
    try:
        return get_writer(LogModule.CACHE)
    except RuntimeError:
        return None

def log_cache(level: str, message: str):
    logger = get_cache_logger()
    if logger:
        getattr(logger, level)(message)


session_id_context: contextvars.ContextVar[str] = contextvars.ContextVar('session_id', default='default')

def set_session_context(session_id: str):
    session_id_context.set(session_id)

def get_session_context() -> str:
    return session_id_context.get()


def _serialize_file_item(file_item: FileItem) -> Dict[str, Any]:
    return {
        "__type__": "FileItem",
        "id": file_item.id,
        "name": file_item.name,
        "path": file_item.path,
        "size": file_item.size,
        "is_dir": file_item.is_dir,
        "modified": file_item.modified.isoformat() if file_item.modified else None,
        "created": file_item.created.isoformat() if file_item.created else None,
        "download_url": file_item.download_url,
        "thumbnail_url": file_item.thumbnail_url,
        "mime_type": file_item.mime_type,
        "extra": file_item.extra or {}
    }


def _deserialize_file_item(data: Dict[str, Any]) -> FileItem:
    modified = datetime.fromisoformat(data["modified"]) if data.get("modified") else None
    created = datetime.fromisoformat(data["created"]) if data.get("created") else None
    return FileItem(
        id=data.get("id", ""),
        name=data.get("name", ""),
        path=data.get("path", ""),
        size=data.get("size", 0),
        is_dir=data.get("is_dir", False),
        modified=modified,
        created=created,
        download_url=data.get("download_url"),
        thumbnail_url=data.get("thumbnail_url"),
        mime_type=data.get("mime_type"),
        extra=data.get("extra") or {}
    )


def _serialize_cache_value(value: Any) -> Any:
    if isinstance(value, FileItem):
        return _serialize_file_item(value)
    if isinstance(value, list):
        return [_serialize_cache_value(item) for item in value]
    if isinstance(value, dict):
        return {k: _serialize_cache_value(v) for k, v in value.items()}
    return value


def _deserialize_cache_value(value: Any) -> Any:
    if isinstance(value, dict):
        if value.get("__type__") == "FileItem":
            return _deserialize_file_item(value)
        return {k: _deserialize_cache_value(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_deserialize_cache_value(item) for item in value]
    return value


class GlobalCacheManager:
    def __init__(self, config: Optional[CacheConfig] = None):
        self.config = config or CacheConfig()
        self.cache: OrderedDict[str, CacheItem] = OrderedDict()
        self.stats = CacheStats()
        self.lock = asyncio.Lock()

        self._account_config_cache: Dict[str, Dict] = {}
        self._config_cache_time: Dict[str, datetime] = {}

        self._last_memory_check = time.time()
        self._memory_check_interval = 60

        self._persistence_save_task: Optional[asyncio.Task] = None
        self._last_persistence_save = None
        self._persistence_interval_seconds = 300

        self._init_persistence()

    def _init_persistence(self):
        if self.config.enable_persistence and self.config.persistence_path:
            os.makedirs(self.config.persistence_path, exist_ok=True)
            log_cache("info", f"缓存持久化已启用，路径: {self.config.persistence_path}")

    def _is_cache_globally_enabled(self) -> bool:
        try:
            from config import config_manager
            cache_enabled = config_manager.get('cache_enabled')
            result = cache_enabled if cache_enabled is not None else True
            log_cache("debug", f"全局缓存开关检查: cache_enabled={cache_enabled}, result={result}")
            return result
        except Exception as e:
            log_cache("error", f"检查全局缓存开关失败: {e}")
            return True

    def _extract_account_id_from_key(self, key: str) -> Optional[str]:
        try:
            from .cache_keys import CacheKeyValidator
            return CacheKeyValidator.extract_account_id(key)
        except Exception:
            return None

    async def _get_account_cache_ttl(self, account_id: str) -> Optional[int]:
        """账号级 TTL（秒）：0=该账号禁用缓存；None=落到全局默认。"""
        normalized_account_id = str(account_id or "").strip()
        if not normalized_account_id or normalized_account_id == "temp_test":
            return None
        # 临时探测账号（如 tv_bind_quark_probe）不是 DB 数字 id，跳过账号级 TTL
        if not normalized_account_id.isdigit():
            return None

        try:
            from database.db import db

            account = await db.get_account(int(normalized_account_id))
            if account:
                config = account['config']
                if isinstance(config, str):
                    import json
                    config = json.loads(config)
                cache_ttl_minutes = config.get('cache_ttl')

                if cache_ttl_minutes is not None:
                    # 数据库里存的是分钟，内部统一按秒用
                    cache_ttl_seconds = int(cache_ttl_minutes) * 60
                    return cache_ttl_seconds
                else:
                    return None
            else:
                return None
        except Exception as e:
            log_cache("error", f"获取账号{normalized_account_id}缓存TTL失败: {e}")
            return None

    async def get(self, key: str, default: Any = None) -> Any:
        if not self._is_cache_globally_enabled():
            cache_type = self._get_cache_type_from_key(key)
            self.stats.record_miss(cache_type)
            log_cache("debug", f"全局缓存已禁用，跳过缓存获取: {key}")
            return default
        
        account_id = self._extract_account_id_from_key(key)
        if account_id:
            account_ttl = await self._get_account_cache_ttl(account_id)
            if account_ttl is not None and account_ttl == 0:
                cache_type = self._get_cache_type_from_key(key)
                self.stats.record_miss(cache_type)
                log_cache("debug", f"账号{account_id}缓存已禁用(TTL=0)，跳过缓存获取: {key}")
                return default

        async with self.lock:
            if key not in self.cache:
                cache_type = self._get_cache_type_from_key(key)
                self.stats.record_miss(cache_type)
                return default

            item = self.cache[key]

            if item.is_expired():
                del self.cache[key]
                self.stats.expirations += 1
                self.stats.record_miss(item.cache_type)
                self._update_stats_on_remove(item)
                return default

            item.touch()

            # LRU：命中后挪到尾部
            self.cache.move_to_end(key)

            self.stats.record_hit(item.cache_type)

            return item.value

    async def set(self, key: str, value: Any, ttl: Optional[int] = None,
                  cache_type: CacheType = CacheType.DIRECTORY) -> bool:
        """写入缓存；ttl=None 会按 cache_type + 账号配置推导实际 TTL，ttl=0 跳过写入。"""
        if not self._is_cache_globally_enabled():
            log_cache("debug", f"全局缓存已禁用，跳过缓存存储: {key}")
            return True

        account_id = self._extract_account_id_from_key(key)
        if account_id:
            account_ttl = await self._get_account_cache_ttl(account_id)
            if account_ttl is not None and account_ttl == 0:
                log_cache("debug", f"账号{account_id}缓存已禁用(TTL=0)，跳过缓存存储: {key}")
                return True
        
        try:
            if ttl is None:
                if cache_type in [CacheType.DOWNLOAD_MODE, CacheType.ACCOUNT_CONFIG]:
                    ttl = self.config.get_ttl_for_type(cache_type)
                else:
                    account_id = self._extract_account_id_from_key(key)
                    if account_id:
                        account_ttl = await self._get_account_cache_ttl(account_id)
                        if account_ttl is not None:
                            ttl = account_ttl
                        else:
                            try:
                                from config import config_manager
                                global_ttl = config_manager.get('cache_ttl') or 3600
                                ttl = global_ttl
                            except Exception:
                                ttl = self.config.default_ttl
                    else:
                        try:
                            from config import config_manager
                            global_ttl = config_manager.get('cache_ttl') or 3600
                            ttl = global_ttl
                        except Exception:
                            ttl = self.config.default_ttl

            if ttl == 0:
                log_cache("debug", f"TTL=0，跳过缓存存储: {key}")
                return True

            await self._check_memory_pressure()

            expires_at = datetime.now() + timedelta(seconds=ttl) if ttl > 0 else None

            item = CacheItem(
                key=key,
                value=value,
                cache_type=cache_type,
                created_at=datetime.now(),
                expires_at=expires_at
            )

            async with self.lock:
                if key in self.cache:
                    old_item = self.cache.pop(key)
                    self._update_stats_on_remove(old_item)

                if len(self.cache) >= self.config.max_items:
                    await self._evict_items()

                self.cache[key] = item
                self._update_stats_on_add(item)

            return True

        except Exception as e:
            log_cache("error", f"设置缓存失败: {key}, 错误: {str(e)}")
            return False
    
    async def delete(self, key: str) -> bool:
        async with self.lock:
            if key in self.cache:
                item = self.cache.pop(key)
                self._update_stats_on_remove(item)
                log_cache("debug", f"删除缓存项: {key}")
                return True
            return False
    
    async def clear_by_prefix(self, prefix: str) -> int:
        async with self.lock:
            keys_to_delete = [
                key for key in self.cache.keys()
                if CacheKeyValidator.match_prefix(key, prefix)
            ]
            
            count = 0
            for key in keys_to_delete:
                if key in self.cache:
                    item = self.cache.pop(key)
                    self._update_stats_on_remove(item)
                    count += 1
            
            if count > 0:
                log_cache("debug", f"按前缀清理缓存: {prefix}, 清理项目数: {count}")
            
            return count

    def _get_persistence_file_path(self) -> str:
        return os.path.join(self.config.persistence_path, "cache_data.json")

    async def save_to_disk(self) -> bool:
        if not self.config.enable_persistence:
            return False
        
        try:
            async with self.lock:
                cache_data = {}
                for key, item in self.cache.items():
                    if item.expires_at and item.expires_at < datetime.now():
                        continue
                    
                    cache_data[key] = {
                        'value': _serialize_cache_value(item.value),
                        'created_at': item.created_at.isoformat() if item.created_at else None,
                        'expires_at': item.expires_at.isoformat() if item.expires_at else None,
                        'last_access': item.last_access.isoformat() if item.last_access else None,
                        'size_bytes': item.size_bytes,
                        'cache_type': item.cache_type.value if hasattr(item.cache_type, 'value') else str(item.cache_type)
                    }
                
                file_path = self._get_persistence_file_path()
                temp_path = file_path + '.tmp'
                
                with open(temp_path, 'w', encoding='utf-8') as f:
                    json.dump({
                        'version': 1,
                        'saved_at': datetime.now().isoformat(),
                        'stats': {
                            'total_items': self.stats.total_items,
                            'total_size_bytes': self.stats.total_size_bytes,
                            'hits': self.stats.hits,
                            'misses': self.stats.misses
                        },
                        'cache': cache_data
                    }, f, ensure_ascii=False, indent=2)
                
                os.replace(temp_path, file_path)
                self._last_persistence_save = datetime.now()
                log_cache("debug", f"缓存已持久化: {len(cache_data)} 项")
                return True
                
        except Exception as e:
            log_cache("error", f"缓存持久化失败: {e}")
            return False
    
    async def load_from_disk(self) -> int:
        if not self.config.enable_persistence:
            return 0
        
        file_path = self._get_persistence_file_path()
        if not os.path.exists(file_path):
            return 0
        
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            cache_data = data.get('cache', {})
            loaded_count = 0
            
            async with self.lock:
                for key, item_data in cache_data.items():
                    try:
                        created_at = None
                        if item_data.get('created_at'):
                            created_at = datetime.fromisoformat(item_data['created_at'])
                        
                        expires_at = None
                        if item_data.get('expires_at'):
                            expires_at = datetime.fromisoformat(item_data['expires_at'])
                            if expires_at < datetime.now():
                                continue
                        
                        last_access = None
                        if item_data.get('last_access'):
                            last_access = datetime.fromisoformat(item_data['last_access'])
                        
                        cache_type = CacheType.DIRECTORY
                        type_str = item_data.get('cache_type', 'DIRECTORY')
                        try:
                            cache_type = CacheType(type_str)
                        except ValueError:
                            pass
                        
                        item = CacheItem(
                            key=key,
                            value=_deserialize_cache_value(item_data['value']),
                            cache_type=cache_type,
                            created_at=created_at,
                            expires_at=expires_at,
                            last_access=last_access,
                            size_bytes=item_data.get('size_bytes', 0)
                        )
                        self.cache[key] = item
                        self._update_stats_on_add(item)
                        loaded_count += 1

                    except Exception as e:
                        log_cache("warning", f"加载缓存项失败 {key}: {e}")
                        continue

            if loaded_count > 0:
                log_cache("info", f"缓存已从磁盘恢复: {loaded_count} 项，总大小 {self.stats.total_size_bytes / (1024*1024):.2f} MB")
            
            return loaded_count
            
        except Exception as e:
            log_cache("error", f"缓存恢复失败: {e}")
            return 0
    
    async def start_periodic_save(self, interval_seconds: int = 300):
        if not self.config.enable_persistence:
            return

        normalized_interval = max(60, int(interval_seconds or 300))
        if (
            self._persistence_save_task and
            not self._persistence_save_task.done() and
            self._persistence_interval_seconds == normalized_interval
        ):
            return

        if self._persistence_save_task and not self._persistence_save_task.done():
            await self.stop_periodic_save()

        async def save_loop():
            while True:
                try:
                    await asyncio.sleep(normalized_interval)
                    await self.save_to_disk()
                except asyncio.CancelledError:
                    break
                except Exception as e:
                    log_cache("error", f"定期保存失败: {e}")
        
        self._persistence_interval_seconds = normalized_interval
        self._persistence_save_task = asyncio.create_task(save_loop())
        log_cache("info", f"缓存定期保存已启动，间隔: {normalized_interval}秒")
    
    async def stop_periodic_save(self):
        if self._persistence_save_task:
            self._persistence_save_task.cancel()
            try:
                await self._persistence_save_task
            except asyncio.CancelledError:
                pass
            self._persistence_save_task = None
            log_cache("info", "缓存定期保存已停止")

    async def configure_persistence(self, enabled: bool, interval_seconds: int = 300) -> None:
        """热更新缓存持久化开关与保存间隔。"""
        normalized_interval = max(60, int(interval_seconds or 300))
        self.config.enable_persistence = bool(enabled)
        self._persistence_interval_seconds = normalized_interval

        if not self.config.enable_persistence:
            await self.stop_periodic_save()
            return

        await self.start_periodic_save(normalized_interval)

    async def enforce_item_limit(self) -> int:
        """按当前 max_items 强制裁剪条目，返回淘汰数量。

        典型用于用户在全局设置中调小 max_items 后立即收缩缓存。
        """
        async with self.lock:
            if not self.cache:
                return 0
            limit = max(0, int(self.config.max_items))
            excess = len(self.cache) - limit
            if excess <= 0:
                return 0
            await self._evict_items(force_count=excess)
            log_cache(
                "info",
                f"已根据新上限收缩缓存：淘汰 {excess} 项，剩余 {len(self.cache)} 项 / 上限 {limit}"
            )
            return excess

    async def clear_all(self) -> int:
        async with self.lock:
            count = len(self.cache)
            self.cache.clear()
            self.stats = CacheStats()
            log_cache("info", f"清理所有缓存, 清理项目数: {count}")
            return count

    async def get_directory_cache(self, account_id: AccountId, parent_id: ParentId,
                                 page: int = 1, search: Optional[str] = None) -> Optional[Dict]:
        key = CacheKeyGenerator.directory_key(account_id, parent_id, page, search)
        return await self.get(key)

    async def set_directory_cache(self, account_id: AccountId, parent_id: ParentId,
                                 data: Dict, page: int = 1, search: Optional[str] = None,
                                 ttl: Optional[int] = None) -> bool:
        key = CacheKeyGenerator.directory_key(account_id, parent_id, page, search)

        if ttl is None:
            ttl = self.config.get_ttl_for_type(CacheType.DIRECTORY)

        return await self.set(key, data, ttl, CacheType.DIRECTORY)

    async def get_dir_folder_sizes_cache(self, account_id: AccountId, parent_id: ParentId) -> Optional[Dict]:
        key = CacheKeyGenerator.dir_folder_sizes_key(account_id, parent_id)
        cached = await self.get(key)
        return cached if isinstance(cached, dict) else None

    async def set_dir_folder_sizes_cache(
        self,
        account_id: AccountId,
        parent_id: ParentId,
        data: Dict,
        ttl: Optional[int] = None,
    ) -> bool:
        key = CacheKeyGenerator.dir_folder_sizes_key(account_id, parent_id)
        if ttl is None:
            ttl = self.config.get_ttl_for_type(CacheType.DIRECTORY)
        return await self.set(key, data, ttl, CacheType.DIRECTORY)

    async def get_file_info_cache(self, account_id: AccountId, file_id: FileId) -> Optional[Dict]:
        key = CacheKeyGenerator.file_info_key(account_id, file_id)
        return await self.get(key)

    async def set_file_info_cache(self, account_id: AccountId, file_id: FileId,
                                 data: Dict, ttl: Optional[int] = None) -> bool:
        key = CacheKeyGenerator.file_info_key(account_id, file_id)
        return await self.set(key, data, ttl, CacheType.FILE_INFO)

    async def get_path_mapping_cache(self, account_id: AccountId, path: str) -> Optional[Any]:
        key = CacheKeyGenerator.path_mapping_key(account_id, path)
        return await self.get(key)

    async def set_path_mapping_cache(self, account_id: AccountId, path: str, value: Any, ttl: Optional[int] = None) -> bool:
        key = CacheKeyGenerator.path_mapping_key(account_id, path)
        return await self.set(key, value, ttl, CacheType.PATH_MAPPING)

    async def get_webdav_metadata_cache(self, account_id: AccountId, path: str) -> Optional[Any]:
        key = CacheKeyGenerator.webdav_metadata_key(account_id, path)
        return await self.get(key)

    async def set_webdav_metadata_cache(self, account_id: AccountId, path: str, value: Any, ttl: Optional[int] = None) -> bool:
        key = CacheKeyGenerator.webdav_metadata_key(account_id, path)
        return await self.set(key, value, ttl, CacheType.WEBDAV_METADATA)

    async def get_download_mode_cache(self, account_id: AccountId) -> Optional[str]:
        key = CacheKeyGenerator.download_mode_key(account_id)
        return await self.get(key)

    async def set_download_mode_cache(self, account_id: AccountId, mode: str,
                                     ttl: Optional[int] = None) -> bool:
        key = CacheKeyGenerator.download_mode_key(account_id)
        return await self.set(key, mode, ttl, CacheType.DOWNLOAD_MODE)

    async def get_download_url_cache(self, account_id: AccountId, file_id: FileId) -> Optional[str]:
        key = CacheKeyGenerator.download_url_key(account_id, file_id)
        return await self.get(key)

    async def set_download_url_cache(self, account_id: AccountId, file_id: FileId,
                                     url: str, ttl: Optional[int] = None) -> bool:
        key = CacheKeyGenerator.download_url_key(account_id, file_id)
        return await self.set(key, url, ttl, CacheType.DOWNLOAD_URL)

    async def clear_download_url_cache(self, account_id: AccountId, file_id: FileId):
        key = CacheKeyGenerator.download_url_key(account_id, file_id)
        await self.delete(key)

    async def clear_account_download_url_cache(self, account_id: AccountId):
        prefix = CacheConstants.KEY_PREFIX_DOWNLOAD_URL + CacheConstants.KEY_SEPARATOR + str(account_id)
        await self.clear_by_prefix(prefix)

    async def clear_download_mode_cache(self, account_id: AccountId):
        cache_key = CacheKeyGenerator.download_mode_key(account_id)
        await self.delete(cache_key)
        log_cache("debug", f"已清除账号{account_id}的下载模式缓存")

    async def clear_account_cache(self, account_id: AccountId):
        prefixes = [
            CacheKeyGenerator.directory_prefix(account_id),
            CacheKeyGenerator.dir_folder_sizes_prefix(account_id),
            CacheKeyGenerator.file_info_prefix(account_id),
            CacheKeyGenerator.path_mapping_prefix(account_id),
            CacheKeyGenerator.webdav_metadata_prefix(account_id),
        ]
        for prefix in prefixes:
            await self.clear_by_prefix(prefix)

        await self.clear_download_mode_cache(account_id)

        config_key = f"account_config_{account_id}"
        if config_key in self._account_config_cache:
            del self._account_config_cache[config_key]
        if config_key in self._config_cache_time:
            del self._config_cache_time[config_key]

        log_cache("debug", f"已清除账号{account_id}的所有缓存")

    async def clear_file_operation_cache(self, account_id: AccountId, parent_id: ParentId,
                                       file_id: Optional[FileId] = None):
        prefixes = CacheKeyGenerator.get_related_keys_for_file_operation(
            account_id, parent_id, file_id
        )

        for prefix in prefixes:
            # 完整 key（≥2 个分隔符）走 delete，前缀走 clear_by_prefix
            if prefix.count(CacheConstants.KEY_SEPARATOR) >= 2:
                await self.delete(prefix)
            else:
                await self.clear_by_prefix(prefix)

        log_cache("debug", f"已清除账号{account_id}文件操作相关缓存")

    async def clear_account_update_cache(self, account_id: AccountId):
        config_key = f"account_config_{account_id}"
        if config_key in self._account_config_cache:
            del self._account_config_cache[config_key]
        if config_key in self._config_cache_time:
            del self._config_cache_time[config_key]

        await self.clear_download_mode_cache(account_id)

        log_cache("info", f"已清除账号{account_id}的更新相关缓存")

    async def _check_memory_pressure(self):
        """周期性内存压力检查：只对「缓存自身体积」做处置，系统内存高只 warn 不动 cache。

        理由：系统内存往往被 Emby、Docker 等其他进程占用，砍掉自家 cache 既救不回系统内存，
        又会拖垮 115 这类风控严的网盘（缓存保持任务白跑）。所以只看自身 memory_limit_mb。
        """
        current_time = time.time()
        if current_time - self._last_memory_check < self._memory_check_interval:
            return

        self._last_memory_check = current_time

        try:
            cache_memory_mb = self.stats.memory_usage_mb
            limit_mb = self.config.memory_limit_mb

            if cache_memory_mb > limit_mb:
                log_cache("warning", f"缓存超限触发紧急清理: {cache_memory_mb:.1f}MB > {limit_mb}MB")
                await self._emergency_cleanup()
            elif cache_memory_mb > limit_mb * 0.8:
                log_cache("info", f"缓存使用接近上限: {cache_memory_mb:.1f}MB / {limit_mb}MB，预防性 LRU 淘汰")
                async with self.lock:
                    await self._evict_items(force_count=min(100, max(1, len(self.cache) // 10)))

            try:
                memory = psutil.virtual_memory()
                memory_usage_percent = memory.percent / 100.0
                if memory_usage_percent > CacheConstants.MEMORY_CRITICAL_THRESHOLD:
                    log_cache(
                        "warning",
                        f"系统内存压力过高: {memory_usage_percent:.1%}"
                        f"（缓存当前 {cache_memory_mb:.1f}MB / {limit_mb}MB，未触发清理）"
                    )
            except Exception:
                pass

        except Exception as e:
            log_cache("error", f"检查内存压力失败: {str(e)}")

    async def _emergency_cleanup(self):
        """缓存超限：按 LRU 顺序弹到 memory_limit_mb * 0.7，比"砍一半"更精准。"""
        target_bytes = int(self.config.memory_limit_mb * 1024 * 1024 * 0.7)
        before_mb = self.stats.memory_usage_mb
        evicted = 0

        async with self.lock:
            while self.stats.total_size_bytes > target_bytes and self.cache:
                _, item = self.cache.popitem(last=False)  # OrderedDict 头部 = 最旧 LRU
                self._update_stats_on_remove(item)
                self.stats.record_eviction(item.cache_type, memory_pressure=True)
                evicted += 1

        after_mb = self.stats.memory_usage_mb
        log_cache(
            "info",
            f"紧急清理完成: 淘汰 {evicted} 项，{before_mb:.1f}MB → {after_mb:.1f}MB"
            f"（目标 {self.config.memory_limit_mb * 0.7:.0f}MB）"
        )

    async def _evict_items(self, force_count: Optional[int] = None):
        """常规 LRU 淘汰；调用方需自行持有 self.lock（避免和 set/紧急清理嵌套加锁死锁）。"""
        if not self.cache:
            return

        if force_count is None:
            if len(self.cache) >= self.config.max_items:
                force_count = max(1, len(self.cache) // 10)
            else:
                return

        evicted_count = 0
        for _ in range(force_count):
            if not self.cache:
                break
            _, item = self.cache.popitem(last=False)  # OrderedDict 头部 = 最旧 LRU
            self._update_stats_on_remove(item)
            self.stats.record_eviction(item.cache_type, memory_pressure=True)
            evicted_count += 1

        if evicted_count > 0:
            log_cache(
                "debug",
                f"LRU 淘汰 {evicted_count} 项（剩余 {len(self.cache)} 项 / {self.stats.memory_usage_mb:.1f}MB）"
            )

    def _update_stats_on_add(self, item: CacheItem):
        self.stats.total_items += 1
        self.stats.total_size_bytes += item.size_bytes

        if item.cache_type in self.stats.type_stats:
            self.stats.type_stats[item.cache_type]['items'] += 1
            self.stats.type_stats[item.cache_type]['size_bytes'] += item.size_bytes

    def _update_stats_on_remove(self, item: CacheItem):
        self.stats.total_items = max(0, self.stats.total_items - 1)
        self.stats.total_size_bytes = max(0, self.stats.total_size_bytes - item.size_bytes)

        if item.cache_type in self.stats.type_stats:
            self.stats.type_stats[item.cache_type]['items'] = max(0,
                self.stats.type_stats[item.cache_type]['items'] - 1)
            self.stats.type_stats[item.cache_type]['size_bytes'] = max(0,
                self.stats.type_stats[item.cache_type]['size_bytes'] - item.size_bytes)

    def _get_cache_type_from_key(self, key: str) -> CacheType:
        cache_type_str = CacheKeyValidator.extract_cache_type(key)

        type_mapping = {
            CacheConstants.KEY_PREFIX_DIRECTORY: CacheType.DIRECTORY,
            CacheConstants.KEY_PREFIX_DIR_FOLDER_SIZES: CacheType.DIRECTORY,
            CacheConstants.KEY_PREFIX_FILE_INFO: CacheType.FILE_INFO,
            CacheConstants.KEY_PREFIX_PATH: CacheType.PATH_MAPPING,
            CacheConstants.KEY_PREFIX_WEBDAV: CacheType.WEBDAV_METADATA,
            CacheConstants.KEY_PREFIX_DOWNLOAD_MODE: CacheType.DOWNLOAD_MODE,
            CacheConstants.KEY_PREFIX_ACCOUNT_CONFIG: CacheType.ACCOUNT_CONFIG,
        }

        return type_mapping.get(cache_type_str, CacheType.DIRECTORY)

    def get_stats(self) -> CacheStats:
        return self.stats

    def get_cache_info(self) -> Dict[str, Any]:
        return {
            'total_items': len(self.cache),
            'memory_usage_mb': self.stats.memory_usage_mb,
            'hit_rate': self.stats.hit_rate,
            'config': {
                'max_items': self.config.max_items,
                'memory_limit_mb': self.config.memory_limit_mb,
                'default_ttl': self.config.default_ttl,
            },
            'type_stats': self.stats.type_stats
        }

    def should_use_cache(self, ttl: Optional[int] = None, cache_type: CacheType = CacheType.DIRECTORY) -> bool:
        """ttl=0 表示不走缓存。"""
        if ttl is None:
            ttl = self.config.get_ttl_for_type(cache_type)

        return ttl > 0