feat: 添加缩略图缓存管理功能，包括生成、清理和统计接口

src/webui/emoji_routes.py

@@ -13,9 +13,134 @@ import os
 import hashlib
 from PIL import Image
 import io
+from pathlib import Path
+import threading
 
 logger = get_logger("webui.emoji")
 
+# ==================== 缩略图缓存配置 ====================
+# 缩略图缓存目录
+THUMBNAIL_CACHE_DIR = Path("data/emoji_thumbnails")
+# 缩略图尺寸 (宽, 高)
+THUMBNAIL_SIZE = (200, 200)
+# 缩略图质量 (WebP 格式, 1-100)
+THUMBNAIL_QUALITY = 80
+# 缓存锁，防止并发生成同一缩略图
+_thumbnail_locks: dict[str, threading.Lock] = {}
+_locks_lock = threading.Lock()
+
+
+def _get_thumbnail_lock(file_hash: str) -> threading.Lock:
+    """获取指定文件哈希的锁，用于防止并发生成同一缩略图"""
+    with _locks_lock:
+        if file_hash not in _thumbnail_locks:
+            _thumbnail_locks[file_hash] = threading.Lock()
+        return _thumbnail_locks[file_hash]
+
+
+def _ensure_thumbnail_cache_dir() -> Path:
+    """确保缩略图缓存目录存在"""
+    THUMBNAIL_CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    return THUMBNAIL_CACHE_DIR
+
+
+def _get_thumbnail_cache_path(file_hash: str) -> Path:
+    """获取缩略图缓存路径"""
+    return THUMBNAIL_CACHE_DIR / f"{file_hash}.webp"
+
+
+def _generate_thumbnail(source_path: str, file_hash: str) -> Path:
+    """
+    生成缩略图并保存到缓存目录
+    
+    Args:
+        source_path: 原图路径
+        file_hash: 文件哈希值，用作缓存文件名
+        
+    Returns:
+        缩略图路径
+        
+    Features:
+        - GIF: 提取第一帧作为缩略图
+        - 所有格式统一转为 WebP
+        - 保持宽高比缩放
+    """
+    _ensure_thumbnail_cache_dir()
+    cache_path = _get_thumbnail_cache_path(file_hash)
+    
+    # 使用锁防止并发生成同一缩略图
+    lock = _get_thumbnail_lock(file_hash)
+    with lock:
+        # 双重检查，可能在等待锁时已被其他线程生成
+        if cache_path.exists():
+            return cache_path
+        
+        try:
+            with Image.open(source_path) as img:
+                # GIF 处理：提取第一帧
+                if hasattr(img, 'n_frames') and img.n_frames > 1:
+                    img.seek(0)  # 确保在第一帧
+                
+                # 转换为 RGB/RGBA（WebP 支持透明度）
+                if img.mode in ('P', 'PA'):
+                    # 调色板模式转换为 RGBA 以保留透明度
+                    img = img.convert('RGBA')
+                elif img.mode == 'LA':
+                    img = img.convert('RGBA')
+                elif img.mode not in ('RGB', 'RGBA'):
+                    img = img.convert('RGB')
+                
+                # 创建缩略图（保持宽高比）
+                img.thumbnail(THUMBNAIL_SIZE, Image.Resampling.LANCZOS)
+                
+                # 保存为 WebP 格式
+                img.save(cache_path, 'WEBP', quality=THUMBNAIL_QUALITY, method=6)
+                
+                logger.debug(f"生成缩略图: {file_hash} -> {cache_path}")
+                
+        except Exception as e:
+            logger.warning(f"生成缩略图失败 {file_hash}: {e}，将返回原图")
+            # 生成失败时不创建缓存文件，下次会重试
+            raise
+    
+    return cache_path
+
+
+def cleanup_orphaned_thumbnails() -> tuple[int, int]:
+    """
+    清理孤立的缩略图缓存（原图已不存在的缩略图）
+    
+    Returns:
+        (清理数量, 保留数量)
+    """
+    if not THUMBNAIL_CACHE_DIR.exists():
+        return 0, 0
+    
+    # 获取所有表情包的哈希值
+    valid_hashes = set()
+    for emoji in Emoji.select(Emoji.emoji_hash):
+        valid_hashes.add(emoji.emoji_hash)
+    
+    cleaned = 0
+    kept = 0
+    
+    for cache_file in THUMBNAIL_CACHE_DIR.glob("*.webp"):
+        file_hash = cache_file.stem
+        if file_hash not in valid_hashes:
+            try:
+                cache_file.unlink()
+                cleaned += 1
+                logger.debug(f"清理孤立缩略图: {cache_file.name}")
+            except Exception as e:
+                logger.warning(f"清理缩略图失败 {cache_file.name}: {e}")
+        else:
+            kept += 1
+    
+    if cleaned > 0:
+        logger.info(f"清理孤立缩略图: 删除 {cleaned} 个，保留 {kept} 个")
+    
+    return cleaned, kept
+
 # 模块级别的类型别名（解决 B008 ruff 错误）
 EmojiFile = Annotated[UploadFile, File(description="表情包图片文件")]
 EmojiFiles = Annotated[List[UploadFile], File(description="多个表情包图片文件")]
@@ -472,18 +597,26 @@ async def get_emoji_thumbnail(
     token: Optional[str] = Query(None, description="访问令牌"),
     maibot_session: Optional[str] = Cookie(None),
     authorization: Optional[str] = Header(None),
+    original: bool = Query(False, description="是否返回原图"),
 ):
     """
-    获取表情包缩略图
+    获取表情包缩略图（懒加载生成 + 缓存）
 
     Args:
         emoji_id: 表情包ID
         token: 访问令牌（通过 query parameter，用于向后兼容）
         maibot_session: Cookie 中的 token
         authorization: Authorization header
+        original: 是否返回原图（用于详情页查看原图）
 
     Returns:
-        表情包图片文件
+        表情包缩略图（WebP 格式）或原图
+        
+    Features:
+        - 懒加载：首次请求时生成缩略图
+        - 缓存：后续请求直接返回缓存
+        - GIF 支持：提取第一帧作为缩略图
+        - 格式统一：所有缩略图统一为 WebP 格式
     """
     try:
         token_manager = get_token_manager()
@@ -513,19 +646,55 @@ async def get_emoji_thumbnail(
         if not os.path.exists(emoji.full_path):
             raise HTTPException(status_code=404, detail="表情包文件不存在")
 
-        # 根据格式设置 MIME 类型
+        # 如果请求原图，直接返回原文件
-        mime_types = {
+        if original:
-            "png": "image/png",
+            mime_types = {
-            "jpg": "image/jpeg",
+                "png": "image/png",
-            "jpeg": "image/jpeg",
+                "jpg": "image/jpeg",
-            "gif": "image/gif",
+                "jpeg": "image/jpeg",
-            "webp": "image/webp",
+                "gif": "image/gif",
-            "bmp": "image/bmp",
+                "webp": "image/webp",
-        }
+                "bmp": "image/bmp",
+            }
+            media_type = mime_types.get(emoji.format.lower(), "application/octet-stream")
+            return FileResponse(
+                path=emoji.full_path, 
+                media_type=media_type, 
+                filename=f"{emoji.emoji_hash}.{emoji.format}"
+            )
 
-        media_type = mime_types.get(emoji.format.lower(), "application/octet-stream")
+        # 尝试获取或生成缩略图
-
+        cache_path = _get_thumbnail_cache_path(emoji.emoji_hash)
-        return FileResponse(path=emoji.full_path, media_type=media_type, filename=f"{emoji.emoji_hash}.{emoji.format}")
+        
+        # 检查缓存是否存在
+        if not cache_path.exists():
+            try:
+                # 生成缩略图
+                _generate_thumbnail(emoji.full_path, emoji.emoji_hash)
+            except Exception as e:
+                # 生成失败，回退到原图
+                logger.warning(f"缩略图生成失败，返回原图: {e}")
+                mime_types = {
+                    "png": "image/png",
+                    "jpg": "image/jpeg",
+                    "jpeg": "image/jpeg",
+                    "gif": "image/gif",
+                    "webp": "image/webp",
+                    "bmp": "image/bmp",
+                }
+                media_type = mime_types.get(emoji.format.lower(), "application/octet-stream")
+                return FileResponse(
+                    path=emoji.full_path, 
+                    media_type=media_type, 
+                    filename=f"{emoji.emoji_hash}.{emoji.format}"
+                )
+        
+        # 返回缩略图
+        return FileResponse(
+            path=str(cache_path), 
+            media_type="image/webp", 
+            filename=f"{emoji.emoji_hash}_thumb.webp"
+        )
 
     except HTTPException:
         raise
@@ -877,3 +1046,228 @@ async def batch_upload_emoji(
     except Exception as e:
         logger.exception(f"批量上传表情包失败: {e}")
         raise HTTPException(status_code=500, detail=f"批量上传失败: {str(e)}") from e
+
+
+# ==================== 缩略图缓存管理 API ====================
+
+
+class ThumbnailCacheStatsResponse(BaseModel):
+    """缩略图缓存统计响应"""
+    
+    success: bool
+    cache_dir: str
+    total_count: int
+    total_size_mb: float
+    emoji_count: int
+    coverage_percent: float
+
+
+class ThumbnailCleanupResponse(BaseModel):
+    """缩略图清理响应"""
+    
+    success: bool
+    message: str
+    cleaned_count: int
+    kept_count: int
+
+
+class ThumbnailPreheatResponse(BaseModel):
+    """缩略图预热响应"""
+    
+    success: bool
+    message: str
+    generated_count: int
+    skipped_count: int
+    failed_count: int
+
+
+@router.get("/thumbnail-cache/stats", response_model=ThumbnailCacheStatsResponse)
+async def get_thumbnail_cache_stats(
+    maibot_session: Optional[str] = Cookie(None),
+    authorization: Optional[str] = Header(None),
+):
+    """
+    获取缩略图缓存统计信息
+    
+    Returns:
+        缓存目录、缓存数量、总大小、覆盖率等统计信息
+    """
+    try:
+        verify_auth_token(maibot_session, authorization)
+        
+        _ensure_thumbnail_cache_dir()
+        
+        # 统计缓存文件
+        cache_files = list(THUMBNAIL_CACHE_DIR.glob("*.webp"))
+        total_count = len(cache_files)
+        total_size = sum(f.stat().st_size for f in cache_files)
+        total_size_mb = round(total_size / (1024 * 1024), 2)
+        
+        # 统计表情包总数
+        emoji_count = Emoji.select().count()
+        
+        # 计算覆盖率
+        coverage_percent = round((total_count / emoji_count * 100) if emoji_count > 0 else 0, 1)
+        
+        return ThumbnailCacheStatsResponse(
+            success=True,
+            cache_dir=str(THUMBNAIL_CACHE_DIR.absolute()),
+            total_count=total_count,
+            total_size_mb=total_size_mb,
+            emoji_count=emoji_count,
+            coverage_percent=coverage_percent,
+        )
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception(f"获取缩略图缓存统计失败: {e}")
+        raise HTTPException(status_code=500, detail=f"获取统计失败: {str(e)}") from e
+
+
+@router.post("/thumbnail-cache/cleanup", response_model=ThumbnailCleanupResponse)
+async def cleanup_thumbnail_cache(
+    maibot_session: Optional[str] = Cookie(None),
+    authorization: Optional[str] = Header(None),
+):
+    """
+    清理孤立的缩略图缓存（原图已删除的表情包对应的缩略图）
+    
+    Returns:
+        清理结果
+    """
+    try:
+        verify_auth_token(maibot_session, authorization)
+        
+        cleaned, kept = cleanup_orphaned_thumbnails()
+        
+        return ThumbnailCleanupResponse(
+            success=True,
+            message=f"清理完成：删除 {cleaned} 个孤立缓存，保留 {kept} 个有效缓存",
+            cleaned_count=cleaned,
+            kept_count=kept,
+        )
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception(f"清理缩略图缓存失败: {e}")
+        raise HTTPException(status_code=500, detail=f"清理失败: {str(e)}") from e
+
+
+@router.post("/thumbnail-cache/preheat", response_model=ThumbnailPreheatResponse)
+async def preheat_thumbnail_cache(
+    limit: int = Query(100, ge=1, le=1000, description="最多预热数量"),
+    maibot_session: Optional[str] = Cookie(None),
+    authorization: Optional[str] = Header(None),
+):
+    """
+    预热缩略图缓存（提前生成未缓存的缩略图）
+    
+    优先处理使用次数高的表情包
+    
+    Args:
+        limit: 最多预热数量 (1-1000)
+        
+    Returns:
+        预热结果
+    """
+    try:
+        verify_auth_token(maibot_session, authorization)
+        
+        _ensure_thumbnail_cache_dir()
+        
+        # 获取使用次数最高的表情包（未缓存的优先）
+        emojis = (
+            Emoji.select()
+            .where(Emoji.is_banned == False)  # noqa: E712  Peewee ORM requires == for boolean comparison
+            .order_by(Emoji.usage_count.desc())
+            .limit(limit * 2)  # 多查一些，因为有些可能已缓存
+        )
+        
+        generated = 0
+        skipped = 0
+        failed = 0
+        
+        for emoji in emojis:
+            if generated >= limit:
+                break
+                
+            cache_path = _get_thumbnail_cache_path(emoji.emoji_hash)
+            
+            # 已缓存，跳过
+            if cache_path.exists():
+                skipped += 1
+                continue
+            
+            # 原文件不存在，跳过
+            if not os.path.exists(emoji.full_path):
+                failed += 1
+                continue
+            
+            try:
+                _generate_thumbnail(emoji.full_path, emoji.emoji_hash)
+                generated += 1
+            except Exception as e:
+                logger.warning(f"预热缩略图失败 {emoji.emoji_hash}: {e}")
+                failed += 1
+        
+        return ThumbnailPreheatResponse(
+            success=True,
+            message=f"预热完成：生成 {generated} 个，跳过 {skipped} 个已缓存，失败 {failed} 个",
+            generated_count=generated,
+            skipped_count=skipped,
+            failed_count=failed,
+        )
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception(f"预热缩略图缓存失败: {e}")
+        raise HTTPException(status_code=500, detail=f"预热失败: {str(e)}") from e
+
+
+@router.delete("/thumbnail-cache/clear", response_model=ThumbnailCleanupResponse)
+async def clear_all_thumbnail_cache(
+    maibot_session: Optional[str] = Cookie(None),
+    authorization: Optional[str] = Header(None),
+):
+    """
+    清空所有缩略图缓存（下次访问时会重新生成）
+    
+    Returns:
+        清理结果
+    """
+    try:
+        verify_auth_token(maibot_session, authorization)
+        
+        if not THUMBNAIL_CACHE_DIR.exists():
+            return ThumbnailCleanupResponse(
+                success=True,
+                message="缓存目录不存在，无需清理",
+                cleaned_count=0,
+                kept_count=0,
+            )
+        
+        cleaned = 0
+        for cache_file in THUMBNAIL_CACHE_DIR.glob("*.webp"):
+            try:
+                cache_file.unlink()
+                cleaned += 1
+            except Exception as e:
+                logger.warning(f"删除缓存文件失败 {cache_file.name}: {e}")
+        
+        logger.info(f"已清空缩略图缓存: 删除 {cleaned} 个文件")
+        
+        return ThumbnailCleanupResponse(
+            success=True,
+            message=f"已清空所有缩略图缓存：删除 {cleaned} 个文件",
+            cleaned_count=cleaned,
+            kept_count=0,
+        )
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception(f"清空缩略图缓存失败: {e}")
+        raise HTTPException(status_code=500, detail=f"清空失败: {str(e)}") from e