重构路由架构,修复导致的前端出错

backend/core/models/PromptComponent.py

@@ -16,7 +16,7 @@ class PromptComponent(BaseModel):
 	@validator('role')
 	def validate_role(cls, v):
 		"""验证角色值是否在有效范围内"""
-		if v not in [0, 1, 2]:
+		if not isinstance(v, int) or v not in [0, 1, 2]:
 			raise ValueError("角色值必须是0(System)、1(User)或2(Assistant)")
 		return v
 
@@ -49,7 +49,7 @@ class PromptComponent(BaseModel):
 	@classmethod
 	def from_dict(cls, data: Dict[str, Any]) -> 'PromptComponent':
 		"""
-		从字典创建组件实例
+		从字典创建组件实例，自动处理role字段的类型转换
 
 		参数:
 			data: 包含组件数据的字典
@@ -57,4 +57,9 @@ class PromptComponent(BaseModel):
 		返回:
 			PromptComponent: 组件实例
 		"""
+		# 处理role字段，将字符串转换为整数
+		if 'role' in data and isinstance(data['role'], str):
+			role_map = {'system': 0, 'user': 1, 'assistant': 2}
+			data['role'] = role_map.get(data['role'].lower(), 0)
+
 		return cls(**data)

backend/core/models/PromptList.py

@@ -1,5 +1,7 @@
 from pydantic import BaseModel, Field, validator
 from typing import List, Dict, Any, Optional
+from pathlib import Path
+import json
 from .PromptComponent import PromptComponent
 
 
@@ -69,6 +71,379 @@ class AIDesignSpec(BaseModel):
 				raise ValueError(f"prompt_order中包含不存在的组件ID: {invalid_ids}")
 		return v
 
+	@classmethod
+	def get_preset_dir(cls) -> Path:
+		"""获取预设目录路径"""
+		try:
+			from backend.core.config import settings
+			preset_dir = settings.DATA_PATH / "preset"
+			# 如果路径不存在，尝试使用相对路径
+			if not preset_dir.exists():
+				# 尝试从当前工作目录构建路径
+				cwd_preset_dir = Path.cwd() / "data" / "preset"
+				if cwd_preset_dir.exists():
+					return cwd_preset_dir
+				# 尝试从脚本所在目录构建路径
+				script_dir = Path(__file__).resolve().parent.parent.parent
+				script_preset_dir = script_dir / "data" / "preset"
+				if script_preset_dir.exists():
+					return script_preset_dir
+				# 如果都不存在，返回默认路径
+				return Path("data/preset")
+			return preset_dir
+		except ImportError:
+			# 如果无法导入settings，尝试使用相对路径
+			cwd_preset_dir = Path.cwd() / "data" / "preset"
+			if cwd_preset_dir.exists():
+				return cwd_preset_dir
+			# 尝试从脚本所在目录构建路径
+			script_dir = Path(__file__).resolve().parent.parent.parent
+			script_preset_dir = script_dir / "data" / "preset"
+			if script_preset_dir.exists():
+				return script_preset_dir
+			# 如果都不存在，返回默认路径
+			return Path("data/preset")
+
+	@classmethod
+	async def list_all_presets(cls) -> Dict[str, List[Dict]]:
+		"""获取所有预设列表及其基本信息"""
+		preset_dir = cls.get_preset_dir()
+		if not preset_dir.exists():
+			return {"presets": []}
+
+		presets = []
+		for preset_file in preset_dir.glob("*.json"):
+			try:
+				with open(preset_file, 'r', encoding='utf-8') as f:
+					preset_data = json.load(f)
+					presets.append({
+						"name": preset_file.stem,
+						"description": preset_data.get("description", ""),
+						"component_count": len(preset_data.get("prompts", [])),
+						"temperature": preset_data.get("temperature", 1.0)
+					})
+			except Exception:
+				continue  # 跳过损坏的预设文件
+		return {"presets": presets}
+
+	@classmethod
+	async def get_preset(cls, preset_name: str) -> Dict[str, Any]:
+		"""获取指定预设的完整内容"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 处理prompt_order，简化为单角色配置
+			if 'prompt_order' in preset_data and isinstance(preset_data['prompt_order'], list) and len(
+					preset_data['prompt_order']) > 0:
+				# 检查第一个元素是否为字典（多角色配置）
+				first_item = preset_data['prompt_order'][0]
+				if isinstance(first_item, dict) and 'order' in first_item:
+					# 提取第一个角色的order配置
+					first_role_order = first_item
+					if isinstance(first_role_order['order'], list):
+						# 简化为只包含enabled为True的identifier列表
+						simplified_order = [
+							item.get('identifier')
+							for item in first_role_order['order']
+							if item.get('enabled', True)
+						]
+						preset_data['prompt_order'] = simplified_order
+
+			# 转换为AIDesignSpec对象进行验证
+			ai_design_spec = cls.from_dict(preset_data)
+
+			# 构建返回数据，确保格式与前端期望的一致
+			result = {
+				# 基础参数
+				"temperature": ai_design_spec.temperature,
+				"frequency_penalty": ai_design_spec.frequency_penalty,
+				"presence_penalty": ai_design_spec.presence_penalty,
+				"top_p": ai_design_spec.top_p,
+				"top_k": ai_design_spec.top_k,
+				"max_context": ai_design_spec.max_context,
+				"max_tokens": ai_design_spec.max_tokens,
+				"max_context_unlocked": ai_design_spec.max_context_unlocked,
+				"stream_openai": ai_design_spec.stream,
+				"seed": ai_design_spec.seed,
+				"n": ai_design_spec.n,
+
+				# 兼容旧格式
+				"openai_max_context": ai_design_spec.max_context,
+				"openai_max_tokens": ai_design_spec.max_tokens,
+
+				# 其他参数
+				"top_a": ai_design_spec.top_a,
+				"min_p": ai_design_spec.min_p,
+				"repetition_penalty": ai_design_spec.repetition_penalty,
+				"names_behavior": ai_design_spec.names_behavior,
+				"send_if_empty": ai_design_spec.send_if_empty,
+				"impersonation_prompt": ai_design_spec.impersonation_prompt,
+				"new_chat_prompt": ai_design_spec.new_chat_prompt,
+				"new_group_chat_prompt": ai_design_spec.new_group_chat_prompt,
+				"new_example_chat_prompt": ai_design_spec.new_example_chat_prompt,
+				"continue_nudge_prompt": ai_design_spec.continue_nudge_prompt,
+				"bias_preset_selected": ai_design_spec.bias_preset_selected,
+				"wi_format": ai_design_spec.wi_format,
+				"scenario_format": ai_design_spec.scenario_format,
+				"personality_format": ai_design_spec.personality_format,
+				"group_nudge_prompt": ai_design_spec.group_nudge_prompt,
+				"assistant_prefill": ai_design_spec.assistant_prefill,
+				"assistant_impersonation": ai_design_spec.assistant_impersonation,
+				"use_sysprompt": ai_design_spec.use_sysprompt,
+				"squash_system_messages": ai_design_spec.squash_system_messages,
+				"media_inlining": ai_design_spec.media_inlining,
+				"continue_prefill": ai_design_spec.continue_prefill,
+				"continue_postfix": ai_design_spec.continue_postfix,
+
+				# 处理组件
+				"prompts": []
+			}
+
+			# 处理组件列表
+			if ai_design_spec.prompts:
+				# 获取当前角色的prompt_order（简化后的字符串列表）
+				current_order = ai_design_spec.prompt_order if ai_design_spec.prompt_order else []
+
+				# 构建组件列表
+				for prompt in ai_design_spec.prompts:
+					# 检查组件是否在order中
+					is_in_order = prompt.identifier in current_order
+
+					# 构建组件对象
+					component = {
+						"identifier": prompt.identifier,
+						"name": prompt.name,
+						"content": prompt.content if hasattr(prompt, 'content') else "",
+						"role": prompt.role if hasattr(prompt, 'role') else (0 if prompt.system_prompt else 1),
+						"system_prompt": prompt.system_prompt,
+						"marker": prompt.marker,
+						"enabled": is_in_order if current_order else True
+					}
+
+					result["prompts"].append(component)
+
+				# 按照order排序组件
+				if current_order:
+					result["prompts"].sort(
+							key=lambda x: current_order.index(x["identifier"]) if x[
+								                                                      "identifier"] in current_order else len(
+									current_order))
+
+			# 添加prompt_order
+			result["prompt_order"] = ai_design_spec.prompt_order if ai_design_spec.prompt_order else []
+
+			return result
+		except Exception as e:
+			raise Exception(f"Failed to load preset: {str(e)}")
+
+	@classmethod
+	async def create_preset(cls, preset_name: str, preset_data: Dict) -> Dict[str, str]:
+		"""创建新预设"""
+		preset_dir = cls.get_preset_dir()
+		preset_dir.mkdir(parents=True, exist_ok=True)
+		preset_path = preset_dir / f"{preset_name}.json"
+
+		if preset_path.exists():
+			raise FileExistsError(f"Preset already exists: {preset_name}")
+
+		try:
+			# 验证并转换为AIDesignSpec对象
+			ai_design_spec = cls.from_dict(preset_data)
+
+			# 保存到文件
+			with open(preset_path, 'w', encoding='utf-8') as f:
+				json.dump(ai_design_spec.dict(), f, ensure_ascii=False, indent=2)
+
+			return {"message": "Preset created successfully", "name": preset_name}
+		except Exception as e:
+			raise Exception(f"Failed to create preset: {str(e)}")
+
+	@classmethod
+	async def update_preset(cls, preset_name: str, update_data: Dict) -> Dict[str, str]:
+		"""更新预设配置"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			# 加载现有预设
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 更新字段
+			for key, value in update_data.items():
+				preset_data[key] = value
+
+			# 验证并转换为AIDesignSpec对象
+			ai_design_spec = cls.from_dict(preset_data)
+
+			# 保存更新
+			with open(preset_path, 'w', encoding='utf-8') as f:
+				json.dump(ai_design_spec.dict(), f, ensure_ascii=False, indent=2)
+
+			return {"message": "Preset updated successfully"}
+		except Exception as e:
+			raise Exception(f"Failed to update preset: {str(e)}")
+
+	@classmethod
+	async def delete_preset(cls, preset_name: str) -> Dict[str, str]:
+		"""删除指定预设"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			preset_path.unlink()
+			return {"message": "Preset deleted successfully"}
+		except Exception as e:
+			raise Exception(f"Failed to delete preset: {str(e)}")
+
+	@classmethod
+	async def list_components(cls, preset_name: str) -> Dict[str, List[Dict]]:
+		"""获取预设中的所有组件"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 获取组件列表
+			components = preset_data.get("prompts", [])
+			return {"components": components}
+		except Exception as e:
+			raise Exception(f"Failed to load components: {str(e)}")
+
+	@classmethod
+	async def get_component(cls, preset_name: str, component_id: str) -> Dict[str, Any]:
+		"""获取指定组件的详情"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 查找组件
+			components = preset_data.get("prompts", [])
+			component = next((c for c in components if c.get("identifier") == component_id), None)
+
+			if not component:
+				raise FileNotFoundError(f"Component not found: {component_id}")
+
+			return component
+		except FileNotFoundError:
+			raise
+		except Exception as e:
+			raise Exception(f"Failed to load component: {str(e)}")
+
+	@classmethod
+	async def add_component_to_preset(cls, preset_name: str, component_data: Dict) -> Dict[str, str]:
+		"""向预设添加新组件"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			# 加载预设数据
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 验证组件数据
+			component = PromptComponent(**component_data)
+
+			# 检查组件ID是否已存在
+			components = preset_data.get("prompts", [])
+			if any(c.get("identifier") == component.identifier for c in components):
+				raise ValueError(f"Component identifier already exists: {component.identifier}")
+
+			# 添加组件
+			components.append(component.dict())
+			preset_data["prompts"] = components
+
+			# 保存更新
+			with open(preset_path, 'w', encoding='utf-8') as f:
+				json.dump(preset_data, f, ensure_ascii=False, indent=2)
+
+			return {"message": "Component added successfully", "identifier": component.identifier}
+		except (FileNotFoundError, ValueError):
+			raise
+		except Exception as e:
+			raise Exception(f"Failed to add component: {str(e)}")
+
+	@classmethod
+	async def update_component_in_preset(cls, preset_name: str, component_id: str, update_data: Dict) -> Dict[str, str]:
+		"""更新指定组件"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			# 加载预设数据
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 查找并更新组件
+			components = preset_data.get("prompts", [])
+			component_index = next((i for i, c in enumerate(components) if c.get("identifier") == component_id), None)
+
+			if component_index is None:
+				raise FileNotFoundError(f"Component not found: {component_id}")
+
+			# 更新组件字段
+			for key, value in update_data.items():
+				components[component_index][key] = value
+
+			# 保存更新
+			with open(preset_path, 'w', encoding='utf-8') as f:
+				json.dump(preset_data, f, ensure_ascii=False, indent=2)
+
+			return {"message": "Component updated successfully"}
+		except FileNotFoundError:
+			raise
+		except Exception as e:
+			raise Exception(f"Failed to update component: {str(e)}")
+
+	@classmethod
+	async def delete_component_from_preset(cls, preset_name: str, component_id: str) -> Dict[str, str]:
+		"""从预设中删除指定组件"""
+		preset_path = cls.get_preset_dir() / f"{preset_name}.json"
+		if not preset_path.exists():
+			raise FileNotFoundError(f"Preset not found: {preset_name}")
+
+		try:
+			# 加载预设数据
+			with open(preset_path, 'r', encoding='utf-8') as f:
+				preset_data = json.load(f)
+
+			# 查找并删除组件
+			components = preset_data.get("prompts", [])
+			original_length = len(components)
+			components = [c for c in components if c.get("identifier") != component_id]
+
+			if len(components) == original_length:
+				raise FileNotFoundError(f"Component not found: {component_id}")
+
+			# 更新预设数据
+			preset_data["prompts"] = components
+
+			# 保存更新
+			with open(preset_path, 'w', encoding='utf-8') as f:
+				json.dump(preset_data, f, ensure_ascii=False, indent=2)
+
+			return {"message": "Component deleted successfully"}
+		except FileNotFoundError:
+			raise
+		except Exception as e:
+			raise Exception(f"Failed to delete component: {str(e)}")
+
 	# ========== 组件管理方法 ==========
 
 	def add_component(self, component: PromptComponent) -> None:

backend/core/models/chat_history.py

@@ -88,7 +88,175 @@ class ChatHistory(BaseModel):
 	class Config:
 		arbitrary_types_allowed = True
 
-	@classmethod  # 类方法装饰器，表示这是一个类方法，可以通过类名直接调用
+	@classmethod
+	def get_data_path(cls) -> Path:
+		"""获取数据目录路径"""
+		try:
+			from backend.core.config import settings
+			return settings.DATA_PATH / "chat"
+		except ImportError:
+			return Path("data")
+
+	@classmethod
+	async def list_all_chats(cls) -> Dict[str, List[Dict]]:
+		"""获取所有角色的所有聊天列表"""
+		data_dir = cls.get_data_path()
+		if not data_dir.exists():
+			return {"chats": []}
+
+		chats = []
+		for role_dir in data_dir.iterdir():
+			if role_dir.is_dir():
+				for chat_file in role_dir.glob("*.jsonl"):
+					try:
+						with open(chat_file, 'r', encoding='utf-8') as f:
+							# 读取第一行获取元数据
+							first_line = f.readline()
+							metadata = json.loads(first_line)
+							chats.append({
+								"role_name": role_dir.name,
+								"chat_name": chat_file.stem,
+								"user_name": metadata.get("user_name", "User"),
+								"character_name": metadata.get("character_name", "Assistant"),
+								"last_modified": metadata.get("last_modified", ""),
+								"message_count": sum(1 for _ in f)  # 统计剩余行数(消息数)
+							})
+					except Exception:
+						continue  # 跳过损坏的聊天文件
+		return {"chats": chats}
+
+	@classmethod
+	async def get_chat(cls, role_name: str, chat_name: str) -> Dict[str, Any]:
+		"""获取指定聊天的完整内容"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+		return {
+			"metadata": chat_history.chat_metadata.dict(),
+			"messages": chat_history.to_chatbox_format()
+		}
+
+	@classmethod
+	async def create_chat(cls, role_name: str, chat_name: str, metadata: Optional[Dict] = None) -> Dict[str, str]:
+		"""创建新聊天"""
+		base_path = cls.get_data_path()
+		role_dir = base_path / role_name
+		role_dir.mkdir(parents=True, exist_ok=True)
+		chat_path = role_dir / f"{chat_name}.jsonl"
+
+		if chat_path.exists():
+			raise FileExistsError(f"Chat already exists: {chat_path}")
+
+		# 创建聊天历史对象
+		chat_history = cls(
+				chat_metadata=ChatMetadata(**(metadata or {})),
+				messages=[]
+		)
+
+		# 保存到文件
+		chat_history.save_to_file(role_name, chat_name, base_path)
+		return {"message": "Chat created successfully"}
+
+	@classmethod
+	async def update_chat(cls, role_name: str, chat_name: str, update_data: Dict) -> Dict[str, str]:
+		"""更新聊天元数据"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+
+		# 更新元数据
+		if "metadata" in update_data:
+			for key, value in update_data["metadata"].items():
+				if hasattr(chat_history.chat_metadata, key):
+					setattr(chat_history.chat_metadata, key, value)
+
+		# 保存更改
+		base_path = cls.get_data_path()
+		chat_history.save_to_file(role_name, chat_name, base_path)
+		return {"message": "Chat metadata updated successfully"}
+
+	@classmethod
+	async def delete_chat(cls, role_name: str, chat_name: str) -> Dict[str, str]:
+		"""删除指定聊天"""
+		base_path = cls.get_data_path()
+		chat_path = base_path / role_name / f"{chat_name}.jsonl"
+
+		if not chat_path.exists():
+			raise FileNotFoundError(f"Chat not found: {chat_path}")
+
+		chat_path.unlink()
+		return {"message": "Chat deleted successfully"}
+
+	@classmethod
+	async def list_messages(cls, role_name: str, chat_name: str) -> Dict[str, List[Dict]]:
+		"""获取聊天的所有消息"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+		return {"messages": chat_history.to_chatbox_format()}
+
+	@classmethod
+	async def get_message(cls, role_name: str, chat_name: str, floor: int) -> Dict[str, Any]:
+		"""获取指定楼层的消息"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+		message = next((msg for msg in chat_history.messages if msg.floor == floor), None)
+
+		if not message:
+			raise FileNotFoundError(f"Message not found: floor {floor}")
+
+		return message.dict()
+
+	@classmethod
+	async def add_message(cls, role_name: str, chat_name: str, message_data: Dict) -> Dict[str, Any]:
+		"""向聊天添加新消息"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+
+		# 创建消息对象
+		message = Message(**message_data)
+
+		# 检查楼层是否已存在
+		if any(msg.floor == message.floor for msg in chat_history.messages):
+			raise ValueError(f"Message floor already exists: {message.floor}")
+
+		# 添加消息
+		chat_history.messages.append(message)
+
+		# 保存更改
+		base_path = cls.get_data_path()
+		chat_history.save_to_file(role_name, chat_name, base_path)
+		return {"message": "Message added successfully", "floor": message.floor}
+
+	@classmethod
+	async def update_message(cls, role_name: str, chat_name: str, floor: int, update_data: Dict) -> Dict[str, str]:
+		"""更新指定楼层的消息"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+		message = next((msg for msg in chat_history.messages if msg.floor == floor), None)
+
+		if not message:
+			raise FileNotFoundError(f"Message not found: floor {floor}")
+
+		# 更新消息字段
+		for key, value in update_data.items():
+			if hasattr(message, key):
+				setattr(message, key, value)
+
+		# 保存更改
+		base_path = cls.get_data_path()
+		chat_history.save_to_file(role_name, chat_name, base_path)
+		return {"message": "Message updated successfully"}
+
+	@classmethod
+	async def delete_message(cls, role_name: str, chat_name: str, floor: int) -> Dict[str, str]:
+		"""删除指定楼层的消息"""
+		chat_history = cls.load_from_file(role_name, chat_name)
+
+		# 查找并删除消息
+		original_length = len(chat_history.messages)
+		chat_history.messages = [msg for msg in chat_history.messages if msg.floor != floor]
+
+		if len(chat_history.messages) == original_length:
+			raise FileNotFoundError(f"Message not found: floor {floor}")
+
+		# 保存更改
+		base_path = cls.get_data_path()
+		chat_history.save_to_file(role_name, chat_name, base_path)
+		return {"message": "Message deleted successfully"}
+
+	@classmethod
 	def load_from_file(cls, role_name: str, chat_name: str, base_path: Path = None) -> 'ChatHistory':
 		"""
 		从JSONL文件加载聊天历史
@@ -105,10 +273,9 @@ class ChatHistory(BaseModel):
 			FileNotFoundError: 当文件不存在时抛出
 			json.JSONDecodeError: 当JSON解析失败时抛出
 		"""
-		# 设置默认基础路径 - 如果未提供base_path，则从配置中获取默认路径
+		# 设置默认基础路径
 		if base_path is None:
-			from backend.core.config import settings  # 延迟导入配置模块
-			base_path = settings.DATA_PATH / "chat"  # 构建默认路径
+			base_path = cls.get_data_path()
 
 		# 构建文件路径
 		file_path = base_path / role_name / f"{chat_name}.jsonl"
@@ -258,7 +425,7 @@ class ChatHistory(BaseModel):
 		"""
 		# 设置默认基础路径
 		if base_path is None:
-			base_path = Path("data")
+			base_path = self.get_data_path()
 
 		# 构建文件路径
 		file_path = base_path / role_name / f"{chat_name}.jsonl"