347 lines
10 KiB
Python
347 lines
10 KiB
Python
"""
|
|
图片元数据服务
|
|
|
|
负责管理生成图片的元数据，支持绑定到角色/聊天的特定楼层
|
|
数据持久化到 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()
|