845 lines
27 KiB
Python
845 lines
27 KiB
Python
from typing import Dict, List, Optional
|
||
|
||
import json
|
||
import time
|
||
import requests
|
||
import oss2
|
||
from datetime import datetime
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from pydantic import BaseModel, Field, field_validator
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
from ..db import get_db
|
||
from ..deps import AuthedUser, get_current_user
|
||
from decimal import Decimal
|
||
from ..models import Lover, OutfitItem, OutfitLook, User, OutfitPurchaseLog, UserMoneyLog
|
||
from ..response import ApiResponse, success_response
|
||
from ..config import settings
|
||
|
||
router = APIRouter(prefix="/outfit", tags=["outfit"])
|
||
|
||
def _ensure_balance(user_row: User) -> Decimal:
|
||
try:
|
||
return Decimal(str(user_row.money or "0"))
|
||
except Exception:
|
||
return Decimal("0")
|
||
|
||
|
||
def _now_ts() -> int:
|
||
import time
|
||
return int(time.time())
|
||
|
||
|
||
def _parse_owned_outfits(raw: Optional[str]) -> set[int]:
|
||
owned: set[int] = set()
|
||
if not raw:
|
||
return owned
|
||
# 已是列表
|
||
if isinstance(raw, list):
|
||
for v in raw:
|
||
try:
|
||
owned.add(int(v))
|
||
except Exception:
|
||
continue
|
||
return owned
|
||
# JSON 字符串
|
||
if isinstance(raw, str):
|
||
try:
|
||
parsed = json.loads(raw)
|
||
if isinstance(parsed, list):
|
||
for v in parsed:
|
||
try:
|
||
owned.add(int(v))
|
||
except Exception:
|
||
continue
|
||
return owned
|
||
except Exception:
|
||
# 逗号分隔兜底
|
||
for part in str(raw).split(","):
|
||
part = part.strip()
|
||
if part.isdigit():
|
||
owned.add(int(part))
|
||
return owned
|
||
return owned
|
||
|
||
|
||
def _serialize_owned_outfits(ids: set[int]) -> list[int]:
|
||
return sorted(list(ids))
|
||
|
||
|
||
def _cdnize(url: Optional[str]) -> Optional[str]:
|
||
"""
|
||
将以 /uploads 开头的相对路径补全为 CDN/OSS 完整 URL。
|
||
"""
|
||
if not url:
|
||
return url
|
||
if url.startswith("http://") or url.startswith("https://"):
|
||
return url
|
||
|
||
# 优先使用 CDN 域名
|
||
cdn_domain = getattr(settings, 'ALIYUN_OSS_CDN_DOMAIN', None)
|
||
if cdn_domain:
|
||
prefix = cdn_domain.rstrip("/")
|
||
else:
|
||
# 兜底使用 OSS 默认域名
|
||
bucket_name = getattr(settings, 'ALIYUN_OSS_BUCKET_NAME', 'hello12312312')
|
||
endpoint = getattr(settings, 'ALIYUN_OSS_ENDPOINT', 'https://oss-cn-hangzhou.aliyuncs.com')
|
||
endpoint_clean = endpoint.replace('https://', '').replace('http://', '')
|
||
prefix = f"https://{bucket_name}.{endpoint_clean}"
|
||
|
||
if url.startswith("/"):
|
||
return prefix + url
|
||
return f"{prefix}/{url}"
|
||
|
||
|
||
def _clean_url(url: Optional[str]) -> Optional[str]:
|
||
"""去掉首尾空白，避免试衣接口因空格拒绝。"""
|
||
if url is None:
|
||
return None
|
||
return url.strip()
|
||
|
||
|
||
def _upload_to_oss(file_bytes: bytes, object_name: str) -> str:
|
||
"""上传到 OSS，返回可访问 URL（优先 CDN 域名）。"""
|
||
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 _call_aitryon(person_image_url: str, top_url: Optional[str], bottom_url: Optional[str]) -> str:
|
||
"""
|
||
调用 DashScope AI试衣-基础版（aitryon），同步轮询结果，返回 image_url。
|
||
"""
|
||
api_key = (settings.DASHSCOPE_API_KEY or "").strip()
|
||
if not api_key:
|
||
raise HTTPException(status_code=500, detail="未配置 DASHSCOPE_API_KEY")
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
"Authorization": f"Bearer {api_key}",
|
||
"X-DashScope-Async": "enable",
|
||
}
|
||
payload = {
|
||
"model": settings.TRYON_MODEL or "aitryon",
|
||
"input": {
|
||
"person_image_url": _clean_url(person_image_url),
|
||
},
|
||
"parameters": {
|
||
"resolution": -1,
|
||
"restore_face": True,
|
||
},
|
||
}
|
||
if top_url:
|
||
payload["input"]["top_garment_url"] = _clean_url(top_url)
|
||
if bottom_url:
|
||
payload["input"]["bottom_garment_url"] = _clean_url(bottom_url)
|
||
|
||
try:
|
||
create_resp = requests.post(
|
||
"https://dashscope.aliyuncs.com/api/v1/services/aigc/image2image/image-synthesis/",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=10,
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="试衣任务创建失败") from exc
|
||
if create_resp.status_code != 200:
|
||
err_msg = create_resp.text or "试衣任务创建失败"
|
||
raise HTTPException(status_code=502, detail=f"试衣任务创建失败({create_resp.status_code}): {err_msg}")
|
||
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="试衣任务未返回task_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:
|
||
q_resp = requests.get(
|
||
query_url,
|
||
headers={
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Accept": "application/json",
|
||
},
|
||
timeout=8,
|
||
)
|
||
except Exception:
|
||
continue
|
||
if q_resp.status_code != 200:
|
||
continue
|
||
data = q_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"):
|
||
image_url = (
|
||
data.get("output", {}).get("image_url")
|
||
or (data.get("output", {}).get("results") or [{}])[0].get("url")
|
||
)
|
||
if image_url:
|
||
return image_url
|
||
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="试衣任务超时")
|
||
|
||
|
||
class OutfitItemOut(BaseModel):
|
||
id: int
|
||
name: str
|
||
category: str
|
||
gender: str
|
||
image_url: str
|
||
is_free: bool
|
||
price_gold: int
|
||
is_vip_only: bool
|
||
is_owned: bool
|
||
|
||
|
||
class OutfitLookOut(BaseModel):
|
||
id: int
|
||
image_url: str
|
||
|
||
|
||
class OutfitListResponse(BaseModel):
|
||
top: List[OutfitItemOut]
|
||
bottom: List[OutfitItemOut]
|
||
dress: List[OutfitItemOut]
|
||
top_total: int
|
||
bottom_total: int
|
||
dress_total: int
|
||
page: int
|
||
size: int
|
||
balance: float
|
||
clothes_num: int
|
||
outfit_slots: int
|
||
owned_outfit_ids: List[int] = Field(default_factory=list)
|
||
current_outfit: Optional[dict] = None
|
||
looks: List[OutfitLookOut] = Field(default_factory=list)
|
||
|
||
|
||
class OutfitPurchaseIn(BaseModel):
|
||
item_id: int
|
||
# 目前仅小程序，预留字段，未来扩展平台时再打开校验
|
||
platform: Optional[str] = Field(default=None)
|
||
|
||
|
||
class OutfitPurchaseOut(BaseModel):
|
||
item_id: int
|
||
is_owned: bool
|
||
balance: float
|
||
owned_outfit_ids: List[int]
|
||
|
||
|
||
class OutfitChangeIn(BaseModel):
|
||
top_item_id: Optional[int] = None
|
||
bottom_item_id: Optional[int] = None
|
||
dress_item_id: Optional[int] = None
|
||
save_to_look: bool = False
|
||
look_name: Optional[str] = None
|
||
result_image_url: Optional[str] = None # 若前端已有生成结果，可传入落库
|
||
|
||
@field_validator("top_item_id", "bottom_item_id", "dress_item_id", mode="before")
|
||
@classmethod
|
||
def empty_str_to_none(cls, v):
|
||
# 兼容前端传空字符串导致 422 的情况
|
||
if isinstance(v, str) and v.strip() == "":
|
||
return None
|
||
return v
|
||
|
||
|
||
class OutfitChangeOut(BaseModel):
|
||
image_url: Optional[str] = None
|
||
current_outfit: dict
|
||
clothes_num: int
|
||
owned_outfit_ids: List[int]
|
||
looks: List[OutfitLookOut] = Field(default_factory=list)
|
||
|
||
|
||
class OutfitLooksResponse(BaseModel):
|
||
looks: List[OutfitLookOut]
|
||
|
||
|
||
class OutfitUseResponse(BaseModel):
|
||
image_url: str
|
||
current_outfit: dict
|
||
looks: List[OutfitLookOut]
|
||
|
||
|
||
class OutfitMallItem(BaseModel):
|
||
id: int
|
||
name: str
|
||
image_url: str
|
||
price_gold: int
|
||
gender: str
|
||
category: str
|
||
|
||
|
||
class OutfitMallResponse(BaseModel):
|
||
items: List[OutfitMallItem]
|
||
owned_outfit_ids: List[int]
|
||
balance: float
|
||
|
||
|
||
@router.post("/purchase", response_model=ApiResponse[OutfitPurchaseOut])
|
||
def purchase_outfit(
|
||
payload: OutfitPurchaseIn,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
购买服饰：
|
||
- 校验性别匹配(恋人 gender 或 unisex)
|
||
- 免费不可购买
|
||
- 已拥有不可购买
|
||
- 校验金币余额
|
||
"""
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未找到")
|
||
|
||
item = db.query(OutfitItem).filter(OutfitItem.id == payload.item_id, OutfitItem.status == "1").first()
|
||
if not item:
|
||
raise HTTPException(status_code=404, detail="服饰不存在或已下架")
|
||
|
||
# 性别校验
|
||
if item.gender not in ("unisex", lover.gender):
|
||
raise HTTPException(status_code=400, detail="服饰与恋人性别不匹配")
|
||
|
||
# 免费服饰无需购买
|
||
if bool(item.is_free):
|
||
raise HTTPException(status_code=400, detail="该服饰为免费，无需购买")
|
||
|
||
try:
|
||
user_row = (
|
||
db.query(User)
|
||
.filter(User.id == user.id)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
except Exception:
|
||
user_row = None
|
||
if not user_row:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
balance = _ensure_balance(user_row)
|
||
owned_ids = _parse_owned_outfits(user_row.owned_outfit_ids)
|
||
if int(item.id) in owned_ids:
|
||
raise HTTPException(status_code=400, detail="已拥有该服饰，无需重复购买")
|
||
|
||
price = Decimal(item.price_gold or 0)
|
||
if price <= 0:
|
||
raise HTTPException(status_code=400, detail="服饰价格异常")
|
||
if balance < price:
|
||
raise HTTPException(status_code=400, detail="余额不足")
|
||
|
||
# 扣减余额，写拥有数据 & 日志（行锁保证并发安全）
|
||
before_balance = balance
|
||
balance -= price
|
||
user_row.money = float(balance)
|
||
owned_ids.add(int(item.id))
|
||
user_row.owned_outfit_ids = _serialize_owned_outfits(owned_ids)
|
||
db.add(user_row)
|
||
db.add(
|
||
OutfitPurchaseLog(
|
||
user_id=user.id,
|
||
item_id=item.id,
|
||
price_gold=int(price),
|
||
platform="miniapp", # 目前仅小程序
|
||
status="success",
|
||
createtime=_now_ts(),
|
||
updatetime=_now_ts(),
|
||
remark=None,
|
||
)
|
||
)
|
||
db.add(
|
||
UserMoneyLog(
|
||
user_id=user.id,
|
||
money=-price,
|
||
before=before_balance,
|
||
after=Decimal(user_row.money),
|
||
memo=f"购买服饰:{item.name}",
|
||
createtime=_now_ts(),
|
||
)
|
||
)
|
||
try:
|
||
db.flush()
|
||
except IntegrityError:
|
||
db.rollback()
|
||
raise HTTPException(status_code=409, detail="购买请求冲突，请重试")
|
||
|
||
return success_response(
|
||
OutfitPurchaseOut(
|
||
item_id=item.id,
|
||
is_owned=True,
|
||
balance=float(balance),
|
||
owned_outfit_ids=sorted(list(owned_ids)),
|
||
)
|
||
)
|
||
|
||
|
||
@router.post("/change", response_model=ApiResponse[OutfitChangeOut])
|
||
def change_outfit(
|
||
payload: OutfitChangeIn,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
换装接口：
|
||
- 校验换装次数 `clothes_num` > 0
|
||
- 校验所选服饰存在、性别匹配、已拥有或免费
|
||
- 连衣裙(dress) 与 上/下装互斥；上/下装可单选或组合
|
||
- 可选择保存当前形象到形象栏（需有空位）
|
||
- 更新当前穿戴服饰 ID，落库传入的 result_image_url（如未提供则沿用旧图）
|
||
"""
|
||
lover = (
|
||
db.query(Lover)
|
||
.filter(Lover.user_id == user.id)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未找到")
|
||
|
||
user_row = (
|
||
db.query(User)
|
||
.filter(User.id == user.id)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not user_row:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 换装次数校验
|
||
clothes_num = int(user_row.clothes_num or 0)
|
||
if clothes_num <= 0:
|
||
raise HTTPException(status_code=400, detail="换装次数不足")
|
||
|
||
# 选择校验
|
||
top_id = payload.top_item_id
|
||
bottom_id = payload.bottom_item_id
|
||
dress_id = payload.dress_item_id
|
||
if not any([top_id, bottom_id, dress_id]):
|
||
raise HTTPException(status_code=400, detail="请选择至少一个服饰")
|
||
if dress_id and (top_id or bottom_id):
|
||
raise HTTPException(status_code=400, detail="连衣裙/连体服与上/下装不可同时选择")
|
||
|
||
# 获取拥有集合
|
||
owned_ids = _parse_owned_outfits(user_row.owned_outfit_ids)
|
||
|
||
def check_item(item_id: int, expect_category: str):
|
||
item = (
|
||
db.query(OutfitItem)
|
||
.filter(
|
||
OutfitItem.id == item_id,
|
||
OutfitItem.category == expect_category,
|
||
OutfitItem.status == "1",
|
||
)
|
||
.first()
|
||
)
|
||
if not item:
|
||
raise HTTPException(status_code=404, detail="服饰不存在或已下架")
|
||
if item.gender not in ("unisex", lover.gender):
|
||
raise HTTPException(status_code=400, detail="服饰与恋人性别不匹配")
|
||
if not bool(item.is_free) and int(item.id) not in owned_ids:
|
||
raise HTTPException(status_code=400, detail="该服饰未拥有")
|
||
return item
|
||
|
||
top_item = check_item(top_id, "top") if top_id else None
|
||
bottom_item = check_item(bottom_id, "bottom") if bottom_id else None
|
||
dress_item = check_item(dress_id, "dress") if dress_id else None
|
||
|
||
# 保存前的形象信息（成功后若需要再写入形象栏）
|
||
prev_image = lover.image_url
|
||
prev_top = lover.outfit_top_id
|
||
prev_bottom = lover.outfit_bottom_id
|
||
prev_dress = lover.outfit_dress_id
|
||
slot_limit = int(user_row.outfit_slots or 0) or 0
|
||
if payload.save_to_look:
|
||
if slot_limit <= 0:
|
||
raise HTTPException(status_code=400, detail="形象栏位已满")
|
||
existing_count = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.with_for_update()
|
||
.count()
|
||
)
|
||
if existing_count >= slot_limit:
|
||
raise HTTPException(status_code=400, detail="形象栏位已满")
|
||
|
||
# 如需生成，调用试衣模型
|
||
result_url = payload.result_image_url
|
||
if not result_url:
|
||
person_image = _cdnize(_clean_url(lover.image_url))
|
||
if not person_image:
|
||
raise HTTPException(status_code=400, detail="缺少模特图")
|
||
top_url = _cdnize(_clean_url(top_item.image_url)) if top_item else None
|
||
bottom_url = _cdnize(_clean_url(bottom_item.image_url)) if bottom_item else None
|
||
# 连衣裙走 top_garment_url
|
||
if dress_item:
|
||
top_url = _cdnize(_clean_url(dress_item.image_url))
|
||
bottom_url = None
|
||
result_url = _call_aitryon(person_image, top_url, bottom_url)
|
||
|
||
# 扣减换装次数
|
||
user_row.clothes_num = max(0, clothes_num - 1)
|
||
|
||
# 更新当前穿戴
|
||
lover.outfit_top_id = top_item.id if top_item else None
|
||
lover.outfit_bottom_id = bottom_item.id if bottom_item else None
|
||
lover.outfit_dress_id = dress_item.id if dress_item else None
|
||
# 下载试衣结果并转存 OSS，避免临时 URL 失效
|
||
try:
|
||
img_resp = requests.get(result_url, timeout=15)
|
||
if img_resp.status_code != 200:
|
||
raise HTTPException(status_code=502, detail="生成图片下载失败")
|
||
object_name = f"lover/{lover.id}/images/{int(time.time())}_outfit.jpg"
|
||
oss_url = _upload_to_oss(img_resp.content, object_name)
|
||
result_url = oss_url
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail="生成图片保存失败") from exc
|
||
|
||
lover.image_url = _cdnize(result_url)
|
||
|
||
db.add(user_row)
|
||
db.add(lover)
|
||
# 成功后按需保存先前形象到形象栏
|
||
if payload.save_to_look:
|
||
db.add(
|
||
OutfitLook(
|
||
user_id=user.id,
|
||
lover_id=lover.id,
|
||
name=payload.look_name or "当前形象",
|
||
image_url=_cdnize(prev_image) or "",
|
||
top_item_id=prev_top,
|
||
bottom_item_id=prev_bottom,
|
||
dress_item_id=prev_dress,
|
||
createtime=datetime.utcnow(),
|
||
updatetime=datetime.utcnow(),
|
||
)
|
||
)
|
||
|
||
db.flush()
|
||
|
||
looks = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.order_by(OutfitLook.id.desc())
|
||
.all()
|
||
)
|
||
|
||
return success_response(
|
||
OutfitChangeOut(
|
||
image_url=lover.image_url,
|
||
current_outfit={
|
||
"top_id": lover.outfit_top_id,
|
||
"bottom_id": lover.outfit_bottom_id,
|
||
"dress_id": lover.outfit_dress_id,
|
||
},
|
||
clothes_num=user_row.clothes_num,
|
||
owned_outfit_ids=_serialize_owned_outfits(owned_ids),
|
||
looks=[
|
||
OutfitLookOut(
|
||
id=lk.id,
|
||
image_url=_cdnize(lk.image_url),
|
||
)
|
||
for lk in looks
|
||
],
|
||
)
|
||
)
|
||
|
||
|
||
@router.delete("/looks/{look_id}", response_model=ApiResponse[OutfitLooksResponse])
|
||
def delete_look(
|
||
look_id: int,
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
删除形象栏记录（软删），返回删除后的形象栏列表。
|
||
"""
|
||
look = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.id == look_id, OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.first()
|
||
)
|
||
if not look:
|
||
raise HTTPException(status_code=404, detail="形象不存在")
|
||
|
||
look.deleted_at = datetime.utcnow()
|
||
db.add(look)
|
||
db.flush()
|
||
|
||
looks = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.order_by(OutfitLook.id.desc())
|
||
.all()
|
||
)
|
||
return success_response(
|
||
OutfitLooksResponse(
|
||
looks=[
|
||
OutfitLookOut(
|
||
id=lk.id,
|
||
image_url=_cdnize(lk.image_url),
|
||
)
|
||
for lk in looks
|
||
]
|
||
)
|
||
)
|
||
|
||
|
||
@router.post("/looks/{look_id}/use", response_model=ApiResponse[OutfitUseResponse])
|
||
@router.post("/looks/use/{look_id}", response_model=ApiResponse[OutfitUseResponse])
|
||
def use_look(
|
||
look_id: int,
|
||
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="恋人未找到")
|
||
|
||
# 记录当前恋人形象，用于互换
|
||
prev_image = lover.image_url
|
||
prev_top = lover.outfit_top_id
|
||
prev_bottom = lover.outfit_bottom_id
|
||
prev_dress = lover.outfit_dress_id
|
||
|
||
look = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.id == look_id, OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.first()
|
||
)
|
||
if not look:
|
||
raise HTTPException(status_code=404, detail="形象不存在")
|
||
if not look.image_url:
|
||
raise HTTPException(status_code=400, detail="形象缺少图片")
|
||
|
||
lover.outfit_top_id = look.top_item_id
|
||
lover.outfit_bottom_id = look.bottom_item_id
|
||
lover.outfit_dress_id = look.dress_item_id
|
||
lover.image_url = _cdnize(look.image_url)
|
||
|
||
# 将原恋人形象写回该形象栏记录，实现互换
|
||
look.top_item_id = prev_top
|
||
look.bottom_item_id = prev_bottom
|
||
look.dress_item_id = prev_dress
|
||
look.image_url = _cdnize(prev_image) if prev_image else look.image_url
|
||
|
||
db.add(lover)
|
||
db.add(look)
|
||
db.flush()
|
||
|
||
looks = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.order_by(OutfitLook.id.desc())
|
||
.all()
|
||
)
|
||
return success_response(
|
||
OutfitUseResponse(
|
||
image_url=lover.image_url,
|
||
current_outfit={
|
||
"top_id": lover.outfit_top_id,
|
||
"bottom_id": lover.outfit_bottom_id,
|
||
"dress_id": lover.outfit_dress_id,
|
||
},
|
||
looks=[
|
||
OutfitLookOut(
|
||
id=lk.id,
|
||
image_url=_cdnize(lk.image_url),
|
||
)
|
||
for lk in looks
|
||
],
|
||
)
|
||
)
|
||
|
||
@router.get("/list", response_model=ApiResponse[OutfitListResponse])
|
||
def list_outfits(
|
||
page: int = Query(1, ge=1),
|
||
size: int = Query(7, ge=1),
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
根据恋人性别 + unisex 返回三类服饰列表（上装/下装/连衣裙），分页（默认每页 7），并回传当前穿戴的服饰 ID 与形象栏信息。
|
||
"""
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未找到")
|
||
|
||
gender_filter = [lover.gender, "unisex"]
|
||
|
||
# 用户已拥有的服饰集合：仅使用 nf_user.owned_outfit_ids（逗号或 JSON）
|
||
owned_ids: set[int] = set()
|
||
user_row = db.query(User).filter(User.id == user.id).first()
|
||
balance_val = 0.0
|
||
clothes_num_val = 0
|
||
outfit_slots_val = 0
|
||
if user_row:
|
||
balance_val = float(_ensure_balance(user_row))
|
||
owned_ids = _parse_owned_outfits(user_row.owned_outfit_ids)
|
||
clothes_num_val = int(user_row.clothes_num or 0)
|
||
outfit_slots_val = int(user_row.outfit_slots or 0)
|
||
|
||
def fetch_category(cat: str):
|
||
q = (
|
||
db.query(OutfitItem)
|
||
.filter(
|
||
OutfitItem.category == cat,
|
||
OutfitItem.gender.in_(gender_filter),
|
||
OutfitItem.status == "1",
|
||
)
|
||
.order_by(OutfitItem.weigh.desc(), OutfitItem.id.asc())
|
||
)
|
||
total_count = q.count()
|
||
records = q.offset((page - 1) * size).limit(size).all()
|
||
return total_count, records
|
||
|
||
looks = (
|
||
db.query(OutfitLook)
|
||
.filter(OutfitLook.user_id == user.id, OutfitLook.deleted_at.is_(None))
|
||
.order_by(OutfitLook.id.desc())
|
||
.all()
|
||
)
|
||
|
||
top_total, top_items = fetch_category("top")
|
||
bottom_total, bottom_items = fetch_category("bottom")
|
||
dress_total, dress_items = fetch_category("dress")
|
||
|
||
return success_response(
|
||
OutfitListResponse(
|
||
top=[
|
||
OutfitItemOut(
|
||
id=item.id,
|
||
name=item.name,
|
||
category=item.category,
|
||
gender=item.gender,
|
||
image_url=_cdnize(item.image_url),
|
||
is_free=bool(item.is_free),
|
||
price_gold=item.price_gold or 0,
|
||
is_vip_only=bool(item.is_vip_only),
|
||
is_owned=item.id in owned_ids,
|
||
)
|
||
for item in top_items
|
||
],
|
||
bottom=[
|
||
OutfitItemOut(
|
||
id=item.id,
|
||
name=item.name,
|
||
category=item.category,
|
||
gender=item.gender,
|
||
image_url=_cdnize(item.image_url),
|
||
is_free=bool(item.is_free),
|
||
price_gold=item.price_gold or 0,
|
||
is_vip_only=bool(item.is_vip_only),
|
||
is_owned=item.id in owned_ids,
|
||
)
|
||
for item in bottom_items
|
||
],
|
||
dress=[
|
||
OutfitItemOut(
|
||
id=item.id,
|
||
name=item.name,
|
||
category=item.category,
|
||
gender=item.gender,
|
||
image_url=_cdnize(item.image_url),
|
||
is_free=bool(item.is_free),
|
||
price_gold=item.price_gold or 0,
|
||
is_vip_only=bool(item.is_vip_only),
|
||
is_owned=item.id in owned_ids,
|
||
)
|
||
for item in dress_items
|
||
],
|
||
top_total=top_total,
|
||
bottom_total=bottom_total,
|
||
dress_total=dress_total,
|
||
page=page,
|
||
size=size,
|
||
balance=balance_val,
|
||
clothes_num=clothes_num_val,
|
||
outfit_slots=outfit_slots_val,
|
||
owned_outfit_ids=sorted(list(owned_ids)),
|
||
current_outfit={
|
||
"top_id": lover.outfit_top_id,
|
||
"bottom_id": lover.outfit_bottom_id,
|
||
"dress_id": lover.outfit_dress_id,
|
||
},
|
||
looks=[
|
||
OutfitLookOut(
|
||
id=lk.id,
|
||
image_url=_cdnize(lk.image_url),
|
||
)
|
||
for lk in looks
|
||
],
|
||
)
|
||
)
|
||
|
||
|
||
@router.get("/mall", response_model=ApiResponse[OutfitMallResponse])
|
||
def list_paid_outfits(
|
||
db: Session = Depends(get_db),
|
||
user: AuthedUser = Depends(get_current_user),
|
||
):
|
||
"""
|
||
金币商场：返回当前恋人性别（含 unisex）的所有付费服饰、已拥有列表和金币余额。
|
||
"""
|
||
lover = db.query(Lover).filter(Lover.user_id == user.id).first()
|
||
if not lover:
|
||
raise HTTPException(status_code=404, detail="恋人未找到")
|
||
|
||
user_row = db.query(User).filter(User.id == user.id).first()
|
||
if not user_row:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
balance = float(_ensure_balance(user_row))
|
||
owned_ids = _parse_owned_outfits(user_row.owned_outfit_ids)
|
||
|
||
gender_filter = [lover.gender, "unisex"]
|
||
items = (
|
||
db.query(OutfitItem)
|
||
.filter(
|
||
OutfitItem.status == "1",
|
||
OutfitItem.is_free.is_(False),
|
||
OutfitItem.price_gold > 0,
|
||
OutfitItem.gender.in_(gender_filter),
|
||
)
|
||
.order_by(OutfitItem.weigh.desc(), OutfitItem.id.asc())
|
||
.all()
|
||
)
|
||
|
||
return success_response(
|
||
OutfitMallResponse(
|
||
items=[
|
||
OutfitMallItem(
|
||
id=item.id,
|
||
name=item.name,
|
||
image_url=_cdnize(item.image_url),
|
||
price_gold=item.price_gold or 0,
|
||
gender=item.gender,
|
||
category=item.category,
|
||
)
|
||
for item in items
|
||
],
|
||
owned_outfit_ids=sorted(list(owned_ids)),
|
||
balance=balance,
|
||
)
|
||
)
|