小鱼开发
475758beed
后端： - 新增 POST /script/generate-title API - 新增提示词模板 title_system.txt / title.txt（文件管理） - 根据脚本内容调用 LLM 生成大标题（≤8字）/小标题（≤30字） 前端： - 大标题/小标题输入框右侧新增【智能生成】按钮 - 点击后根据 utterances 拼接脚本内容调用 API - 添加 title-input-row / title-generate-btn CSS 样式
268 lines
8.0 KiB
Python
268 lines
8.0 KiB
Python
"""
|
||
脚本生成 API
|
||
============
|
||
|
||
提供脚本生成、润色、模型健康检查等功能。
|
||
支持 SSE 流式响应。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
|
||
from fastapi import APIRouter, Request
|
||
from fastapi.responses import StreamingResponse
|
||
|
||
from app.schemas.common import ApiResponse, success_response
|
||
from app.ai.model_router import get_model_router
|
||
from app.ai.prompts import list_categories, load_prompt, render_template
|
||
from app.schemas.script import (
|
||
CategoryItem,
|
||
GenerateScriptRequest,
|
||
GenerateTitleRequest,
|
||
GenerateTitleResponse,
|
||
ModelHealthResponse,
|
||
PolishRequest,
|
||
ScriptGenerationEvent,
|
||
ScriptShot,
|
||
TestModelRequest,
|
||
TestModelResponse,
|
||
)
|
||
from app.services.script_service import get_script_service
|
||
|
||
router = APIRouter()
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@router.get("/categories", response_model=ApiResponse[list[CategoryItem]])
|
||
async def get_categories():
|
||
"""
|
||
获取提示词分类列表
|
||
|
||
返回所有大类和小类结构，供前端选择。
|
||
"""
|
||
categories = list_categories()
|
||
return success_response(
|
||
data=categories,
|
||
message="获取分类列表成功",
|
||
)
|
||
|
||
|
||
@router.post("/generate", response_model=ApiResponse[list[ScriptShot]])
|
||
async def generate_script(request: GenerateScriptRequest):
|
||
"""
|
||
同步生成脚本
|
||
|
||
直接返回生成的分镜列表，适合快速预览。
|
||
"""
|
||
service = get_script_service()
|
||
|
||
shots = await service.generate_script(
|
||
category=request.category,
|
||
subcategory=request.subcategory,
|
||
duration=request.duration,
|
||
script_type=request.script_type,
|
||
model=request.model,
|
||
)
|
||
|
||
return success_response(
|
||
data=shots,
|
||
message=f"成功生成 {len(shots)} 个分镜",
|
||
)
|
||
|
||
|
||
@router.post("/generate/stream")
|
||
async def generate_script_stream(request: Request, data: GenerateScriptRequest):
|
||
"""
|
||
流式生成脚本（SSE）
|
||
|
||
返回 Server-Sent Events，包含进度更新和最终结果。
|
||
前端通过 EventSource 接收实时进度。
|
||
|
||
**SSE 事件类型：**
|
||
- `start`: 开始生成
|
||
- `generating`: AI 生成中
|
||
- `complete`: 完成，包含 result 字段
|
||
- `error`: 错误
|
||
|
||
**示例事件流：**
|
||
```
|
||
data: {"type": "start", "message": "正在创作脚本..."}
|
||
|
||
data: {"type": "generating", "message": "正在创作脚本..."}
|
||
|
||
data: {"type": "complete", "message": "脚本生成成功", "result": [...]}
|
||
```
|
||
"""
|
||
service = get_script_service()
|
||
|
||
async def event_generator():
|
||
"""SSE 事件生成器，带客户端断开检测"""
|
||
try:
|
||
async for event in service.generate_script_stream(
|
||
category=data.category,
|
||
subcategory=data.subcategory,
|
||
duration=data.duration,
|
||
script_type=data.script_type,
|
||
model=data.model,
|
||
):
|
||
# 检查客户端是否已断开
|
||
if await request.is_disconnected():
|
||
logger.info("[SSE] 客户端已断开连接，停止生成")
|
||
break
|
||
|
||
# SSE 格式：data: {...}\n\n
|
||
try:
|
||
yield f"data: {event.model_dump_json()}\n\n"
|
||
await asyncio.sleep(0.05) # 确保事件被 flush，前端有时间渲染
|
||
except Exception as e:
|
||
logger.error(f"[SSE] 序列化事件失败: {e}")
|
||
continue
|
||
|
||
# 发送结束标记（如果客户端还连接着）
|
||
if not await request.is_disconnected():
|
||
yield "data: [DONE]\n\n"
|
||
|
||
except Exception as e:
|
||
logger.exception("[SSE] 事件生成器异常")
|
||
# 尝试发送错误信息给客户端
|
||
try:
|
||
error_event = ScriptGenerationEvent(
|
||
type="error",
|
||
progress=0,
|
||
message=f"服务器错误: {str(e)}",
|
||
)
|
||
yield f"data: {error_event.model_dump_json()}\n\n"
|
||
yield "data: [DONE]\n\n"
|
||
except:
|
||
pass
|
||
|
||
return StreamingResponse(
|
||
event_generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
"X-Accel-Buffering": "no", # 禁用 Nginx 缓冲
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/polish", response_model=ApiResponse[str])
|
||
async def polish_content(request: PolishRequest):
|
||
"""
|
||
AI 润色文案/画面描述
|
||
|
||
- `polishType=scene`: 润色画面描述（根据 shot_type 自动区分分镜/空镜）
|
||
- `polishType=voiceover`: 润色配音文案
|
||
|
||
参数:
|
||
- `shot_type`: "segment"（分镜）或 "empty_shot"（空镜），画面润色时必填
|
||
"""
|
||
service = get_script_service()
|
||
|
||
polished = await service.polish_content(
|
||
content=request.content,
|
||
polish_type=request.polish_type,
|
||
shot_type=request.shot_type or "segment",
|
||
)
|
||
|
||
type_name = "画面" if request.polish_type == "scene" else "文案"
|
||
return success_response(
|
||
data=polished,
|
||
message=f"{type_name}润色完成",
|
||
)
|
||
|
||
|
||
@router.get("/model-health", response_model=ApiResponse[ModelHealthResponse])
|
||
async def check_model_health():
|
||
"""
|
||
检查 AI 模型健康状态
|
||
|
||
返回所有配置的模型及其可用性状态。
|
||
"""
|
||
service = get_script_service()
|
||
health_data = await service.check_model_health()
|
||
|
||
return success_response(
|
||
data=ModelHealthResponse(**health_data),
|
||
message="模型健康检查完成",
|
||
)
|
||
|
||
|
||
@router.post("/test-model", response_model=ApiResponse[TestModelResponse])
|
||
async def test_model(request: TestModelRequest):
|
||
"""
|
||
测试指定模型连接
|
||
|
||
发送一个简单的测试请求，验证模型是否可用。
|
||
"""
|
||
service = get_script_service()
|
||
result = await service.test_model(request.model_id)
|
||
|
||
return success_response(
|
||
data=TestModelResponse(**result),
|
||
message="模型测试完成" if result["success"] else f"模型测试失败: {result.get('error')}",
|
||
)
|
||
|
||
|
||
@router.post("/generate-title", response_model=ApiResponse[GenerateTitleResponse])
|
||
async def generate_title(request: GenerateTitleRequest):
|
||
"""
|
||
根据脚本内容智能生成标题
|
||
|
||
调用 LLM 根据脚本内容生成大标题或小标题。
|
||
提示词从文件加载，支持热更新。
|
||
"""
|
||
model_router = await get_model_router()
|
||
|
||
# 加载提示词
|
||
system_prompt = load_prompt("user/title_system")
|
||
user_template = load_prompt("user/title")
|
||
|
||
if not system_prompt or not user_template:
|
||
return success_response(
|
||
code=500,
|
||
message="标题生成提示词文件缺失",
|
||
data=None,
|
||
)
|
||
|
||
# 渲染用户提示词
|
||
title_type_desc = "大标题（主标题，提炼核心卖点，吸睛）" if request.title_type == "main" else "小标题（副标题，补充说明或制造悬念）"
|
||
user_prompt = render_template(
|
||
user_template,
|
||
title_type=request.title_type,
|
||
title_type_desc=title_type_desc,
|
||
script_content=request.script_content,
|
||
max_length=request.max_length,
|
||
)
|
||
|
||
try:
|
||
result = await model_router.generate(
|
||
prompt=user_prompt,
|
||
system_prompt=system_prompt,
|
||
task_type="script",
|
||
temperature=0.8,
|
||
max_tokens=64,
|
||
)
|
||
|
||
title = result.content.strip() if result.content else ""
|
||
# 去除可能的引号
|
||
title = title.strip('"').strip("'").strip('「」').strip('『』').strip('《》')
|
||
# 截断到最大长度
|
||
if len(title) > request.max_length:
|
||
title = title[:request.max_length]
|
||
|
||
return success_response(
|
||
data=GenerateTitleResponse(title=title),
|
||
message="标题生成成功",
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"[generate_title] 标题生成失败: {e}")
|
||
return success_response(
|
||
code=500,
|
||
message=f"标题生成失败: {str(e)}",
|
||
data=None,
|
||
)
|