川
032c43525a
- 实现AI多视角设计图生成功能，支持6个可选设计参数配置 - 集成SiliconFlow FLUX.1与火山引擎Seedream 4.5双模型切换 - 构建专业中文转英文prompt系统，提升AI生成质量 - 前端设计预览支持多视角切换与视角指示器展示 - 增加多视角设计图片DesignImage模型关联及存储 - 后端设计服务异步调用AI接口，失败时降级生成mock图 - 新增管理员后台管理路由及完整的权限校验机制 - 实现后台模块：仪表盘、系统配置、用户/品类/设计管理 - 配置数据库系统配置表，支持动态AI配置及热更新 - 增加用户管理员标识字段，管理后台登录鉴权支持 - 更新API接口支持多视角设计参数及后台管理接口 - 优化设计删除逻辑，删除多视角相关图片文件 - 前端新增管理后台页面与路由，布局样式独立分离 - 更新环境变量增加AI模型相关Key与参数配置说明 - 引入httpx异步HTTP客户端用于AI接口调用及图片下载 - README文档完善AI多视角生图与后台管理详细功能与流程说明
310 lines
9.5 KiB
Python
310 lines
9.5 KiB
Python
"""
|
||
设计服务
|
||
处理设计相关的业务逻辑，支持 AI 多视角生图 + mock 降级
|
||
"""
|
||
import os
|
||
import logging
|
||
from typing import List, Optional, Tuple
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import desc
|
||
|
||
from ..models import Design, DesignImage, Category, SubType, Color
|
||
from ..schemas import DesignCreate
|
||
from ..config import settings
|
||
from .mock_generator import generate_mock_design
|
||
from .prompt_builder import get_views_for_category, build_prompt
|
||
from . import ai_generator
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _has_ai_key() -> bool:
|
||
"""检查是否配置了 AI API Key"""
|
||
model = settings.AI_IMAGE_MODEL
|
||
if model == "seedream-4.5":
|
||
return bool(settings.VOLCENGINE_API_KEY)
|
||
return bool(settings.SILICONFLOW_API_KEY)
|
||
|
||
|
||
async def create_design_async(db: Session, user_id: int, design_data: DesignCreate) -> Design:
|
||
"""
|
||
创建设计记录（异步版本，支持 AI 多视角生图）
|
||
|
||
流程：
|
||
1. 创建 Design 记录（status=generating）
|
||
2. 获取品类视角列表
|
||
3. 循环每个视角：构建 prompt → 调用 AI 生图 → 下载保存 → 创建 DesignImage
|
||
4. 第一张效果图 URL 存入 design.image_url（兼容旧逻辑）
|
||
5. 更新 status=completed
|
||
6. 失败时降级到 mock_generator
|
||
"""
|
||
# 获取关联信息
|
||
category = db.query(Category).filter(Category.id == design_data.category_id).first()
|
||
if not category:
|
||
raise ValueError(f"品类不存在: {design_data.category_id}")
|
||
|
||
sub_type = None
|
||
if design_data.sub_type_id:
|
||
sub_type = db.query(SubType).filter(SubType.id == design_data.sub_type_id).first()
|
||
|
||
color = None
|
||
if design_data.color_id:
|
||
color = db.query(Color).filter(Color.id == design_data.color_id).first()
|
||
|
||
# 创建设计记录
|
||
design = Design(
|
||
user_id=user_id,
|
||
category_id=design_data.category_id,
|
||
sub_type_id=design_data.sub_type_id,
|
||
color_id=design_data.color_id,
|
||
prompt=design_data.prompt,
|
||
carving_technique=design_data.carving_technique,
|
||
design_style=design_data.design_style,
|
||
motif=design_data.motif,
|
||
size_spec=design_data.size_spec,
|
||
surface_finish=design_data.surface_finish,
|
||
usage_scene=design_data.usage_scene,
|
||
status="generating"
|
||
)
|
||
db.add(design)
|
||
db.flush() # 获取 ID
|
||
|
||
# 尝试 AI 生图
|
||
if _has_ai_key():
|
||
try:
|
||
await _generate_ai_images(db, design, category, sub_type, color, design_data)
|
||
db.commit()
|
||
db.refresh(design)
|
||
return design
|
||
except Exception as e:
|
||
logger.error(f"AI 生图全部失败，降级到 mock: {e}")
|
||
db.rollback()
|
||
# 重新查询，因为 rollback 后 ORM 对象可能失效
|
||
design = db.query(Design).filter(Design.id == design.id).first()
|
||
if not design:
|
||
# rollback 导致 design 也没了，重新创建
|
||
design = Design(
|
||
user_id=user_id,
|
||
category_id=design_data.category_id,
|
||
sub_type_id=design_data.sub_type_id,
|
||
color_id=design_data.color_id,
|
||
prompt=design_data.prompt,
|
||
carving_technique=design_data.carving_technique,
|
||
design_style=design_data.design_style,
|
||
motif=design_data.motif,
|
||
size_spec=design_data.size_spec,
|
||
surface_finish=design_data.surface_finish,
|
||
usage_scene=design_data.usage_scene,
|
||
status="generating"
|
||
)
|
||
db.add(design)
|
||
db.flush()
|
||
|
||
# 降级到 mock 生成
|
||
_generate_mock_fallback(db, design, category, sub_type, color, design_data)
|
||
db.commit()
|
||
db.refresh(design)
|
||
return design
|
||
|
||
|
||
async def _generate_ai_images(
|
||
db: Session,
|
||
design: Design,
|
||
category,
|
||
sub_type,
|
||
color,
|
||
design_data: DesignCreate,
|
||
) -> None:
|
||
"""使用 AI 模型为每个视角生成图片"""
|
||
views = get_views_for_category(category.name)
|
||
model = settings.AI_IMAGE_MODEL
|
||
|
||
for idx, view_name in enumerate(views):
|
||
# 构建 prompt
|
||
prompt_text = build_prompt(
|
||
category_name=category.name,
|
||
view_name=view_name,
|
||
sub_type_name=sub_type.name if sub_type else None,
|
||
color_name=color.name if color else None,
|
||
user_prompt=design_data.prompt,
|
||
carving_technique=design_data.carving_technique,
|
||
design_style=design_data.design_style,
|
||
motif=design_data.motif,
|
||
size_spec=design_data.size_spec,
|
||
surface_finish=design_data.surface_finish,
|
||
usage_scene=design_data.usage_scene,
|
||
)
|
||
|
||
# 调用 AI 生图
|
||
remote_url = await ai_generator.generate_image(prompt_text, model)
|
||
|
||
# 下载保存到本地
|
||
save_path = os.path.join(
|
||
settings.UPLOAD_DIR, "designs", f"{design.id}_{view_name}.png"
|
||
)
|
||
local_url = await ai_generator.download_and_save(remote_url, save_path)
|
||
|
||
# 创建 DesignImage 记录
|
||
design_image = DesignImage(
|
||
design_id=design.id,
|
||
view_name=view_name,
|
||
image_url=local_url,
|
||
model_used=model,
|
||
prompt_used=prompt_text,
|
||
sort_order=idx,
|
||
)
|
||
db.add(design_image)
|
||
|
||
# 第一张图（效果图）存入 design.image_url 兼容旧逻辑
|
||
if idx == 0:
|
||
design.image_url = local_url
|
||
|
||
design.status = "completed"
|
||
|
||
|
||
def _generate_mock_fallback(
|
||
db: Session,
|
||
design: Design,
|
||
category,
|
||
sub_type,
|
||
color,
|
||
design_data: DesignCreate,
|
||
) -> None:
|
||
"""降级使用 mock 生成器"""
|
||
save_path = os.path.join(settings.UPLOAD_DIR, "designs", f"{design.id}.png")
|
||
image_url = generate_mock_design(
|
||
category_name=category.name,
|
||
sub_type_name=sub_type.name if sub_type else None,
|
||
color_name=color.name if color else None,
|
||
prompt=design_data.prompt,
|
||
save_path=save_path,
|
||
carving_technique=design_data.carving_technique,
|
||
design_style=design_data.design_style,
|
||
motif=design_data.motif,
|
||
size_spec=design_data.size_spec,
|
||
surface_finish=design_data.surface_finish,
|
||
usage_scene=design_data.usage_scene,
|
||
)
|
||
design.image_url = image_url
|
||
design.status = "completed"
|
||
logger.info(f"Mock 降级生成完成: design_id={design.id}")
|
||
|
||
|
||
def create_design(db: Session, user_id: int, design_data: DesignCreate) -> Design:
|
||
"""
|
||
同步版本创建设计（兼容旧调用，仅用 mock）
|
||
"""
|
||
category = db.query(Category).filter(Category.id == design_data.category_id).first()
|
||
if not category:
|
||
raise ValueError(f"品类不存在: {design_data.category_id}")
|
||
|
||
sub_type = None
|
||
if design_data.sub_type_id:
|
||
sub_type = db.query(SubType).filter(SubType.id == design_data.sub_type_id).first()
|
||
|
||
color = None
|
||
if design_data.color_id:
|
||
color = db.query(Color).filter(Color.id == design_data.color_id).first()
|
||
|
||
design = Design(
|
||
user_id=user_id,
|
||
category_id=design_data.category_id,
|
||
sub_type_id=design_data.sub_type_id,
|
||
color_id=design_data.color_id,
|
||
prompt=design_data.prompt,
|
||
carving_technique=design_data.carving_technique,
|
||
design_style=design_data.design_style,
|
||
motif=design_data.motif,
|
||
size_spec=design_data.size_spec,
|
||
surface_finish=design_data.surface_finish,
|
||
usage_scene=design_data.usage_scene,
|
||
status="generating"
|
||
)
|
||
db.add(design)
|
||
db.flush()
|
||
|
||
_generate_mock_fallback(db, design, category, sub_type, color, design_data)
|
||
db.commit()
|
||
db.refresh(design)
|
||
return design
|
||
|
||
|
||
def get_user_designs(
|
||
db: Session,
|
||
user_id: int,
|
||
page: int = 1,
|
||
page_size: int = 20
|
||
) -> Tuple[List[Design], int]:
|
||
"""
|
||
分页查询用户设计历史
|
||
|
||
Returns:
|
||
(设计列表, 总数)
|
||
"""
|
||
query = db.query(Design).filter(Design.user_id == user_id)
|
||
|
||
# 获取总数
|
||
total = query.count()
|
||
|
||
# 分页查询，按创建时间倒序
|
||
offset = (page - 1) * page_size
|
||
designs = query.order_by(desc(Design.created_at)).offset(offset).limit(page_size).all()
|
||
|
||
return designs, total
|
||
|
||
|
||
def get_design_by_id(db: Session, design_id: int, user_id: int) -> Optional[Design]:
|
||
"""
|
||
获取单个设计
|
||
只返回属于该用户的设计
|
||
"""
|
||
return db.query(Design).filter(
|
||
Design.id == design_id,
|
||
Design.user_id == user_id
|
||
).first()
|
||
|
||
|
||
def delete_design(db: Session, design_id: int, user_id: int) -> bool:
|
||
"""
|
||
删除设计
|
||
|
||
1. 查找设计（必须属于该用户）
|
||
2. 删除图片文件
|
||
3. 删除数据库记录
|
||
|
||
Returns:
|
||
是否删除成功
|
||
"""
|
||
design = db.query(Design).filter(
|
||
Design.id == design_id,
|
||
Design.user_id == user_id
|
||
).first()
|
||
|
||
if not design:
|
||
return False
|
||
|
||
# 删除主图片文件
|
||
if design.image_url:
|
||
file_path = design.image_url.lstrip("/")
|
||
if os.path.exists(file_path):
|
||
try:
|
||
os.remove(file_path)
|
||
except Exception:
|
||
pass
|
||
|
||
# 删除多视角图片文件
|
||
for img in design.images:
|
||
if img.image_url:
|
||
fp = img.image_url.lstrip("/")
|
||
if os.path.exists(fp):
|
||
try:
|
||
os.remove(fp)
|
||
except Exception:
|
||
pass
|
||
|
||
# 删除数据库记录
|
||
db.delete(design)
|
||
db.commit()
|
||
|
||
return True
|