meijiaka-zy/python-api/app/api/v1/script.py

"""
脚本生成 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.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,
    PolishRequest,
)
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}_{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)}")