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()
|