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": "", "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": "", "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, ) )