bwstudio
ceed63fcb0
High priority: - Fix concurrent race condition for view_count/like_count (atomic update) - Add route request ID tracking to prevent race conditions - Filter get_joke by status=approved (no pending content leak) - Add error feedback for like button Performance: - Optimize random joke query (avoid full table sort) - Limit page_size max to 100 (DoS prevention) Medium: - Add localStorage quota error handling - Handle empty AI response gracefully - Fix generate content title extraction Low: - Add rejected_jokes to stats API - Update dashboard to show rejected count
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
"""智能笑话生成器 API"""
|
||
|
||
from datetime import datetime
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from openai import OpenAI
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.database import get_db
|
||
from app.models.setting import AiSetting
|
||
from app.schemas.joke import GenerateRequest, GenerateResponse
|
||
|
||
|
||
router = APIRouter(prefix="/generate", tags=["生成器"])
|
||
|
||
|
||
# AI Prompt
|
||
GENERATION_PROMPT = """你是一位幽默大师，专门创作轻松搞笑的短笑话。
|
||
|
||
{context}
|
||
|
||
要求：
|
||
1. 根据场景和关键词创作一条原创笑话
|
||
2. 笑话要有反转或意外结局
|
||
3. 语言简洁，30-150字
|
||
4. 直接输出笑话内容，不需要解释
|
||
|
||
格式：
|
||
标题：xxx
|
||
内容：xxx
|
||
"""
|
||
|
||
|
||
def _build_prompt(scenarios: list[str], keywords: list[str]) -> str:
|
||
"""构建 AI prompt"""
|
||
parts = []
|
||
if scenarios:
|
||
parts.append(f"场景：{', '.join(scenarios)}")
|
||
if keywords:
|
||
parts.append(f"关键词：{', '.join(keywords)}")
|
||
if not parts:
|
||
parts.append("场景：日常生活的各种趣事（不指定具体场景）")
|
||
return GENERATION_PROMPT.format(context="\n".join(parts))
|
||
|
||
|
||
@router.post("", response_model=GenerateResponse)
|
||
def generate_joke(
|
||
req: GenerateRequest,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""调用 AI 生成笑话"""
|
||
# 获取激活的 AI 配置
|
||
setting = db.query(AiSetting).filter(AiSetting.is_active == True).first()
|
||
if not setting:
|
||
# 如果没有配置，尝试返回默认配置
|
||
setting = db.query(AiSetting).first()
|
||
|
||
if not setting or not setting.api_key:
|
||
raise HTTPException(status_code=503, detail="AI 服务未配置，请联系管理员")
|
||
|
||
# 调用 AI
|
||
try:
|
||
client = OpenAI(base_url=setting.api_base, api_key=setting.api_key)
|
||
prompt = _build_prompt(req.scenarios, req.keywords)
|
||
|
||
response = client.chat.completions.create(
|
||
model=setting.model_name,
|
||
messages=[{"role": "user", "content": prompt}],
|
||
temperature=setting.temperature,
|
||
max_tokens=setting.max_tokens,
|
||
)
|
||
|
||
raw = response.choices[0].message.content
|
||
return _parse_response(raw)
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}")
|
||
|
||
|
||
def _parse_response(raw: str | None) -> GenerateResponse:
|
||
"""解析 AI 返回内容，提取标题和内容"""
|
||
# 防御：处理空或 None 输入
|
||
if not raw or not raw.strip():
|
||
raise ValueError("AI 返回内容为空")
|
||
|
||
title = ""
|
||
content = raw
|
||
|
||
# 尝试提取 "标题：xxx" 或 "标题:xxx"
|
||
for line in raw.split("\n"):
|
||
line = line.strip()
|
||
if line.startswith("标题：") or line.startswith("标题:"):
|
||
title = line.split("：", 1)[-1].split(":", 1)[-1].strip()
|
||
# 只替换这一行，不要 replace 全局
|
||
lines = content.split("\n")
|
||
for i, l in enumerate(lines):
|
||
if l.strip() == line:
|
||
lines[i] = ""
|
||
break
|
||
content = "\n".join(lines).strip()
|
||
break
|
||
|
||
# 如果没有提取到标题，取第一行
|
||
if not title:
|
||
first_line = raw.split("\n")[0].strip()
|
||
if first_line.startswith("标题"):
|
||
first_line = first_line.split("：", 1)[-1].split(":", 1)[-1].strip()
|
||
title = first_line[:30] if len(first_line) > 30 else first_line
|
||
|
||
# 防御：content 不能为空
|
||
if not content.strip():
|
||
content = "（内容生成失败，请重新生成）"
|
||
|
||
return GenerateResponse(
|
||
title=title or "生成的笑话",
|
||
content=content.strip(),
|
||
created_at=datetime.now(),
|
||
) |