1691 lines
58 KiB
Python
1691 lines
58 KiB
Python
from datetime import date, datetime
|
||
from typing import Any, Dict, List, Optional
|
||
import time
|
||
import os
|
||
|
||
import requests
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||
from pydantic import BaseModel, Field, ConfigDict, field_validator
|
||
from sqlalchemy.orm import Session
|
||
import oss2
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
from ..db import get_db
|
||
from ..deps import get_current_user, AuthedUser
|
||
from ..response import ApiResponse, success_response
|
||
from ..models import (
|
||
GirlfriendEyeColor,
|
||
GirlfriendHairStyle,
|
||
GirlfriendHobbies,
|
||
GirlfriendMould,
|
||
Lover,
|
||
GenerationTask,
|
||
VoiceLibrary,
|
||
ChatSession,
|
||
ChatMessage,
|
||
)
|
||
from ..config import settings
|
||
from urllib.parse import urlparse
|
||
|
||
router = APIRouter(prefix="/lover", tags=["lover"])
|
||
|
||
|
||
@router.get("/test")
|
||
def test_endpoint():
|
||
"""测试接口，不需要认证"""
|
||
return {"status": "ok", "message": "Lover API is working", "env": settings.APP_ENV, "debug": settings.DEBUG}
|
||
|
||
|
||
def _user_gender_to_str(user_gender: Optional[int]) -> str:
|
||
# 假设 1=男,2=女,其它默认女
|
||
if user_gender == 1:
|
||
return "male"
|
||
if user_gender == 2:
|
||
return "female"
|
||
return "female"
|
||
|
||
|
||
def _opposite_gender(user_gender: Optional[int]) -> str:
|
||
return "female" if _user_gender_to_str(user_gender) == "male" else "male"
|
||
|
||
|
||
def _check_text_safe(text: Optional[str], field_name: str):
|
||
"""调用违禁词检测，检测到风险则抛出 400；无法调用时不阻断流程。"""
|
||
if not text:
|
||
return
|
||
try:
|
||
resp = requests.post(
|
||
"https://uapis.cn/api/v1/text/profanitycheck",
|
||
json={"text": text},
|
||
headers={"User-Agent": "lover-app/1.0"},
|
||
timeout=5,
|
||
)
|
||
except Exception:
|
||
return
|
||
if resp.status_code != 200:
|
||
return
|
||
try:
|
||
data = resp.json()
|
||
except Exception:
|
||
return
|
||
|
||
# 期望字段：status = ok / forbidden
|
||
status_str = str(data.get("status") or "").lower()
|
||
forbidden_words = [
|
||
str(w).strip()
|
||
for w in (data.get("forbidden_words") or [])
|
||
if str(w).strip()
|
||
]
|
||
# 接口命中违禁词时直接返回命中的词，便于前端提示
|
||
if status_str == "forbidden" or forbidden_words:
|
||
words_str = ", ".join(forbidden_words) if forbidden_words else ""
|
||
detail = (
|
||
f"{field_name}包含违禁词: {words_str}"
|
||
if words_str
|
||
else f"{field_name}包含违禁词，请调整后再试"
|
||
)
|
||
raise HTTPException(status_code=400, detail=detail)
|
||
|
||
|
||
def _build_lover_config_response(db: Session, user: AuthedUser) -> "LoverConfigResponse":
|
||
"""
|
||
复用恋人配置构建逻辑，便于正常接口和调试接口共用。
|
||
"""
|
||
lover = _get_or_create_lover(db, user)
|
||
|
||
gender = lover.gender or _opposite_gender(user.gender)
|
||
voices = (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.gender == gender)
|
||
.order_by(VoiceLibrary.id.asc())
|
||
.all()
|
||
)
|
||
default_voice = _get_default_voice(db, gender)
|
||
|
||
personality_options = (
|
||
db.query(GirlfriendMould)
|
||
.filter(
|
||
(GirlfriendMould.gender == gender) | (GirlfriendMould.gender.is_(None))
|
||
)
|
||
.order_by(GirlfriendMould.weigh.desc(), GirlfriendMould.id.asc())
|
||
.all()
|
||
)
|
||
hobby_options = (
|
||
db.query(GirlfriendHobbies)
|
||
.order_by(GirlfriendHobbies.weigh.desc(), GirlfriendHobbies.id.asc())
|
||
.all()
|
||
)
|
||
|
||
return LoverConfigResponse(
|
||
reg_step=user.reg_step or 1,
|
||
gender=gender,
|
||
lover=lover,
|
||
voices=voices,
|
||
default_voice_id=default_voice.id if default_voice else None,
|
||
personality_options=[
|
||
OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in personality_options
|
||
],
|
||
hobby_options=[
|
||
OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in hobby_options
|
||
],
|
||
)
|
||
|
||
|
||
class VoiceOut(BaseModel):
|
||
id: int
|
||
name: str
|
||
gender: str
|
||
style_tag: Optional[str] = None
|
||
avatar_url: Optional[str] = None
|
||
sample_audio_url: Optional[str] = None
|
||
tts_model_id: Optional[str] = None
|
||
is_default: bool = False
|
||
voice_code: str
|
||
is_owned: bool
|
||
price_gold: int
|
||
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
|
||
class LoverOut(BaseModel):
|
||
id: int
|
||
user_id: int
|
||
name: Optional[str] = None
|
||
gender: str
|
||
intro: Optional[str] = None
|
||
story_background: Optional[str] = None
|
||
personality_tag: Optional[int] = None
|
||
interest_tags: Optional[List[int]] = None
|
||
opening_line: Optional[str] = None
|
||
hair_style_id: Optional[int] = None
|
||
eye_color_id: Optional[int] = None
|
||
outfit_desc: Optional[str] = None
|
||
voice_id: Optional[int] = None
|
||
image_url: Optional[str] = None
|
||
image_gen_used: int
|
||
image_gen_limit: int
|
||
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
|
||
class LoverConfigResponse(BaseModel):
|
||
reg_step: int
|
||
gender: str
|
||
lover: Optional[LoverOut] = None
|
||
voices: List[VoiceOut] = Field(default_factory=list)
|
||
default_voice_id: Optional[int] = None
|
||
personality_options: List["OptionOut"] = Field(default_factory=list)
|
||
hobby_options: List["OptionOut"] = Field(default_factory=list)
|
||
|
||
|
||
class OptionOut(BaseModel):
|
||
id: int
|
||
name: str
|
||
weigh: Optional[int] = None
|
||
|
||
|
||
class LoverConfigIn(BaseModel):
|
||
gender: str = Field(default="female")
|
||
name: str
|
||
intro: str
|
||
story_background: str
|
||
personality_tag: int
|
||
interest_tags: List[int]
|
||
opening_line: str
|
||
|
||
|
||
class LoverVoiceUpdate(BaseModel):
|
||
gender: str = Field(default="female", pattern="^(male|female)$")
|
||
voice_id: int
|
||
|
||
|
||
class LoverVoiceSwitch(BaseModel):
|
||
voice_id: int
|
||
|
||
|
||
class AppearanceParams(BaseModel):
|
||
hair_style_id: Optional[int] = None
|
||
eye_color_id: Optional[int] = None
|
||
outfit_desc: Optional[str] = Field(default=None)
|
||
extra: Optional[Dict[str, Any]] = None # 预留扩展字段
|
||
|
||
@field_validator("hair_style_id", "eye_color_id")
|
||
@classmethod
|
||
def _positive(cls, v):
|
||
if v is not None and v <= 0:
|
||
raise ValueError("必须为正整数")
|
||
return v
|
||
|
||
|
||
class LoverAppearanceResponse(BaseModel):
|
||
lover_id: int
|
||
user_id: int
|
||
appearance_prompt: Optional[str] = None
|
||
appearance_params: Optional[AppearanceParams] = None
|
||
hair_style_id: Optional[int] = None
|
||
eye_color_id: Optional[int] = None
|
||
outfit_desc: Optional[str] = None
|
||
image_url: Optional[str] = None
|
||
last_image_task_id: Optional[int] = None
|
||
hair_style_options: List[OptionOut] = Field(default_factory=list)
|
||
eye_color_options: List[OptionOut] = Field(default_factory=list)
|
||
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
|
||
class LoverAppearanceIn(BaseModel):
|
||
appearance_prompt: Optional[str] = Field(default=None)
|
||
appearance_params: Optional[AppearanceParams] = None
|
||
|
||
|
||
class GenerateImageResponse(BaseModel):
|
||
task_id: int
|
||
image_url: Optional[str] = None
|
||
|
||
|
||
class LoverInitResponse(BaseModel):
|
||
session_id: int
|
||
opening_line: str
|
||
reg_step: int
|
||
personality_prompt: Optional[str] = None
|
||
|
||
|
||
class GenderOptionsResponse(BaseModel):
|
||
gender: str
|
||
voices: List[VoiceOut]
|
||
default_voice_id: Optional[int] = None
|
||
selected_voice_id: Optional[int] = None
|
||
personality_options: List[OptionOut]
|
||
|
||
|
||
class LoverBasicResponse(BaseModel):
|
||
lover_id: int
|
||
name: Optional[str] = None
|
||
image_url: Optional[str] = None
|
||
gender: Optional[str] = None
|
||
|
||
|
||
LoverConfigResponse.model_rebuild()
|
||
LoverAppearanceResponse.model_rebuild()
|
||
|
||
|
||
def _get_default_voice(db: Session, gender: str) -> Optional[VoiceLibrary]:
|
||
return (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.gender == gender, VoiceLibrary.is_default.is_(True))
|
||
.first()
|
||
) or (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.gender == gender)
|
||
.order_by(VoiceLibrary.id.asc())
|
||
.first()
|
||
)
|
||
|
||
|
||
def _get_or_create_lover(db: Session, user: AuthedUser) -> Lover:
|
||
"""
|
||
获取恋人记录，若不存在则按相反性别 + 默认音色自动创建。
|
||
并发场景下加行锁与唯一性兜底，避免重复插入。
|
||
"""
|
||
lover = (
|
||
db.query(Lover)
|
||
.filter(Lover.user_id == user.id)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if lover:
|
||
return lover
|
||
|
||
init_gender = _opposite_gender(user.gender)
|
||
default_voice = _get_default_voice(db, init_gender)
|
||
lover = Lover(
|
||
user_id=user.id,
|
||
gender=init_gender,
|
||
voice_id=default_voice.id if default_voice else None,
|
||
image_gen_used=0,
|
||
image_gen_limit=10,
|
||
image_gen_reset_date=date.today(),
|
||
)
|
||
db.add(lover)
|
||
try:
|
||
db.flush()
|
||
except IntegrityError:
|
||
# 并发插入唯一约束冲突时，回滚并读取已创建记录
|
||
db.rollback()
|
||
lover = (
|
||
db.query(Lover)
|
||
.filter(Lover.user_id == user.id)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if lover:
|
||
return lover
|
||
raise
|
||
return lover
|
||
|
||
|
||
def _reset_image_quota_if_needed(lover: Lover):
|
||
today = date.today()
|
||
if lover.image_gen_reset_date != today:
|
||
lover.image_gen_reset_date = today
|
||
lover.image_gen_used = 0
|
||
|
||
|
||
def _build_prompt(
|
||
lover: Lover,
|
||
hair_name: str,
|
||
eye_name: str,
|
||
outfit_desc: str,
|
||
) -> Dict[str, str]:
|
||
"""
|
||
聚合正向/反向提示词，强调写实、正面全身（含双腿双脚），避免半身/特写，杜绝光脚/发光眼/儿童脸，限定东亚/中国面孔。
|
||
"""
|
||
gender_str = "女性" if lover.gender == "female" else "男性"
|
||
positive_parts = [
|
||
lover.appearance_prompt or "",
|
||
"东亚人脸，中国面孔，肤色自然，五官协调，真人照片风格，真实摄影",
|
||
f"{gender_str}，正面全身站立，站在地面，双腿双脚完整清晰可见，无遮挡身体，光线自然",
|
||
f"发型：{hair_name}；瞳色：{eye_name}，仅瞳孔呈现此颜色，巩膜保持洁白自然，不发光；着装：{outfit_desc}，穿着鞋子和服饰，且服饰衣物完整覆盖身体，避免裸露",
|
||
"皮肤质感真实，高清细节，头身比例约 1:7.5，比例自然，姿态自然，眼睛自然不过饱和，无戏剧化光效",
|
||
]
|
||
negative_parts = [
|
||
"半身照",
|
||
"特写",
|
||
"只露上半身",
|
||
"缺失腿",
|
||
"缺失脚",
|
||
"只到膝盖",
|
||
"缺失躯干",
|
||
"只有头和脚",
|
||
"只有头",
|
||
"漂浮的头",
|
||
"身体被截断",
|
||
"裁剪身体",
|
||
"浮空脚",
|
||
"光脚",
|
||
"赤脚",
|
||
"barefoot",
|
||
"发光眼睛",
|
||
"glowing eyes",
|
||
"luminous eyes",
|
||
"shining eyes",
|
||
"eyes with light effects",
|
||
"荧光眼睛",
|
||
"激光眼",
|
||
"眼睛发光",
|
||
"眼睛冒光",
|
||
"眼睛射光",
|
||
"虹膜发光",
|
||
"瞳孔发光",
|
||
"光眼",
|
||
"眼睛闪光",
|
||
"卡通",
|
||
"动漫",
|
||
"二次元",
|
||
"插画",
|
||
"Q版",
|
||
"chibi",
|
||
"big head",
|
||
"large head",
|
||
"disproportionate head",
|
||
"夸张大头",
|
||
"大头娃娃",
|
||
"玩偶",
|
||
"娃娃脸",
|
||
"塑料感",
|
||
"塑料质感",
|
||
"蜡像",
|
||
"3d render",
|
||
"cg render",
|
||
"渲染感",
|
||
"小孩",
|
||
"儿童",
|
||
"未成年",
|
||
"幼年",
|
||
"儿童脸",
|
||
"童颜",
|
||
"婴儿脸",
|
||
"baby",
|
||
"child",
|
||
"teen",
|
||
"外国人",
|
||
"欧美人",
|
||
"裁剪",
|
||
"截断",
|
||
"背影",
|
||
"模糊",
|
||
"低分辨率",
|
||
"多余肢体",
|
||
"畸形",
|
||
"裸体",
|
||
"裸露",
|
||
"不穿衣服",
|
||
"不穿上装",
|
||
"不穿下装",
|
||
"nude",
|
||
"nsfw",
|
||
"透明衣服",
|
||
"整只眼球变色",
|
||
]
|
||
return {
|
||
"prompt": "，".join([p for p in positive_parts if p]),
|
||
"negative_prompt": "，".join(negative_parts),
|
||
}
|
||
|
||
|
||
def _build_prompt_v26(
|
||
lover: Lover,
|
||
hair_name: str,
|
||
eye_name: str,
|
||
outfit_desc: str,
|
||
) -> Dict[str, str]:
|
||
"""
|
||
为 wan2.6 定制的提示词，强化直视镜头/有神眼神，避免侧脸和目光游离。
|
||
"""
|
||
gender_str = "女性" if lover.gender == "female" else "男性"
|
||
def _enhance_prompt(text: Optional[str]) -> str:
|
||
"""简单描述词自动补充细节，避免输出抽象图。"""
|
||
base = (text or "").strip()
|
||
if not base or len(base) <= 12 or len(base.split()) <= 3:
|
||
return f"专业{gender_str}全身模特，{base or ''}，细节丰富，真实感强，写实风格"
|
||
return base
|
||
|
||
appearance_text = _enhance_prompt(lover.appearance_prompt)
|
||
positive_parts = [
|
||
"(masterpiece, best quality, 8k, professional photography:1.3)",
|
||
"(detailed eyes, looking at viewer, natural eye contact:1.4)",
|
||
appearance_text,
|
||
"东亚人脸，中国青年面孔，肤色自然白皙，五官协调，真人照片风格，真实摄影",
|
||
"专业人像摄影，眼神光自然，面部光线均匀，黄金分割构图",
|
||
"正面平视视角，相机高度与眼睛齐平，中距拍摄，无俯拍或仰拍，无鱼眼畸变",
|
||
f"{gender_str}，正面全身站立，站在地面，双腿双脚完整清晰可见，无遮挡身体，光线自然",
|
||
"直视镜头，眼神专注有神，目光自然交流，面部正面向前，表情自然生动",
|
||
"自然站姿，身体重心稳定，脊柱自然弯曲，肩颈放松，手脚姿态自然",
|
||
"头身比例严格约 1:7.5，面部大小与身体协调，避免头部过大或前伸",
|
||
f"发型：{hair_name}；瞳色：{eye_name}，仅瞳孔呈现此颜色，巩膜保持洁白自然，不发光；着装：{outfit_desc}，穿着鞋子和服饰，且服饰衣物完整覆盖身体，避免裸露",
|
||
"皮肤质感真实，高清细节，比例自然，姿态自然，眼睛自然不过饱和，有神且自然的眼神光",
|
||
]
|
||
negative_parts = [
|
||
"半身照",
|
||
"特写",
|
||
"只露上半身",
|
||
"缺失腿",
|
||
"缺失脚",
|
||
"只到膝盖",
|
||
"缺失躯干",
|
||
"只有头和脚",
|
||
"只有头",
|
||
"漂浮的头",
|
||
"身体被截断",
|
||
"裁剪身体",
|
||
"浮空脚",
|
||
"光脚",
|
||
"赤脚",
|
||
"barefoot",
|
||
"发光眼睛",
|
||
"glowing eyes",
|
||
"luminous eyes",
|
||
"shining eyes",
|
||
"eyes with light effects",
|
||
"荧光眼睛",
|
||
"激光眼",
|
||
"眼睛发光",
|
||
"眼睛冒光",
|
||
"眼睛射光",
|
||
"虹膜发光",
|
||
"瞳孔发光",
|
||
"光眼",
|
||
"眼睛闪光",
|
||
"卡通",
|
||
"动漫",
|
||
"二次元",
|
||
"插画",
|
||
"Q版",
|
||
"chibi",
|
||
"big head",
|
||
"large head",
|
||
"disproportionate head",
|
||
"夸张大头",
|
||
"大头娃娃",
|
||
"玩偶",
|
||
"娃娃脸",
|
||
"塑料感",
|
||
"塑料质感",
|
||
"蜡像",
|
||
"3d render",
|
||
"cg render",
|
||
"渲染感",
|
||
"小孩",
|
||
"儿童",
|
||
"未成年",
|
||
"幼年",
|
||
"儿童脸",
|
||
"童颜",
|
||
"婴儿脸",
|
||
"baby",
|
||
"child",
|
||
"teen",
|
||
"外国人",
|
||
"欧美人",
|
||
"裁剪",
|
||
"截断",
|
||
"背影",
|
||
"模糊",
|
||
"低分辨率",
|
||
"多余肢体",
|
||
"畸形",
|
||
"裸体",
|
||
"裸露",
|
||
"不穿衣服",
|
||
"不穿上装",
|
||
"不穿下装",
|
||
"nude",
|
||
"nsfw",
|
||
"透明衣服",
|
||
"整只眼球变色",
|
||
"眼神空洞",
|
||
"目光游离",
|
||
"不看镜头",
|
||
"侧脸",
|
||
"斜视",
|
||
"眼神呆滞",
|
||
"表情僵硬",
|
||
"死鱼眼",
|
||
"无神的眼睛",
|
||
"looking away",
|
||
"avoiding eye contact",
|
||
"blank stare",
|
||
"dull eyes",
|
||
"expressionless",
|
||
"lifeless eyes",
|
||
"no eye contact",
|
||
"looking down",
|
||
"looking up",
|
||
"looking left",
|
||
"looking right",
|
||
"face turned away",
|
||
"鱼眼",
|
||
"广角畸变",
|
||
"透视错误",
|
||
"极近距离拍摄",
|
||
"俯拍",
|
||
"仰拍",
|
||
"面部过度前伸",
|
||
"头部过大",
|
||
"头身比例失调",
|
||
"近大远小",
|
||
"身体被压缩",
|
||
"脸部变形",
|
||
"夸张透视",
|
||
"不自然的姿态",
|
||
]
|
||
return {
|
||
"prompt": "，".join([p for p in positive_parts if p]),
|
||
"negative_prompt": "，".join(negative_parts),
|
||
}
|
||
|
||
|
||
def _upload_to_oss(file_bytes: bytes, object_name: str) -> str:
|
||
auth = oss2.Auth(settings.ALIYUN_OSS_ACCESS_KEY_ID, settings.ALIYUN_OSS_ACCESS_KEY_SECRET)
|
||
endpoint = settings.ALIYUN_OSS_ENDPOINT.rstrip("/")
|
||
bucket = oss2.Bucket(auth, endpoint, settings.ALIYUN_OSS_BUCKET_NAME)
|
||
bucket.put_object(object_name, file_bytes)
|
||
cdn = settings.ALIYUN_OSS_CDN_DOMAIN
|
||
if cdn:
|
||
return f"{cdn.rstrip('/')}/{object_name}"
|
||
return f"https://{settings.ALIYUN_OSS_BUCKET_NAME}.{endpoint.replace('https://', '').replace('http://', '')}/{object_name}"
|
||
|
||
|
||
def _delete_old_images(lover_id: int, current_url: Optional[str]):
|
||
"""删除该 lover 下除 current_url 之外的 OSS 图片，前缀 lover/{lover_id}/images/。失败时忽略。"""
|
||
try:
|
||
auth = oss2.Auth(settings.ALIYUN_OSS_ACCESS_KEY_ID, settings.ALIYUN_OSS_ACCESS_KEY_SECRET)
|
||
endpoint = settings.ALIYUN_OSS_ENDPOINT.rstrip("/")
|
||
bucket = oss2.Bucket(auth, endpoint, settings.ALIYUN_OSS_BUCKET_NAME)
|
||
prefix = f"lover/{lover_id}/images/"
|
||
|
||
current_key = None
|
||
if current_url:
|
||
parsed = urlparse(current_url)
|
||
path = parsed.path.lstrip("/")
|
||
if path.startswith(prefix):
|
||
current_key = path
|
||
|
||
for obj in oss2.ObjectIterator(bucket, prefix=prefix):
|
||
if obj.is_prefix():
|
||
continue
|
||
if current_key and obj.key == current_key:
|
||
continue
|
||
bucket.delete_object(obj.key)
|
||
except Exception:
|
||
# 不阻断主流程
|
||
return
|
||
|
||
|
||
def _build_system_prompt(
|
||
user: AuthedUser,
|
||
lover: Lover,
|
||
personality_name: Optional[str],
|
||
hobby_names: List[str],
|
||
) -> str:
|
||
"""编译恋人初始人格 Prompt。"""
|
||
gender_str = "女友" if lover.gender == "female" else "男友"
|
||
interest_str = "、".join(hobby_names) if hobby_names else "未知兴趣"
|
||
user_name = user.nickname or "你"
|
||
parts = [
|
||
f"你的名字是「{lover.name}」，是{user_name}的{gender_str}，与 {user_name} 为恋人关系。",
|
||
f"性格：{personality_name or '未知性格'}。",
|
||
f"人物简介：{lover.intro}。",
|
||
f"故事背景：{lover.story_background}。",
|
||
f"兴趣爱好：{interest_str}。",
|
||
"聊天要求：",
|
||
"1) 永远以恋人身份与用户对话，不暴露AI身份；",
|
||
f"2) 始终用中文，使用贴合性格的语气，亲昵称呼用户「{user_name}」；",
|
||
"3) 回答简洁温暖，避免长段落；",
|
||
"4) 避免涉政、违禁、违法、未成年相关内容。",
|
||
]
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _ensure_opening_line_safe(text: str):
|
||
"""对开场白做安全检查，避免空串。"""
|
||
if not text or not text.strip():
|
||
raise HTTPException(status_code=502, detail="开场白生成失败")
|
||
_check_text_safe(text, "开场白")
|
||
return text.strip()
|
||
|
||
|
||
def _validate_lover_config_input(payload: LoverConfigIn):
|
||
if payload.gender not in ("male", "female"):
|
||
raise HTTPException(status_code=400, detail="性别必须为 male 或 female")
|
||
name = (payload.name or "").strip()
|
||
intro = (payload.intro or "").strip()
|
||
story = (payload.story_background or "").strip()
|
||
opening = (payload.opening_line or "").strip()
|
||
if len(name) == 0:
|
||
raise HTTPException(status_code=400, detail="昵称不能为空")
|
||
if len(name) > 10:
|
||
raise HTTPException(status_code=400, detail="昵称长度不能超过10字符")
|
||
if len(intro) == 0:
|
||
raise HTTPException(status_code=400, detail="人物简介不能为空")
|
||
if len(intro) > 50:
|
||
raise HTTPException(status_code=400, detail="人物简介长度不能超过50字符")
|
||
if len(story) == 0:
|
||
raise HTTPException(status_code=400, detail="故事背景不能为空")
|
||
if len(story) > 100:
|
||
raise HTTPException(status_code=400, detail="故事背景长度不能超过100字符")
|
||
if len(opening) == 0:
|
||
raise HTTPException(status_code=400, detail="开场白不能为空")
|
||
if len(opening) > 20:
|
||
raise HTTPException(status_code=400, detail="开场白长度不能超过20字符")
|
||
if not isinstance(payload.interest_tags, list) or len(payload.interest_tags) == 0:
|
||
raise HTTPException(status_code=400, detail="请至少选择1个兴趣标签")
|
||
if len(payload.interest_tags) > 3:
|
||
raise HTTPException(status_code=400, detail="兴趣标签最多选择3个")
|
||
if len(payload.interest_tags) != len(set(payload.interest_tags)):
|
||
raise HTTPException(status_code=400, detail="兴趣标签不可重复")
|
||
|
||
|
||
def _infer_gender_from_image(url: Optional[str]) -> Optional[str]:
|
||
"""从文件名后缀推断性别，如 xxx_male.png / xxx_female.png。未匹配返回 None。"""
|
||
if not url:
|
||
return None
|
||
try:
|
||
filename = os.path.basename(urlparse(url).path)
|
||
stem = filename.rsplit(".", 1)[0]
|
||
if stem.endswith("_male"):
|
||
return "male"
|
||
if stem.endswith("_female"):
|
||
return "female"
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _normalize_wan26_size(size: Optional[str]) -> str:
|
||
"""
|
||
wan 系列支持在总像素/宽高比约束内自定义尺寸，这里仅做格式/正数校验。
|
||
"""
|
||
default = "960*1280"
|
||
if not size:
|
||
return default
|
||
try:
|
||
width_str, height_str = str(size).lower().split("*", 1)
|
||
width = int(width_str)
|
||
height = int(height_str)
|
||
if width > 0 and height > 0:
|
||
return f"{width}*{height}"
|
||
except Exception:
|
||
return default
|
||
return default
|
||
|
||
|
||
def _call_wanx_v25(prompt: str, negative_prompt: Optional[str] = None) -> str:
|
||
"""
|
||
调用 wan2.5 文生图，返回图片 URL（有效期24小时）。
|
||
简化：同步轮询任务状态，最多等待60秒。
|
||
"""
|
||
model = settings.IMAGE_GEN_MODEL or "wan2.5-t2i-preview"
|
||
size = settings.IMAGE_GEN_SIZE or "960*1280"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {settings.DASHSCOPE_API_KEY or ''}",
|
||
"X-DashScope-Async": "enable",
|
||
}
|
||
if not settings.DASHSCOPE_API_KEY:
|
||
raise HTTPException(status_code=500, detail="未配置 DASHSCOPE_API_KEY")
|
||
|
||
create_payload: Dict[str, Any] = {
|
||
"model": model,
|
||
"input": {"prompt": prompt},
|
||
"parameters": {"style": "<photography>", "size": size, "n": 1},
|
||
}
|
||
if negative_prompt:
|
||
create_payload["input"]["negative_prompt"] = negative_prompt
|
||
try:
|
||
create_resp = requests.post(
|
||
"https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis",
|
||
headers=headers,
|
||
json=create_payload,
|
||
timeout=10,
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="万相调用失败") from exc
|
||
if create_resp.status_code != 200:
|
||
raise HTTPException(status_code=502, detail="万相创建任务失败")
|
||
task_info = create_resp.json()
|
||
task_id = task_info.get("output", {}).get("task_id") or task_info.get("task_id")
|
||
if not task_id:
|
||
raise HTTPException(status_code=502, detail="万相未返回任务ID")
|
||
|
||
# 轮询任务
|
||
query_url = f"https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}"
|
||
deadline = time.time() + 60
|
||
while time.time() < deadline:
|
||
time.sleep(2)
|
||
try:
|
||
query_resp = requests.get(query_url, headers=headers, timeout=8)
|
||
except Exception:
|
||
continue
|
||
if query_resp.status_code != 200:
|
||
continue
|
||
data = query_resp.json()
|
||
status_str = str(
|
||
data.get("output", {}).get("task_status")
|
||
or data.get("task_status")
|
||
or data.get("status")
|
||
).lower()
|
||
if status_str in ("succeeded", "success", "succeed", "finished"):
|
||
urls = (
|
||
data.get("output", {}).get("results")
|
||
or data.get("output", {}).get("result")
|
||
or data.get("output", {}).get("images")
|
||
)
|
||
if isinstance(urls, list) and urls:
|
||
first = urls[0]
|
||
if isinstance(first, dict):
|
||
return first.get("url") or first.get("image_url")
|
||
if isinstance(first, str):
|
||
return first
|
||
raise HTTPException(status_code=502, detail="万相返回空结果")
|
||
if status_str in ("failed", "error", "canceled"):
|
||
err = data.get("output", {}).get("message") or data.get("message") or "生成失败"
|
||
raise HTTPException(status_code=502, detail=f"万相生成失败: {err}")
|
||
raise HTTPException(status_code=504, detail="万相生成超时")
|
||
|
||
|
||
def _call_wanx_v26(prompt: str, negative_prompt: Optional[str] = None) -> str:
|
||
"""
|
||
调用 wan2.6 文生图（HTTP 同步接口），返回图片 URL。
|
||
"""
|
||
if not settings.DASHSCOPE_API_KEY:
|
||
raise HTTPException(status_code=500, detail="未配置 DASHSCOPE_API_KEY")
|
||
|
||
size = _normalize_wan26_size(settings.IMAGE_GEN_SIZE)
|
||
model = settings.IMAGE_GEN_MODEL or "wan2.6-t2i"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {settings.DASHSCOPE_API_KEY}",
|
||
}
|
||
payload: Dict[str, Any] = {
|
||
"model": model,
|
||
"input": {
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": [{"text": prompt}],
|
||
}
|
||
]
|
||
},
|
||
"parameters": {
|
||
"prompt_extend": True,
|
||
"watermark": False,
|
||
"n": 1,
|
||
"size": size,
|
||
},
|
||
}
|
||
if negative_prompt:
|
||
payload["parameters"]["negative_prompt"] = negative_prompt
|
||
|
||
try:
|
||
resp = requests.post(
|
||
"https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=20,
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="万相接口不可用") from exc
|
||
|
||
if resp.status_code != 200:
|
||
msg = resp.text
|
||
try:
|
||
body = resp.json()
|
||
msg = body.get("message") or msg
|
||
except Exception:
|
||
pass
|
||
raise HTTPException(status_code=502, detail=f"万相生成失败: {msg}")
|
||
|
||
try:
|
||
data = resp.json()
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="万相返回解析失败") from exc
|
||
|
||
choices = data.get("output", {}).get("choices") or []
|
||
if not choices:
|
||
raise HTTPException(status_code=502, detail="万相返回空结果")
|
||
first = choices[0]
|
||
contents = first.get("message", {}).get("content") or []
|
||
for item in contents:
|
||
if isinstance(item, dict) and item.get("type") == "image":
|
||
url = item.get("image")
|
||
if url:
|
||
return url
|
||
raise HTTPException(status_code=502, detail="万相返回缺少图片 URL")
|
||
|
||
|
||
def _call_wanx_v26_async(prompt: str, negative_prompt: Optional[str] = None) -> str:
|
||
"""
|
||
调用 wan2.6 文生图（HTTP 异步接口），返回图片 URL。
|
||
流程：创建任务 -> 轮询 task_id。
|
||
"""
|
||
if not settings.DASHSCOPE_API_KEY:
|
||
raise HTTPException(status_code=500, detail="未配置 DASHSCOPE_API_KEY")
|
||
|
||
size = _normalize_wan26_size(settings.IMAGE_GEN_SIZE)
|
||
model = settings.IMAGE_GEN_MODEL or "wan2.6-t2i"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {settings.DASHSCOPE_API_KEY}",
|
||
"X-DashScope-Async": "enable",
|
||
}
|
||
payload: Dict[str, Any] = {
|
||
"model": model,
|
||
"input": {
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": [{"text": prompt}],
|
||
}
|
||
]
|
||
},
|
||
"parameters": {
|
||
"prompt_extend": True,
|
||
"watermark": False,
|
||
"n": 1,
|
||
"size": size,
|
||
},
|
||
}
|
||
if negative_prompt:
|
||
payload["parameters"]["negative_prompt"] = negative_prompt
|
||
|
||
try:
|
||
create_resp = requests.post(
|
||
"https://dashscope.aliyuncs.com/api/v1/services/aigc/image-generation/generation",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=10,
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="万相接口不可用") from exc
|
||
|
||
if create_resp.status_code != 200:
|
||
msg = create_resp.text
|
||
try:
|
||
body = create_resp.json()
|
||
msg = body.get("message") or msg
|
||
except Exception:
|
||
pass
|
||
raise HTTPException(status_code=502, detail=f"万相创建任务失败: {msg}")
|
||
|
||
try:
|
||
task_info = create_resp.json()
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="万相返回解析失败") from exc
|
||
|
||
task_id = task_info.get("output", {}).get("task_id") or task_info.get("task_id")
|
||
if not task_id:
|
||
raise HTTPException(status_code=502, detail="万相未返回任务ID")
|
||
|
||
query_url = f"https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}"
|
||
deadline = time.time() + 60
|
||
while time.time() < deadline:
|
||
time.sleep(2)
|
||
try:
|
||
query_resp = requests.get(query_url, headers={"Authorization": f"Bearer {settings.DASHSCOPE_API_KEY}"}, timeout=8)
|
||
except Exception:
|
||
continue
|
||
if query_resp.status_code != 200:
|
||
continue
|
||
try:
|
||
data = query_resp.json()
|
||
except Exception:
|
||
continue
|
||
output = data.get("output") or {}
|
||
status_str = str(
|
||
output.get("task_status")
|
||
or data.get("task_status")
|
||
or data.get("status")
|
||
).lower()
|
||
if status_str in ("succeeded", "success", "succeed", "finished"):
|
||
# 优先解析 choices -> message -> content -> image
|
||
choices = output.get("choices") or []
|
||
if choices:
|
||
contents = choices[0].get("message", {}).get("content") or []
|
||
for item in contents:
|
||
if isinstance(item, dict) and item.get("type") == "image":
|
||
url = item.get("image")
|
||
if url:
|
||
return url
|
||
# 兜底解析 results / images
|
||
urls = output.get("results") or output.get("result") or output.get("images")
|
||
if isinstance(urls, list) and urls:
|
||
first = urls[0]
|
||
if isinstance(first, dict):
|
||
return first.get("url") or first.get("image_url")
|
||
if isinstance(first, str):
|
||
return first
|
||
raise HTTPException(status_code=502, detail="万相返回空结果")
|
||
if status_str in ("failed", "error", "canceled"):
|
||
err = output.get("message") or data.get("message") or "生成失败"
|
||
raise HTTPException(status_code=502, detail=f"万相生成失败: {err}")
|
||
raise HTTPException(status_code=504, detail="万相生成超时")
|
||
|
||
|
||
def _call_wanx(prompt: str, negative_prompt: Optional[str] = None) -> str:
|
||
"""
|
||
根据模型选择 wan2.5 或 wan2.6 调用。
|
||
- IMAGE_GEN_MODEL 含 "wan2.6" 时走新版接口，WAN26_ASYNC=true 使用异步任务，否则使用同步。
|
||
- 其他情况默认使用 wan2.5 预览版。
|
||
"""
|
||
model = settings.IMAGE_GEN_MODEL or ""
|
||
if "wan2.6" in model:
|
||
if settings.WAN26_ASYNC:
|
||
return _call_wanx_v26_async(prompt, negative_prompt=negative_prompt)
|
||
return _call_wanx_v26(prompt, negative_prompt=negative_prompt)
|
||
return _call_wanx_v25(prompt, negative_prompt=negative_prompt)
|
||
|
||
|
||
def _detect_image_quality(image_url: str):
|
||
"""
|
||
使用 animate-anyone-detect-gen2 校验生成图是否合格（正脸/全身/无遮挡等）。
|
||
未通过时抛 400，提示调整描述；检测异常抛 502。
|
||
"""
|
||
if not settings.DASHSCOPE_API_KEY:
|
||
raise HTTPException(status_code=500, detail="未配置 DASHSCOPE_API_KEY")
|
||
payload = {
|
||
"model": settings.IMAGE_QUALITY_MODEL or "animate-anyone-detect-gen2",
|
||
"input": {"image_url": image_url},
|
||
}
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {settings.DASHSCOPE_API_KEY}",
|
||
}
|
||
try:
|
||
resp = requests.post(
|
||
"https://dashscope.aliyuncs.com/api/v1/services/aigc/image2video/aa-detect",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=10,
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="生成图片检测失败") from exc
|
||
if resp.status_code != 200:
|
||
msg = resp.text
|
||
try:
|
||
msg = resp.json().get("message") or msg
|
||
except Exception:
|
||
pass
|
||
raise HTTPException(status_code=502, detail=f"生成图片检测失败[{resp.status_code}]: {msg}")
|
||
try:
|
||
data = resp.json()
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="生成图片检测结果解析失败") from exc
|
||
output = data.get("output") or {}
|
||
if output.get("check_pass") is True:
|
||
return
|
||
reason = output.get("message") or "请修改着装/外貌特征中的描述。"
|
||
raise HTTPException(status_code=400, detail=reason)
|
||
|
||
|
||
@router.get("/config", response_model=ApiResponse[LoverConfigResponse])
|
||
def get_lover_config(
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
# Stage 2 及以上均可读取配置（Stage 4 也需要读）
|
||
if (user.reg_step or 0) < 2:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
return success_response(_build_lover_config_response(db, user))
|
||
|
||
|
||
@router.get("/basic", response_model=ApiResponse[LoverBasicResponse])
|
||
def get_lover_basic(
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
返回恋人名称与当前形象 URL 以及已选服饰 ID，供前端换装页默认回显。
|
||
"""
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
# 保持成功响应结构，data 返回 null 便于前端按需处理
|
||
return success_response(None)
|
||
|
||
return success_response(LoverBasicResponse(
|
||
lover_id=lover.id,
|
||
name=lover.name,
|
||
image_url=lover.image_url,
|
||
gender=lover.gender
|
||
))
|
||
|
||
|
||
@router.post("/config", response_model=ApiResponse[LoverOut])
|
||
def save_lover_config(
|
||
payload: LoverConfigIn,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
if user.reg_step != 2:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
_validate_lover_config_input(payload)
|
||
gender = payload.gender or _opposite_gender(user.gender)
|
||
|
||
# 违禁词校验
|
||
_check_text_safe(payload.name, "昵称")
|
||
_check_text_safe(payload.intro, "人物简介")
|
||
_check_text_safe(payload.story_background, "故事背景")
|
||
_check_text_safe(payload.opening_line, "开场白")
|
||
|
||
# 校验性格模板
|
||
mould = (
|
||
db.query(GirlfriendMould.id)
|
||
.filter(
|
||
GirlfriendMould.id == payload.personality_tag,
|
||
(GirlfriendMould.gender == gender) | (GirlfriendMould.gender.is_(None)),
|
||
)
|
||
.first()
|
||
)
|
||
if not mould:
|
||
raise HTTPException(status_code=400, detail="性格模板不存在")
|
||
|
||
# 校验兴趣标签
|
||
if payload.interest_tags:
|
||
found_hobbies = (
|
||
db.query(GirlfriendHobbies.id)
|
||
.filter(GirlfriendHobbies.id.in_(payload.interest_tags))
|
||
.all()
|
||
)
|
||
found_ids = {item.id for item in found_hobbies}
|
||
missing = set(payload.interest_tags) - found_ids
|
||
if missing:
|
||
raise HTTPException(status_code=400, detail="兴趣标签不存在或无效")
|
||
else:
|
||
raise HTTPException(status_code=400, detail="请至少选择1个兴趣标签")
|
||
|
||
# 锁定用户行，避免并发创建多个恋人
|
||
from ..models import User # avoid circular import
|
||
user_row = db.query(User).filter(User.id == user.id).with_for_update().first()
|
||
|
||
# 获取或创建恋人（有唯一索引时可并发兜底）
|
||
lover = _get_or_create_lover(db, user)
|
||
|
||
# 音色必须已选择且与性别匹配（通过音色页设置）
|
||
existing_voice = None
|
||
if lover.voice_id:
|
||
existing_voice = (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.id == lover.voice_id, VoiceLibrary.gender == gender)
|
||
.first()
|
||
)
|
||
if not existing_voice:
|
||
default_voice = _get_default_voice(db, gender)
|
||
if default_voice and lover.voice_id == default_voice.id:
|
||
existing_voice = default_voice
|
||
if not existing_voice:
|
||
raise HTTPException(status_code=400, detail="请先在音色页选择与性别匹配的音色")
|
||
|
||
lover.gender = gender
|
||
lover.name = payload.name
|
||
lover.intro = payload.intro
|
||
lover.story_background = payload.story_background
|
||
lover.personality_tag = payload.personality_tag
|
||
lover.interest_tags = payload.interest_tags
|
||
lover.opening_line = payload.opening_line
|
||
# 同步独立列
|
||
# 外貌相关由外貌接口维护，此处不覆盖
|
||
lover.voice_id = existing_voice.id if existing_voice else None
|
||
|
||
# 必须已生成形象
|
||
if not lover.image_url:
|
||
raise HTTPException(status_code=400, detail="请先生成并确认形象")
|
||
|
||
# 每日生成计数的自然日重置
|
||
today = date.today()
|
||
if lover.image_gen_reset_date != today:
|
||
lover.image_gen_reset_date = today
|
||
lover.image_gen_used = 0
|
||
|
||
db.add(lover)
|
||
# 更新用户阶段（若存在）
|
||
if user_row:
|
||
user_row.reg_step = 3
|
||
db.add(user_row)
|
||
|
||
db.flush() # 保证 lover.id 可用
|
||
|
||
# 清理旧图（保留当前 image_url）
|
||
_delete_old_images(lover.id, lover.image_url)
|
||
|
||
return success_response(lover)
|
||
|
||
|
||
@router.get("/options", response_model=ApiResponse[GenderOptionsResponse])
|
||
def get_gender_options(
|
||
gender: str = Query(..., pattern="^(male|female)$"),
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
根据性别获取性格模板和音色列表（含默认/已选）。
|
||
用于前端切换性别时刷新选项。
|
||
"""
|
||
if (user.reg_step or 0) < 2:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
voices = (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.gender == gender)
|
||
.order_by(VoiceLibrary.id.asc())
|
||
.all()
|
||
)
|
||
default_voice = _get_default_voice(db, gender)
|
||
personality_options = (
|
||
db.query(GirlfriendMould)
|
||
.filter(
|
||
(GirlfriendMould.gender == gender) | (GirlfriendMould.gender.is_(None))
|
||
)
|
||
.order_by(GirlfriendMould.weigh.desc(), GirlfriendMould.id.asc())
|
||
.all()
|
||
)
|
||
selected_voice_id = None
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if lover and lover.voice_id and lover.gender == gender:
|
||
selected_voice_id = lover.voice_id
|
||
|
||
return success_response(
|
||
GenderOptionsResponse(
|
||
gender=gender,
|
||
voices=voices,
|
||
default_voice_id=default_voice.id if default_voice else None,
|
||
selected_voice_id=selected_voice_id,
|
||
personality_options=[
|
||
OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in personality_options
|
||
],
|
||
)
|
||
)
|
||
|
||
|
||
@router.put("/voice", response_model=ApiResponse[LoverOut])
|
||
def update_lover_voice(
|
||
payload: LoverVoiceUpdate,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未初始化")
|
||
|
||
gender = payload.gender or lover.gender or _opposite_gender(user.gender)
|
||
voice = (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.id == payload.voice_id, VoiceLibrary.gender == gender)
|
||
.first()
|
||
)
|
||
if not voice:
|
||
voice = _get_default_voice(db, gender)
|
||
if not voice:
|
||
raise HTTPException(status_code=400, detail="未找到匹配音色")
|
||
|
||
# 若已有形象且文件名包含性别标记，与切换后的性别不一致时阻断，避免“男图女设”
|
||
img_gender = _infer_gender_from_image(lover.image_url)
|
||
if img_gender and img_gender != gender:
|
||
raise HTTPException(status_code=400, detail="当前形象与切换后的性别不匹配，请先生成该性别的形象")
|
||
|
||
lover.gender = gender
|
||
lover.voice_id = voice.id
|
||
db.add(lover)
|
||
db.flush()
|
||
return success_response(lover)
|
||
|
||
|
||
@router.put("/voice/simple", response_model=ApiResponse[LoverOut])
|
||
def update_lover_voice_simple(
|
||
payload: LoverVoiceSwitch,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
简化版更换音色：只需传 voice_id，自动使用当前恋人性别。
|
||
"""
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未初始化")
|
||
|
||
gender = lover.gender or _opposite_gender(user.gender)
|
||
voice = (
|
||
db.query(VoiceLibrary)
|
||
.filter(VoiceLibrary.id == payload.voice_id, VoiceLibrary.gender == gender)
|
||
.first()
|
||
)
|
||
if not voice:
|
||
raise HTTPException(status_code=404, detail="音色不存在或与恋人性别不匹配")
|
||
|
||
img_gender = _infer_gender_from_image(lover.image_url)
|
||
if img_gender and img_gender != gender:
|
||
raise HTTPException(status_code=400, detail="当前形象与切换后的性别不匹配，请先生成该性别的形象")
|
||
|
||
lover.voice_id = voice.id
|
||
db.add(lover)
|
||
db.flush()
|
||
return success_response(lover, msg="音色更新成功")
|
||
|
||
|
||
@router.get("/appearance", response_model=ApiResponse[LoverAppearanceResponse])
|
||
def get_lover_appearance(
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
if user.reg_step != 2:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
lover = _get_or_create_lover(db, user)
|
||
params = AppearanceParams(
|
||
hair_style_id=lover.hair_style_id,
|
||
eye_color_id=lover.eye_color_id,
|
||
outfit_desc=lover.outfit_desc,
|
||
) if any([lover.hair_style_id, lover.eye_color_id, lover.outfit_desc]) else None
|
||
hair_style_id = lover.hair_style_id
|
||
eye_color_id = lover.eye_color_id
|
||
outfit_desc = lover.outfit_desc
|
||
gender = lover.gender
|
||
hair_styles = (
|
||
db.query(GirlfriendHairStyle)
|
||
.filter(
|
||
(GirlfriendHairStyle.gender == gender) | (GirlfriendHairStyle.gender.is_(None))
|
||
)
|
||
.order_by(GirlfriendHairStyle.weigh.desc(), GirlfriendHairStyle.id.asc())
|
||
.all()
|
||
)
|
||
eye_colors = (
|
||
db.query(GirlfriendEyeColor)
|
||
.order_by(GirlfriendEyeColor.weigh.desc(), GirlfriendEyeColor.id.asc())
|
||
.all()
|
||
)
|
||
return success_response(
|
||
LoverAppearanceResponse(
|
||
lover_id=lover.id,
|
||
user_id=lover.user_id,
|
||
appearance_prompt=lover.appearance_prompt,
|
||
appearance_params=params,
|
||
hair_style_id=hair_style_id,
|
||
eye_color_id=eye_color_id,
|
||
outfit_desc=outfit_desc,
|
||
image_url=lover.image_url,
|
||
last_image_task_id=lover.last_image_task_id,
|
||
hair_style_options=[OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in hair_styles],
|
||
eye_color_options=[OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in eye_colors],
|
||
)
|
||
)
|
||
|
||
|
||
@router.post("/appearance", response_model=ApiResponse[LoverAppearanceResponse])
|
||
def save_lover_appearance(
|
||
payload: LoverAppearanceIn,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
if user.reg_step != 2:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
lover = _get_or_create_lover(db, user)
|
||
gender = lover.gender
|
||
# 长度校验改为显式 400，避免 422
|
||
if payload.appearance_prompt and len(payload.appearance_prompt.strip()) > 100:
|
||
raise HTTPException(status_code=400, detail="外貌特征长度不能超过100字符")
|
||
if payload.appearance_params and payload.appearance_params.outfit_desc:
|
||
if len(payload.appearance_params.outfit_desc.strip()) > 50:
|
||
raise HTTPException(status_code=400, detail="描述着装长度不能超过50字符")
|
||
|
||
_check_text_safe(payload.appearance_prompt, "外貌特征")
|
||
if payload.appearance_params and payload.appearance_params.outfit_desc:
|
||
_check_text_safe(payload.appearance_params.outfit_desc, "描述着装")
|
||
|
||
# 校验发型/瞳色
|
||
if payload.appearance_params:
|
||
if payload.appearance_params.hair_style_id:
|
||
exists = (
|
||
db.query(GirlfriendHairStyle.id)
|
||
.filter(
|
||
GirlfriendHairStyle.id == payload.appearance_params.hair_style_id,
|
||
(GirlfriendHairStyle.gender == gender) | (GirlfriendHairStyle.gender.is_(None)),
|
||
)
|
||
.first()
|
||
)
|
||
if not exists:
|
||
raise HTTPException(status_code=400, detail="发型不存在")
|
||
if payload.appearance_params.eye_color_id:
|
||
exists = (
|
||
db.query(GirlfriendEyeColor.id)
|
||
.filter(GirlfriendEyeColor.id == payload.appearance_params.eye_color_id)
|
||
.first()
|
||
)
|
||
if not exists:
|
||
raise HTTPException(status_code=400, detail="瞳色不存在")
|
||
|
||
# 长度校验已在 Pydantic，若需要进一步限制可添加
|
||
lover.appearance_prompt = payload.appearance_prompt
|
||
lover.hair_style_id = payload.appearance_params.hair_style_id if payload.appearance_params else None
|
||
lover.eye_color_id = payload.appearance_params.eye_color_id if payload.appearance_params else None
|
||
lover.outfit_desc = payload.appearance_params.outfit_desc if payload.appearance_params else None
|
||
|
||
db.add(lover)
|
||
db.flush()
|
||
|
||
params = AppearanceParams(
|
||
hair_style_id=lover.hair_style_id,
|
||
eye_color_id=lover.eye_color_id,
|
||
outfit_desc=lover.outfit_desc,
|
||
) if any([lover.hair_style_id, lover.eye_color_id, lover.outfit_desc]) else None
|
||
hair_style_id = lover.hair_style_id
|
||
eye_color_id = lover.eye_color_id
|
||
outfit_desc = lover.outfit_desc
|
||
hair_styles = (
|
||
db.query(GirlfriendHairStyle)
|
||
.filter(
|
||
(GirlfriendHairStyle.gender == gender) | (GirlfriendHairStyle.gender.is_(None))
|
||
)
|
||
.order_by(GirlfriendHairStyle.weigh.desc(), GirlfriendHairStyle.id.asc())
|
||
.all()
|
||
)
|
||
eye_colors = (
|
||
db.query(GirlfriendEyeColor)
|
||
.order_by(GirlfriendEyeColor.weigh.desc(), GirlfriendEyeColor.id.asc())
|
||
.all()
|
||
)
|
||
return success_response(
|
||
LoverAppearanceResponse(
|
||
lover_id=lover.id,
|
||
user_id=lover.user_id,
|
||
appearance_prompt=lover.appearance_prompt,
|
||
appearance_params=params,
|
||
hair_style_id=hair_style_id,
|
||
eye_color_id=eye_color_id,
|
||
outfit_desc=outfit_desc,
|
||
image_url=lover.image_url,
|
||
last_image_task_id=lover.last_image_task_id,
|
||
hair_style_options=[OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in hair_styles],
|
||
eye_color_options=[OptionOut(id=item.id, name=item.name, weigh=item.weigh) for item in eye_colors],
|
||
)
|
||
)
|
||
|
||
|
||
@router.post("/appearance/generate", response_model=ApiResponse[GenerateImageResponse])
|
||
def generate_lover_image(
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
if user.reg_step != 2:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
lover = _get_or_create_lover(db, user)
|
||
_reset_image_quota_if_needed(lover)
|
||
if lover.image_gen_used >= (lover.image_gen_limit or 0):
|
||
raise HTTPException(status_code=400, detail="生成次数已用完")
|
||
|
||
hair_style_id = lover.hair_style_id
|
||
eye_color_id = lover.eye_color_id
|
||
outfit_desc = lover.outfit_desc
|
||
if not hair_style_id or not eye_color_id or not outfit_desc or not lover.appearance_prompt:
|
||
raise HTTPException(status_code=400, detail="请先填写并保存发型、瞳色、着装描述和外貌特征")
|
||
|
||
# 违禁词校验
|
||
_check_text_safe(lover.appearance_prompt, "外貌特征")
|
||
_check_text_safe(outfit_desc, "描述着装")
|
||
|
||
hair = db.query(GirlfriendHairStyle).filter(GirlfriendHairStyle.id == hair_style_id).first()
|
||
eye = db.query(GirlfriendEyeColor).filter(GirlfriendEyeColor.id == eye_color_id).first()
|
||
if not hair:
|
||
raise HTTPException(status_code=400, detail="发型不存在")
|
||
if not eye:
|
||
raise HTTPException(status_code=400, detail="瞳色不存在")
|
||
|
||
# wan2.6 使用定制提示词以强化眼神/朝向；默认沿用 2.5 提示词
|
||
prompt_builder = _build_prompt_v26 if "wan2.6" in (settings.IMAGE_GEN_MODEL or "") else _build_prompt
|
||
prompt_obj = prompt_builder(lover, hair.name, eye.name, outfit_desc)
|
||
prompt = prompt_obj["prompt"]
|
||
negative_prompt = prompt_obj.get("negative_prompt")
|
||
|
||
# 幂等/并发控制：存在 pending 任务则拒绝
|
||
existing_pending = (
|
||
db.query(GenerationTask)
|
||
.filter(
|
||
GenerationTask.user_id == user.id,
|
||
GenerationTask.task_type == "image",
|
||
GenerationTask.status.in_(("pending", "running")),
|
||
)
|
||
.first()
|
||
)
|
||
if existing_pending:
|
||
raise HTTPException(status_code=409, detail="已有生成任务进行中，请稍后再试")
|
||
|
||
# 创建生成任务记录
|
||
task = GenerationTask(
|
||
user_id=user.id,
|
||
lover_id=lover.id,
|
||
task_type="image",
|
||
status="pending",
|
||
payload={
|
||
"prompt": prompt,
|
||
"negative_prompt": negative_prompt,
|
||
"gender": lover.gender,
|
||
"hair_style_id": hair_style_id,
|
||
"hair_style_name": hair.name,
|
||
"eye_color_id": eye_color_id,
|
||
"eye_color_name": eye.name,
|
||
"outfit_desc": outfit_desc,
|
||
"style": "<photography>",
|
||
"size": settings.IMAGE_GEN_SIZE or "960*1280",
|
||
"model": settings.IMAGE_GEN_MODEL or "wan2.5-t2i-preview",
|
||
},
|
||
created_at=datetime.utcnow(),
|
||
updated_at=datetime.utcnow(),
|
||
)
|
||
db.add(task)
|
||
db.flush()
|
||
|
||
image_url = None
|
||
try:
|
||
# 将任务标记为 running，避免长时间 pending
|
||
task.status = "running"
|
||
task.updated_at = datetime.utcnow()
|
||
db.add(task)
|
||
db.flush()
|
||
|
||
temp_url = _call_wanx(prompt, negative_prompt=negative_prompt)
|
||
# 生成后先做形象检测，未通过则提示用户调整描述
|
||
_detect_image_quality(temp_url)
|
||
# 下载图片
|
||
img_resp = requests.get(temp_url, timeout=15)
|
||
if img_resp.status_code != 200:
|
||
raise HTTPException(status_code=502, detail="生成图片下载失败")
|
||
suffix = lover.gender or "unknown"
|
||
object_name = f"lover/{lover.id}/images/{int(time.time())}_{suffix}.png"
|
||
image_url = _upload_to_oss(img_resp.content, object_name)
|
||
|
||
lover.image_url = image_url
|
||
lover.last_image_task_id = task.id
|
||
lover.image_gen_used = (lover.image_gen_used or 0) + 1
|
||
lover.image_gen_reset_date = date.today()
|
||
|
||
task.status = "succeeded"
|
||
task.result_url = image_url
|
||
task.updated_at = datetime.utcnow()
|
||
db.add(task)
|
||
db.add(lover)
|
||
db.flush()
|
||
except HTTPException:
|
||
db.rollback()
|
||
raise
|
||
except Exception as exc:
|
||
db.rollback()
|
||
raise HTTPException(status_code=502, detail="生成图片失败") from exc
|
||
|
||
return success_response(GenerateImageResponse(task_id=task.id, image_url=image_url))
|
||
|
||
|
||
@router.post("/init", response_model=ApiResponse[LoverInitResponse])
|
||
def init_lover_chat(
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
生成专属恋人：编译人格 Prompt，直接落库用户填写的开场白为首条消息，并把 reg_step 置为 4。
|
||
"""
|
||
if (user.reg_step or 0) < 3:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="请先完成前置步骤")
|
||
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未找到")
|
||
|
||
existing_session = (
|
||
db.query(ChatSession)
|
||
.filter(ChatSession.user_id == user.id, ChatSession.lover_id == lover.id)
|
||
.with_for_update()
|
||
.order_by(ChatSession.id.asc())
|
||
.first()
|
||
)
|
||
|
||
# 加载性格/兴趣名称，编译人格 Prompt
|
||
personality = (
|
||
db.query(GirlfriendMould)
|
||
.filter(GirlfriendMould.id == lover.personality_tag)
|
||
.first()
|
||
)
|
||
hobbies = []
|
||
if lover.interest_tags:
|
||
hobbies = (
|
||
db.query(GirlfriendHobbies)
|
||
.filter(GirlfriendHobbies.id.in_(lover.interest_tags))
|
||
.all()
|
||
)
|
||
hobby_names = [h.name for h in hobbies if h and h.name]
|
||
personality_name = personality.name if personality else None
|
||
|
||
system_prompt = _build_system_prompt(user, lover, personality_name, hobby_names)
|
||
|
||
required_fields = [
|
||
("name", lover.name),
|
||
("intro", lover.intro),
|
||
("story_background", lover.story_background),
|
||
("personality_tag", lover.personality_tag),
|
||
("interest_tags", lover.interest_tags),
|
||
("opening_line", lover.opening_line),
|
||
("voice_id", lover.voice_id),
|
||
("image_url", lover.image_url),
|
||
]
|
||
missing = [field for field, val in required_fields if not val]
|
||
if missing:
|
||
raise HTTPException(status_code=400, detail=f"初始化缺少字段: {', '.join(missing)}")
|
||
|
||
# 若已存在会话（可能因上次部分成功），视为幂等成功；如首条消息缺失则补写
|
||
if existing_session:
|
||
has_first_msg = (
|
||
db.query(ChatMessage)
|
||
.filter(ChatMessage.session_id == existing_session.id, ChatMessage.seq == 1)
|
||
.first()
|
||
)
|
||
opening_line = _ensure_opening_line_safe(lover.opening_line or "")
|
||
if not has_first_msg:
|
||
now = datetime.utcnow()
|
||
msg = ChatMessage(
|
||
session_id=existing_session.id,
|
||
user_id=user.id,
|
||
lover_id=lover.id,
|
||
role="lover",
|
||
content_type="text",
|
||
content=opening_line,
|
||
seq=1,
|
||
token_input=None,
|
||
token_output=None,
|
||
model=settings.LLM_MODEL or "qwen-flash",
|
||
created_at=now,
|
||
)
|
||
db.add(msg)
|
||
existing_session.last_message_at = existing_session.last_message_at or now
|
||
existing_session.updated_at = datetime.utcnow()
|
||
db.add(existing_session)
|
||
|
||
# 始终刷新人格 Prompt，避免为空
|
||
lover.personality_prompt = system_prompt
|
||
lover.init_model = settings.LLM_MODEL or "qwen-flash"
|
||
lover.init_at = datetime.utcnow()
|
||
db.add(lover)
|
||
from ..models import User # 延迟导入避免循环
|
||
|
||
user_row = db.query(User).filter(User.id == user.id).first()
|
||
if user_row:
|
||
if (user_row.reg_step or 0) < 4:
|
||
user_row.reg_step = 4
|
||
db.add(user_row)
|
||
final_reg_step = user_row.reg_step or 4
|
||
else:
|
||
final_reg_step = 4
|
||
return success_response(
|
||
LoverInitResponse(
|
||
session_id=existing_session.id,
|
||
opening_line=lover.opening_line or "",
|
||
reg_step=final_reg_step,
|
||
personality_prompt=lover.personality_prompt,
|
||
)
|
||
)
|
||
|
||
# 不调用 LLM，直接使用用户填写的开场白
|
||
opening_line = _ensure_opening_line_safe(lover.opening_line or "")
|
||
|
||
now = datetime.utcnow()
|
||
session = ChatSession(
|
||
user_id=user.id,
|
||
lover_id=lover.id,
|
||
model=settings.LLM_MODEL or "qwen-flash",
|
||
status="active",
|
||
last_message_at=now,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
db.add(session)
|
||
db.flush()
|
||
|
||
# 会话内序号从 1 开始
|
||
msg = ChatMessage(
|
||
session_id=session.id,
|
||
user_id=user.id,
|
||
lover_id=lover.id,
|
||
role="lover",
|
||
content_type="text",
|
||
content=opening_line,
|
||
seq=1,
|
||
token_input=None,
|
||
token_output=None,
|
||
model=settings.LLM_MODEL or "qwen-flash",
|
||
created_at=now,
|
||
)
|
||
db.add(msg)
|
||
|
||
# 更新 lover/personality prompt & reg_step
|
||
lover.personality_prompt = system_prompt
|
||
lover.init_model = settings.LLM_MODEL or "qwen-flash"
|
||
lover.init_at = now
|
||
db.add(lover)
|
||
|
||
from ..models import User # 延迟导入避免循环
|
||
|
||
user_row = db.query(User).filter(User.id == user.id).first()
|
||
final_reg_step = user.reg_step or 1
|
||
if user_row:
|
||
user_row.reg_step = 4
|
||
final_reg_step = 4
|
||
db.add(user_row)
|
||
|
||
return success_response(
|
||
LoverInitResponse(
|
||
session_id=session.id,
|
||
opening_line=opening_line,
|
||
reg_step=final_reg_step,
|
||
personality_prompt=lover.personality_prompt,
|
||
)
|
||
)
|