49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
from typing import List, Optional
|
|
from ...domain.repositories.prompt_repository import PromptRepository
|
|
from ...domain.entities.prompt import Prompt
|
|
from ..schemas.prompt import PromptCreate, PromptUpdate
|
|
|
|
|
|
class PromptUseCases:
|
|
"""提示词应用用例"""
|
|
|
|
def __init__(self, prompt_repository: PromptRepository):
|
|
self.prompt_repository = prompt_repository
|
|
|
|
async def get_all_prompts(self) -> List[Prompt]:
|
|
"""获取所有提示词"""
|
|
return await self.prompt_repository.get_all()
|
|
|
|
async def create_or_update_prompt(self, prompt_data: PromptCreate) -> Prompt:
|
|
"""创建或更新提示词"""
|
|
# 检查是否已存在
|
|
existing_prompt = await self.prompt_repository.get_by_step_and_name(
|
|
prompt_data.step, prompt_data.name
|
|
)
|
|
|
|
if existing_prompt:
|
|
# 更新现有提示词
|
|
existing_prompt.prompt = prompt_data.prompt
|
|
return await self.prompt_repository.save(existing_prompt)
|
|
else:
|
|
# 创建新提示词
|
|
new_prompt = Prompt(
|
|
id=None,
|
|
step=prompt_data.step,
|
|
name=prompt_data.name,
|
|
prompt=prompt_data.prompt
|
|
)
|
|
return await self.prompt_repository.save(new_prompt)
|
|
|
|
async def get_prompt_by_step_and_name(self, step: str, name: str) -> Optional[Prompt]:
|
|
"""根据step和name获取提示词"""
|
|
return await self.prompt_repository.get_by_step_and_name(step, name)
|
|
|
|
async def get_prompts_by_step(self, step: str) -> List[Prompt]:
|
|
"""根据step获取所有提示词"""
|
|
return await self.prompt_repository.get_by_step(step)
|
|
|
|
async def delete_prompt(self, prompt_id: int) -> bool:
|
|
"""删除提示词"""
|
|
return await self.prompt_repository.delete(prompt_id)
|