""" 图片元数据服务 负责管理生成图片的元数据，支持绑定到角色/聊天的特定楼层 数据持久化到 data/image_metadata 目录 """ import json import uuid from pathlib import Path from typing import List, Dict, Optional, Any from datetime import datetime try: from backend.models.internal import ImageMetadata from backend.core.config import settings except ImportError: from models.internal import ImageMetadata from core.config import settings class ImageMetadataService: """ 图片元数据服务 功能： - 记录生成图片的元数据 - 按角色/聊天/楼层组织 - 支持 swipe（同一楼层多张图片） - 提供画廊查询接口 """ def __init__(self): self.metadata_dir = settings.DATA_PATH / "image_metadata" self.metadata_dir.mkdir(parents=True, exist_ok=True) # 图片存储目录 self.images_dir = settings.DATA_PATH / "images" self.images_dir.mkdir(parents=True, exist_ok=True) def _get_chat_metadata_file(self, chat_id: str) -> Path: """获取指定聊天的元数据文件路径""" # chat_id 格式: role_name/chat_name parts = chat_id.split("/") if len(parts) == 2: role_name, chat_name = parts role_dir = self.metadata_dir / role_name role_dir.mkdir(parents=True, exist_ok=True) return role_dir / f"{chat_name}.json" else: # fallback return self.metadata_dir / f"{chat_id.replace('/', '_')}.json" def _load_chat_metadata(self, chat_id: str) -> List[ImageMetadata]: """加载指定聊天的所有图片元数据""" file_path = self._get_chat_metadata_file(chat_id) if not file_path.exists(): return [] try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) return [ImageMetadata(**item) for item in data] except Exception as e: print(f"[ImageMetadata] 加载元数据失败: {e}") return [] def _save_chat_metadata(self, chat_id: str, metadata_list: List[ImageMetadata]): """保存聊天的所有图片元数据""" file_path = self._get_chat_metadata_file(chat_id) try: data = [m.model_dump() for m in metadata_list] with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception as e: print(f"[ImageMetadata] 保存元数据失败: {e}") async def add_image( self, chat_id: str, role_name: str, chat_name: str, floor: int, filename: str, filepath: str, prompt: Optional[str] = None, negative_prompt: Optional[str] = None, seed: Optional[int] = None, model: Optional[str] = None, workflow_name: Optional[str] = None, task_id: Optional[str] = None, generation_time: Optional[float] = None, width: Optional[int] = None, height: Optional[int] = None, file_size: Optional[int] = None ) -> ImageMetadata: """ 添加图片元数据 Args: chat_id: 聊天ID role_name: 角色名称 chat_name: 聊天名称 floor: 楼层号 filename: 文件名 filepath: 文件相对路径 prompt: 提示词 negative_prompt: 负面提示词 seed: 随机种子 model: 使用的模型 workflow_name: 工作流名称 task_id: 任务ID generation_time: 生成耗时 width: 图片宽度 height: 图片高度 file_size: 文件大小 Returns: ImageMetadata: 创建的元数据 """ metadata_list = self._load_chat_metadata(chat_id) # 计算 swipe_index same_floor_images = [m for m in metadata_list if m.floor == floor] swipe_index = len(same_floor_images) # 如果这是该楼层的第一张图片，将其他图片的 isCurrentSwipe 设为 False if swipe_index == 0: for m in metadata_list: if m.floor == floor: m.isCurrentSwipe = False metadata = ImageMetadata( id=str(uuid.uuid4()), chatId=chat_id, roleName=role_name, chatName=chat_name, floor=floor, filename=filename, filepath=filepath, prompt=prompt, negativePrompt=negative_prompt, seed=seed, model=model, workflowName=workflow_name, taskId=task_id, generationTime=generation_time, width=width, height=height, fileSize=file_size, swipeIndex=swipe_index, isCurrentSwipe=True ) metadata_list.append(metadata) self._save_chat_metadata(chat_id, metadata_list) return metadata async def get_images_by_chat( self, chat_id: str, floor: Optional[int] = None ) -> List[ImageMetadata]: """ 获取指定聊天的图片列表 Args: chat_id: 聊天ID floor: 楼层号（可选，用于过滤） Returns: 图片元数据列表 """ metadata_list = self._load_chat_metadata(chat_id) if floor is not None: metadata_list = [m for m in metadata_list if m.floor == floor] # 按楼层和 swipe_index 排序 metadata_list.sort(key=lambda m: (m.floor, m.swipeIndex)) return metadata_list async def get_images_by_role(self, role_name: str) -> List[ImageMetadata]: """获取指定角色的所有图片""" all_images = [] role_dir = self.metadata_dir / role_name if not role_dir.exists(): return [] for chat_file in role_dir.glob("*.json"): chat_name = chat_file.stem chat_id = f"{role_name}/{chat_name}" images = self._load_chat_metadata(chat_id) all_images.extend(images) # 按创建时间排序 all_images.sort(key=lambda m: m.createdAt, reverse=True) return all_images async def delete_image(self, chat_id: str, image_id: str) -> bool: """ 删除图片元数据（不删除实际文件） Args: chat_id: 聊天ID image_id: 图片ID Returns: bool: 是否成功删除 """ metadata_list = self._load_chat_metadata(chat_id) # 找到要删除的图片 target_image = None for m in metadata_list: if m.id == image_id: target_image = m break if not target_image: return False floor = target_image.floor swipe_index = target_image.swipeIndex # 删除该图片 metadata_list = [m for m in metadata_list if m.id != image_id] # 重新调整同一楼层其他图片的 swipe_index same_floor_images = [m for m in metadata_list if m.floor == floor] same_floor_images.sort(key=lambda m: m.swipeIndex) for idx, m in enumerate(same_floor_images): m.swipeIndex = idx m.isCurrentSwipe = (idx == 0) # 第一个为当前显示 self._save_chat_metadata(chat_id, metadata_list) return True async def clear_chat_images(self, chat_id: str) -> int: """ 清空指定聊天的所有图片元数据 Args: chat_id: 聊天ID Returns: int: 删除的图片数量 """ metadata_list = self._load_chat_metadata(chat_id) count = len(metadata_list) # 清空元数据文件 file_path = self._get_chat_metadata_file(chat_id) if file_path.exists(): file_path.unlink() return count async def set_current_swipe(self, chat_id: str, image_id: str) -> bool: """ 设置某张图片为当前显示的 swipe Args: chat_id: 聊天ID image_id: 图片ID Returns: bool: 是否成功设置 """ metadata_list = self._load_chat_metadata(chat_id) target_image = None for m in metadata_list: if m.id == image_id: target_image = m break if not target_image: return False floor = target_image.floor # 将同一楼层的所有图片设为非当前 for m in metadata_list: if m.floor == floor: m.isCurrentSwipe = False # 设置目标图片为当前 target_image.isCurrentSwipe = True self._save_chat_metadata(chat_id, metadata_list) return True async def get_gallery_stats(self) -> Dict[str, Any]: """ 获取画廊统计信息 Returns: 统计信息字典 """ stats = { "totalImages": 0, "byRole": {}, "byChat": {} } if not self.metadata_dir.exists(): return stats for role_dir in self.metadata_dir.iterdir(): if not role_dir.is_dir(): continue role_name = role_dir.name role_count = 0 for chat_file in role_dir.glob("*.json"): chat_name = chat_file.stem chat_id = f"{role_name}/{chat_name}" images = self._load_chat_metadata(chat_id) chat_count = len(images) role_count += chat_count stats["totalImages"] += chat_count if chat_count > 0: stats["byChat"][chat_id] = chat_count if role_count > 0: stats["byRole"][role_name] = role_count return stats def get_image_full_path(self, filepath: str) -> Path: """获取图片的完整路径""" return self.images_dir / filepath # 全局实例 image_metadata_service = ImageMetadataService()