小鱼开发
dc5d0d4959
- voice.py: synthesize/synthesize-batch/synthesize-file/clone/submit 添加前置积分检查 - script.py: polish/generate_title 添加前置积分检查 - 修复 HTTPException(402) 被外层 except Exception 吞掉变成 500 的问题 - 所有端点在执行业务前先检查余额，不足直接返回 402
244 lines
8.4 KiB
Python
244 lines
8.4 KiB
Python
"""
|
|
脚本生成 API
|
|
============
|
|
|
|
提供脚本生成、润色、模型健康检查等功能。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import get_current_user
|
|
from app.db.session import get_db
|
|
from app.ai.model_router import get_model_router
|
|
from app.ai.prompts import list_categories, load_prompt, render_template
|
|
from app.schemas.common import ApiResponse, success_response
|
|
from app.schemas.script import (
|
|
CategoryItem,
|
|
GenerateTitleRequest,
|
|
GenerateTitleResponse,
|
|
ModelHealthResponse,
|
|
PolishRequest,
|
|
TestModelRequest,
|
|
TestModelResponse,
|
|
)
|
|
from app.services.script_service import get_script_service
|
|
from app.services import point_service as ps
|
|
from app.models.user import User
|
|
|
|
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("/polish", response_model=ApiResponse[str])
|
|
async def polish_content(
|
|
request: PolishRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
AI 润色文案/画面描述
|
|
|
|
- `polishType=scene`: 润色画面描述（根据 shot_type 自动区分分镜/空镜）
|
|
- `polishType=voiceover`: 润色配音文本
|
|
|
|
参数:
|
|
- `shot_type`: "segment"（分镜）或 "empty_shot"（空镜），画面润色时必填
|
|
"""
|
|
service = get_script_service()
|
|
type_name = "画面" if request.polish_type == "scene" else "文案"
|
|
|
|
# 前置积分检查
|
|
required_points = ps._calculate_cost("polish")
|
|
check = await ps.check_balance(db, current_user.id, required_points)
|
|
if not check["sufficient"]:
|
|
raise HTTPException(
|
|
status_code=402,
|
|
detail=f"积分不足，需要 {required_points} 积分，当前余额 {check['balance']}",
|
|
)
|
|
|
|
try:
|
|
polished = await service.polish_content(
|
|
content=request.content,
|
|
polish_type=request.polish_type,
|
|
shot_type=request.shot_type or "segment",
|
|
)
|
|
|
|
# 扣费
|
|
points = ps._calculate_cost("polish")
|
|
await ps.consume(
|
|
db,
|
|
user_id=current_user.id,
|
|
points=points,
|
|
source_type="polish",
|
|
source_id=f"polish_{current_user.id}_{asyncio.get_event_loop().time()}",
|
|
description="【文案润色】",
|
|
)
|
|
await db.commit()
|
|
|
|
return success_response(
|
|
data=polished,
|
|
message=f"{type_name}润色完成",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
if "积分不足" in str(e):
|
|
raise HTTPException(status_code=402, detail=str(e))
|
|
logger.warning(f"[Polish] 润色失败: {e}")
|
|
raise HTTPException(status_code=500, detail="润色失败，请检查输入内容后重试")
|
|
except Exception as e:
|
|
logger.error(f"[Polish] 润色异常: {e}")
|
|
raise HTTPException(status_code=500, detail=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,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
根据脚本内容智能生成标题
|
|
|
|
调用 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:
|
|
raise HTTPException(status_code=500, detail="标题生成提示词文件缺失")
|
|
|
|
# 根据使用场景确定描述
|
|
if request.usage == "cover":
|
|
usage_desc = "视频封面标题——用于封面图设计，是决定用户是否点击的第一要素"
|
|
style_requirement = "极具冲击力、抓眼球，适合静态封面大图展示，善用爆款句式"
|
|
usage_note = "- 封面主标题必须极度吸睛，让用户一眼就想点进去，善用数字、疑问、痛点、冲突\n- 封面副标题要补充悬念或细节，激发点击欲望"
|
|
else:
|
|
usage_desc = "视频画面标题——直接叠加在视频画面上，与动态视频内容共存"
|
|
style_requirement = "口语化、精炼有力，适合视频内展示，避免遮挡画面主体"
|
|
usage_note = "- 视频画面上的标题需要精炼，聚焦核心关键词\n- 副标题与主标题形成呼应，补充说明但不喧宾夺主"
|
|
|
|
# 渲染用户提示词
|
|
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,
|
|
usage=request.usage,
|
|
usage_desc=usage_desc,
|
|
style_requirement=style_requirement,
|
|
usage_note=usage_note,
|
|
)
|
|
|
|
# 前置积分检查
|
|
required_points = ps._calculate_cost("title")
|
|
check = await ps.check_balance(db, current_user.id, required_points)
|
|
if not check["sufficient"]:
|
|
raise HTTPException(
|
|
status_code=402,
|
|
detail=f"积分不足，需要 {required_points} 积分，当前余额 {check['balance']}",
|
|
)
|
|
|
|
try:
|
|
async with asyncio.timeout(15):
|
|
result = await model_router.generate(
|
|
prompt=user_prompt,
|
|
system_prompt=system_prompt,
|
|
task_type="script",
|
|
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]
|
|
|
|
# 扣费
|
|
points = ps._calculate_cost("title")
|
|
await ps.consume(
|
|
db,
|
|
user_id=current_user.id,
|
|
points=points,
|
|
source_type="title",
|
|
source_id=f"title_{current_user.id}_{asyncio.get_event_loop().time()}",
|
|
description="【标题生成】",
|
|
)
|
|
await db.commit()
|
|
|
|
return success_response(
|
|
data=GenerateTitleResponse(title=title),
|
|
message="标题生成成功",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except TimeoutError:
|
|
logger.warning("[generate_title] 标题生成超时")
|
|
raise HTTPException(status_code=500, detail="标题生成超时，请重试")
|
|
except ValueError as e:
|
|
if "积分不足" in str(e):
|
|
raise HTTPException(status_code=402, detail=str(e))
|
|
raise HTTPException(status_code=500, detail=f"标题生成失败: {str(e)}")
|
|
except Exception as e:
|
|
logger.error(f"[generate_title] 标题生成失败: {e}")
|
|
raise HTTPException(status_code=500, detail=f"标题生成失败: {str(e)}")
|