Files
meijiaka-zy/python-api/app/api/v1/script.py
T
小鱼开发 aa818b75a8 refactor: 脚本提示词体系重构 — 子类从目录改为单个文件
- 删除 system/bk/ 下所有旧提示词,平铺替换为 23 个新文件
- 文件名格式统一为: 文案——描述.txt
- 后端: _meta.json 扁平化,loader.py 新增 list_prompt_files() + load_prompt_file()
- 后端: API 从 subcategory 改为 filename,按指定文件读取
- 后端: categories 接口返回文件列表(label/desc/filename)供前端展示
- 前端: ScriptCreation 分类选择改为卡片网格,展示文案+描述
- 前端: 清理 subcategoryCode,统一改为 filename
- 前端: 字幕字号调整为 64/96/80px
2026-06-01 13:44:59 +08:00

215 lines
7.5 KiB
Python

"""
脚本生成 API
============
提供脚本分类查询、润色、标题生成等功能。
"""
from __future__ import annotations
import asyncio
import logging
import time
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.ai.model_router import get_model_router
from app.ai.prompts import list_categories, list_prompt_files, load_prompt, render_template
from app.api.deps import get_current_user
from app.db.session import get_db
from app.models.user import User
from app.schemas.common import ApiResponse, success_response
from app.schemas.script import (
CategoryItem,
GenerateTitleRequest,
GenerateTitleResponse,
PolishRequest,
)
from app.services import point_service as ps
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()
for cat in categories:
cat["files"] = list_prompt_files(cat["code"])
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}_{int(time.time() * 1000)}",
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.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}_{int(time.time() * 1000)}",
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)}")