346 lines
12 KiB
Python
346 lines
12 KiB
Python
"""
|
|
角色卡服务 - 严格按照 internal.py 的数据结构
|
|
每个角色一个文件夹，包含 character.json、avatar.png 和 chats/
|
|
"""
|
|
import json
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
try:
|
|
from backend.models.internal import CharacterCard
|
|
from backend.core.config import settings
|
|
from backend.services.character_card_converter import CharacterCardConverter
|
|
except ImportError:
|
|
from models.internal import CharacterCard
|
|
from core.config import settings
|
|
from services.character_card_converter import CharacterCardConverter
|
|
|
|
|
|
class CharacterService:
|
|
"""角色卡管理服务"""
|
|
|
|
def __init__(self):
|
|
self.characters_dir = settings.CHARACTERS_PATH
|
|
self.converter = CharacterCardConverter()
|
|
|
|
# 确保目录存在
|
|
self.characters_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def scan_all_characters(self) -> List[CharacterCard]:
|
|
"""
|
|
扫描所有角色卡
|
|
|
|
Returns:
|
|
按 lastChatAt 排序的角色卡列表（最新的在前）
|
|
"""
|
|
characters = []
|
|
|
|
for char_folder in self.characters_dir.iterdir():
|
|
if not char_folder.is_dir():
|
|
continue
|
|
|
|
try:
|
|
character = self._load_character_from_folder(char_folder)
|
|
if character:
|
|
characters.append(character)
|
|
except Exception as e:
|
|
print(f"加载角色卡失败 {char_folder.name}: {e}")
|
|
continue
|
|
|
|
# 按最后聊天时间排序（None 排最后）
|
|
characters.sort(
|
|
key=lambda c: c.lastChatAt or 0,
|
|
reverse=True
|
|
)
|
|
|
|
return characters
|
|
|
|
def _load_character_from_folder(self, folder: Path) -> Optional[CharacterCard]:
|
|
"""
|
|
从文件夹加载角色卡
|
|
|
|
Args:
|
|
folder: 角色文件夹路径
|
|
|
|
Returns:
|
|
CharacterCard 对象或 None
|
|
"""
|
|
# 1. 读取 character.json（必须存在）
|
|
char_file = folder / "character.json"
|
|
if not char_file.exists():
|
|
return None
|
|
|
|
with open(char_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
# 2. 检查是否有 avatar.png
|
|
avatar_path = None
|
|
avatar_file = folder / "avatar.png"
|
|
if avatar_file.exists():
|
|
# 存储相对路径，用于前端访问
|
|
avatar_path = f"/api/characters/{folder.name}/avatar"
|
|
|
|
# 3. 计算最后聊天时间
|
|
last_chat_at = self._get_last_chat_timestamp(folder)
|
|
|
|
# 4. 构建 CharacterCard 对象（严格按照数据结构）
|
|
character = CharacterCard(
|
|
id=data.get('id', str(uuid.uuid4())),
|
|
name=data['name'],
|
|
description=data.get('description', ''),
|
|
personality=data.get('personality', ''),
|
|
scenario=data.get('scenario', ''),
|
|
first_mes=data.get('first_mes', ''),
|
|
mes_example=data.get('mes_example', ''),
|
|
categories=data.get('categories', []),
|
|
tags=data.get('tags', []), # ✅ 使用标签数组
|
|
worldInfoId=data.get('worldInfoId'),
|
|
outputSchema=data.get('outputSchema'),
|
|
avatarPath=avatar_path,
|
|
alternate_greetings=data.get('alternate_greetings', []),
|
|
tableMaintenancePrompt=data.get('tableMaintenancePrompt'),
|
|
imageGenerationPrompt=data.get('imageGenerationPrompt'),
|
|
tableHeaders=data.get('tableHeaders'), # ✅ 动态表格表头
|
|
tableDefaults=data.get('tableDefaults'), # ✅ 动态表格默认值
|
|
createdAt=data.get('createdAt', int(datetime.now().timestamp())),
|
|
updatedAt=data.get('updatedAt', int(datetime.now().timestamp())),
|
|
lastChatAt=last_chat_at,
|
|
isFavorite=data.get('isFavorite', False),
|
|
version=data.get('version', 1)
|
|
)
|
|
|
|
return character
|
|
|
|
def _get_last_chat_timestamp(self, char_folder: Path) -> Optional[int]:
|
|
"""
|
|
获取角色的最后聊天时间戳
|
|
|
|
通过扫描 chats 目录下所有 .jsonl 文件的修改时间
|
|
"""
|
|
chats_dir = char_folder / "chats"
|
|
if not chats_dir.exists():
|
|
return None
|
|
|
|
latest_time = None
|
|
|
|
for chat_file in chats_dir.glob("*.jsonl"):
|
|
file_mtime = int(chat_file.stat().st_mtime)
|
|
if latest_time is None or file_mtime > latest_time:
|
|
latest_time = file_mtime
|
|
|
|
return latest_time
|
|
|
|
def get_character_by_name(self, name: str) -> Optional[CharacterCard]:
|
|
"""根据角色名获取角色卡"""
|
|
char_folder = self.characters_dir / name
|
|
if not char_folder.exists():
|
|
return None
|
|
|
|
return self._load_character_from_folder(char_folder)
|
|
|
|
def create_character(self, character_data: dict) -> CharacterCard:
|
|
"""
|
|
创建新角色卡
|
|
|
|
Args:
|
|
character_data: 角色数据字典
|
|
|
|
Returns:
|
|
创建的 CharacterCard 对象
|
|
"""
|
|
# 生成唯一ID
|
|
if 'id' not in character_data:
|
|
character_data['id'] = str(uuid.uuid4())
|
|
|
|
# 设置时间戳
|
|
now = int(datetime.now().timestamp())
|
|
character_data['createdAt'] = now
|
|
character_data['updatedAt'] = now
|
|
character_data['lastChatAt'] = None
|
|
|
|
# 创建文件夹
|
|
char_name = character_data['name']
|
|
char_folder = self.characters_dir / char_name
|
|
char_folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 创建 chats 目录
|
|
chats_dir = char_folder / "chats"
|
|
chats_dir.mkdir(exist_ok=True)
|
|
|
|
# 保存 character.json
|
|
char_file = char_folder / "character.json"
|
|
with open(char_file, 'w', encoding='utf-8') as f:
|
|
json.dump(character_data, f, ensure_ascii=False, indent=2)
|
|
|
|
return self._load_character_from_folder(char_folder)
|
|
|
|
def update_character(self, name: str, updates: dict) -> CharacterCard:
|
|
"""
|
|
更新角色卡
|
|
|
|
Args:
|
|
name: 角色名（旧名称，用于定位文件夹）
|
|
updates: 更新的字段（可以包含 name 字段来重命名）
|
|
|
|
Returns:
|
|
更新后的 CharacterCard 对象
|
|
"""
|
|
char_folder = self.characters_dir / name
|
|
char_file = char_folder / "character.json"
|
|
|
|
if not char_file.exists():
|
|
raise FileNotFoundError(f"角色卡不存在: {name}")
|
|
|
|
# 读取现有数据
|
|
with open(char_file, 'r', encoding='utf-8') as f:
|
|
existing_data = json.load(f)
|
|
|
|
# 检查是否需要重命名
|
|
new_name = updates.get('name')
|
|
needs_rename = new_name and new_name != name
|
|
|
|
if needs_rename:
|
|
# 验证新名称是否合法
|
|
if not new_name or new_name.strip() == '':
|
|
raise ValueError("角色名不能为空")
|
|
|
|
# 检查新名称是否已存在
|
|
new_folder = self.characters_dir / new_name
|
|
if new_folder.exists():
|
|
raise FileExistsError(f"角色 '{new_name}' 已存在")
|
|
|
|
# 重命名文件夹
|
|
try:
|
|
import shutil
|
|
shutil.move(str(char_folder), str(new_folder))
|
|
char_folder = new_folder
|
|
char_file = char_folder / "character.json"
|
|
except Exception as e:
|
|
raise RuntimeError(f"重命名文件夹失败: {str(e)}")
|
|
|
|
# 合并更新
|
|
existing_data.update(updates)
|
|
existing_data['updatedAt'] = int(datetime.now().timestamp())
|
|
|
|
# 保存
|
|
with open(char_file, 'w', encoding='utf-8') as f:
|
|
json.dump(existing_data, f, ensure_ascii=False, indent=2)
|
|
|
|
return self._load_character_from_folder(char_folder)
|
|
|
|
def delete_character(self, name: str) -> bool:
|
|
"""
|
|
删除角色卡（包括所有聊天记录）
|
|
|
|
Args:
|
|
name: 角色名
|
|
|
|
Returns:
|
|
是否成功删除
|
|
"""
|
|
char_folder = self.characters_dir / name
|
|
if not char_folder.exists():
|
|
return False
|
|
|
|
import shutil
|
|
shutil.rmtree(char_folder)
|
|
return True
|
|
|
|
def save_avatar(self, name: str, image_data: bytes) -> str:
|
|
"""
|
|
保存角色头像
|
|
|
|
Args:
|
|
name: 角色名
|
|
image_data: 图片二进制数据
|
|
|
|
Returns:
|
|
头像访问路径
|
|
"""
|
|
char_folder = self.characters_dir / name
|
|
avatar_file = char_folder / "avatar.png"
|
|
|
|
with open(avatar_file, 'wb') as f:
|
|
f.write(image_data)
|
|
|
|
return f"/api/characters/{name}/avatar"
|
|
|
|
def import_from_png(self, png_data: bytes, filename: str) -> CharacterCard:
|
|
"""
|
|
从 SillyTavern PNG 导入角色卡
|
|
|
|
Args:
|
|
png_data: PNG 文件二进制数据
|
|
filename: 原始文件名
|
|
|
|
Returns:
|
|
创建的 CharacterCard 对象
|
|
"""
|
|
# 1. 提取嵌入数据
|
|
st_data = self.converter.extract_from_png(png_data)
|
|
if not st_data:
|
|
raise ValueError("PNG文件中没有嵌入角色数据")
|
|
|
|
# 2. 转换为内部格式
|
|
character = self.converter.st_to_internal(st_data)
|
|
|
|
# 3. 创建角色文件夹
|
|
char_name = character.name
|
|
char_folder = self.characters_dir / char_name
|
|
char_folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 4. 保存 PNG 作为 avatar.png
|
|
avatar_file = char_folder / "avatar.png"
|
|
with open(avatar_file, 'wb') as f:
|
|
f.write(png_data)
|
|
|
|
# 5. 保存 character.json
|
|
char_file = char_folder / "character.json"
|
|
with open(char_file, 'w', encoding='utf-8') as f:
|
|
json.dump(character.dict(), f, ensure_ascii=False, indent=2)
|
|
|
|
# 6. 创建 chats 目录
|
|
(char_folder / "chats").mkdir(exist_ok=True)
|
|
|
|
return character
|
|
|
|
def export_as_png(self, name: str) -> bytes:
|
|
"""
|
|
导出角色为 SillyTavern PNG 格式
|
|
|
|
Args:
|
|
name: 角色名
|
|
|
|
Returns:
|
|
PNG 文件二进制数据
|
|
"""
|
|
character = self.get_character_by_name(name)
|
|
if not character:
|
|
raise FileNotFoundError(f"角色 '{name}' 不存在")
|
|
|
|
# 获取头像路径
|
|
avatar_path = None
|
|
if character.avatarPath:
|
|
# 从路径中提取文件名
|
|
avatar_filename = character.avatarPath.split('/')[-1].split('?')[0]
|
|
char_folder = self.characters_dir / name
|
|
avatar_file = char_folder / avatar_filename
|
|
if avatar_file.exists():
|
|
avatar_path = str(avatar_file)
|
|
|
|
# 如果没有头像，使用默认图片
|
|
use_default = False
|
|
if not avatar_path:
|
|
default_avatar = self.characters_dir / "defult.png"
|
|
if default_avatar.exists():
|
|
avatar_path = str(default_avatar)
|
|
use_default = True
|
|
print(f"使用默认头像: {avatar_path}")
|
|
else:
|
|
print("警告: 没有找到默认头像")
|
|
|
|
# 生成 PNG
|
|
return self.converter.export_as_png(character, avatar_path, use_default_avatar=use_default)
|