189 lines
11 KiB
Python
189 lines
11 KiB
Python
"""AI 炫图路由：gpt-image 生图代理（Key 仅存后端），提示词由 drawing_data 结构化构建。
|
||
|
||
- quick 模式: /v1/images/generations 纯提示词生成
|
||
- photo 模式: /v1/images/edits 以用户照片为参考图，把窗型合成进照片
|
||
"""
|
||
import base64
|
||
import json
|
||
|
||
import httpx
|
||
from fastapi import APIRouter
|
||
from pydantic import BaseModel, Field
|
||
|
||
from huamenchuang.core.config import settings
|
||
from huamenchuang.core.deps import CurrentUser, DbSession
|
||
from huamenchuang.core.response import fail, ok
|
||
from huamenchuang.services.ai_prompt import build_prompt, describe_window, pick_size
|
||
|
||
router = APIRouter(prefix="/ai-render", tags=["AI 炫图"])
|
||
|
||
|
||
class AiRenderBody(BaseModel):
|
||
template_id: int | None = None
|
||
drawing_data: str | dict | None = None
|
||
mode: str = Field(default="quick", pattern="^(quick|photo|blueprint)$")
|
||
style: str = Field(default="modern", pattern="^(modern|chinese|european|nordic)$")
|
||
time: str = Field(default="day", pattern="^(day|dusk|night)$")
|
||
# quick: indoor/outdoor/studio; photo: render/install/erase/indoor/villa/fence/blank/sunroom-*
|
||
scene: str = "indoor"
|
||
photo_b64: str | None = Field(default=None, description="照片模式参考图(base64, 不带 dataURL 前缀)")
|
||
rect: dict | None = Field(default=None, description="照片模式框选窗洞 {x,y,w,h}(原始照片像素坐标)")
|
||
sketch_b64: str | None = Field(default=None, description="大样图模式: 2D 图纸截图(base64, 不带 dataURL 前缀)")
|
||
|
||
|
||
async def _resolve_drawing(db, body: AiRenderBody) -> str | dict | None:
|
||
if body.drawing_data:
|
||
return body.drawing_data
|
||
if body.template_id:
|
||
from sqlalchemy import select
|
||
from huamenchuang.models.windoor_template import WindoorTemplate
|
||
t = (await db.execute(select(WindoorTemplate).where(
|
||
WindoorTemplate.id == body.template_id))).scalar_one_or_none()
|
||
return t.drawing_data if t else None
|
||
return None
|
||
|
||
|
||
@router.post("/prompt")
|
||
async def preview_prompt(body: AiRenderBody, user: CurrentUser, db: DbSession):
|
||
"""提示词预览（前端可展开查看优化后的完整提示词）。"""
|
||
drawing = await _resolve_drawing(db, body)
|
||
if drawing is None:
|
||
return fail("未找到图纸数据")
|
||
try:
|
||
info = describe_window(drawing)
|
||
prompt = (build_blueprint_prompt(drawing) if body.mode == "blueprint"
|
||
else build_prompt(drawing, body.style, body.time, body.scene))
|
||
except Exception:
|
||
return fail("图纸数据解析失败")
|
||
return ok(data={"features": info, "prompt": prompt,
|
||
"size": pick_size(info["width_mm"], info["height_mm"])})
|
||
|
||
|
||
def build_blueprint_prompt(drawing: str | dict) -> str:
|
||
"""工程大样图提示词：手绘草图/量尺登记表照片或 2D 图纸 → 黑白 CAD 大样图
|
||
（移植自 autocad 项目草图标准化资产，叠加 drawing_data 结构化特征增强分格保真）。"""
|
||
feat = ""
|
||
if drawing:
|
||
try:
|
||
info = describe_window(drawing)
|
||
feat = f"参考窗型（若输入草图中有相似窗型可参照其分格风格）: {info['summary']}; 外框尺寸 {info['width_mm']}×{info['height_mm']} mm。"
|
||
except Exception:
|
||
feat = ""
|
||
return f"""Use case: style-transfer
|
||
Asset type: 门窗工程审图参考图
|
||
Primary request: 将输入的门窗手绘草图（现场量尺登记表照片）或 2D 图纸重绘为清晰、整洁的工程 CAD 大样图风格预览。
|
||
Input images: 第一张是原始手绘草图照片或 2D 图纸，仅作为几何、分格与标注位置的几何来源。输入可能是一页包含多个窗的手绘量尺登记表——逐窗转换，保持各窗的相对布局与各自分格。
|
||
Style/medium: 黑白二维门窗工程大样图、正投影立面、细黑线框、玻璃淡灰填充、外部尺寸线与分格尺寸标注、中文工程标注、每个窗下方标注"窗1/窗2…"编号、右下角标题栏（图名/比例/日期）。
|
||
Constraints: {feat}
|
||
方向铁律(最重要): 每个窗的横竖方向必须与原图完全一致——原图中横向更宽的窗，输出必须仍是横向(宽>高)；原图中竖向更高的窗，输出必须仍是竖向(高>宽)；严禁旋转、竖立或横放任何窗。宽度尺寸一律标注在窗的正下方水平尺寸线上，高度尺寸一律标注在窗的右侧竖直尺寸线上。登记表手写表格中"宽/mm"栏的数字对应草图的水平方向，"高/mm"栏对应竖直方向。
|
||
数量铁律: 输出大样图的数量必须与输入草图中的窗数量完全相同，一个不多一个不少；每个窗只绘制一次，严禁重复绘制同一个窗，严禁合并或省略任何窗。
|
||
尺寸铁律: 手写数字可能有涂改或划掉——被划掉的数字以改后数字为准；只誊写可清晰辨认的数字；无法辨认的尺寸处只标注"?"，严禁猜测或编造任何数字。
|
||
开向铁律: 每个开启扇的对角三角符号必须与原图的铰链侧、开启方向完全一致，不得镜像或反向。
|
||
斜角铁律: 手稿中的斜边、切角、梯形轮廓必须原样保留——外边界或玻璃分格中出现斜线时，输出必须以相同方向与相近比例的斜线呈现（斜线两端点的位置与原图一致），严禁把斜角窗、梯形窗拉直或画成矩形；手稿标注了斜边尺寸的照常誊写，未标注的按原图比例绘制；圆弧边同样原样保留。
|
||
结构铁律: 严格保留每个窗的分格数量、比例与位置；不要增删或合并分格；如果照片是旋转拍摄的，先将画面转正再逐窗转换；输出仅供人工审图与报价附件，不作为 CAD 几何或生产数据来源；无水印、无装饰性场景、无彩色、不保留登记表表格框线。
|
||
Avoid: 三维渲染、透视效果、写实照片、彩色渲染、虚构品牌、模糊文字、重复的窗、被旋转方向的窗。"""
|
||
|
||
|
||
@router.post("/generate")
|
||
async def generate(body: AiRenderBody, user: CurrentUser, db: DbSession):
|
||
"""云端 AI 生图（gpt-image）：quick=生成 / photo=参考图编辑 / blueprint=手绘草图或2D图纸转工程大样图。"""
|
||
if not settings.ai_api_key:
|
||
return fail("AI 生图服务未配置(HMC_AI_API_KEY)")
|
||
drawing = await _resolve_drawing(db, body)
|
||
# blueprint 纯草图模式(上传手绘登记表, 无样板)不要求图纸数据
|
||
if drawing is None and body.mode != "blueprint":
|
||
return fail("未找到图纸数据")
|
||
if body.mode == "photo" and not body.photo_b64:
|
||
return fail("照片模式需要上传参考照片")
|
||
if body.mode == "blueprint" and not body.sketch_b64:
|
||
return fail("大样图模式需要草图照片或 2D 图纸截图")
|
||
try:
|
||
info = describe_window(drawing) if drawing else {"width_mm": 0, "height_mm": 0, "summary": ""}
|
||
# 照片模式: 把框选洞口位置转成提示词方位描述, 引导 AI 装窗位置
|
||
rect_hint = ""
|
||
if body.mode == "photo" and body.rect:
|
||
try:
|
||
r = body.rect
|
||
# 由前端提供照片宽高与洞口坐标(像素) → 相对方位词
|
||
ph = float(r.get("photo_h") or 0)
|
||
pw = float(r.get("photo_w") or 0)
|
||
if pw > 0 and ph > 0:
|
||
cx = (float(r["x"]) + float(r["w"]) / 2) / pw
|
||
cy = (float(r["y"]) + float(r["h"]) / 2) / ph
|
||
horiz = "left" if cx < 0.37 else ("right" if cx > 0.63 else "center")
|
||
vert = "upper" if cy < 0.37 else ("lower" if cy > 0.63 else "middle")
|
||
area_pct = round((float(r["w"]) * float(r["h"])) / (pw * ph) * 100)
|
||
rect_hint = (f"Install the window into the existing window opening located in the "
|
||
f"{vert}-{horiz} part of the image (opening covers about {area_pct}% of the frame), "
|
||
f"replacing whatever old window or opening is there.")
|
||
except Exception:
|
||
rect_hint = ""
|
||
if body.mode == "blueprint":
|
||
prompt = build_blueprint_prompt(drawing)
|
||
size = "1536x1024" # 大样图横版(量尺登记表整页比例), 对齐 autocad 项目口径
|
||
else:
|
||
prompt = build_prompt(drawing, body.style, body.time, body.scene, rect_hint=rect_hint)
|
||
size = pick_size(info["width_mm"], info["height_mm"])
|
||
except Exception:
|
||
return fail("图纸数据解析失败")
|
||
|
||
base = settings.ai_base_url.rstrip("/")
|
||
headers = {"Authorization": f"Bearer {settings.ai_api_key}"}
|
||
timeout = httpx.Timeout(settings.ai_timeout, connect=15)
|
||
|
||
async def _post_image(client: httpx.AsyncClient) -> httpx.Response:
|
||
if body.mode == "blueprint":
|
||
img_bytes = base64.b64decode(body.sketch_b64.split(",")[-1])
|
||
mime = "image/jpeg" if img_bytes[:2] == b"\xff\xd8" else "image/png"
|
||
ext = "jpg" if mime == "image/jpeg" else "png"
|
||
return await client.post(
|
||
f"{base}/v1/images/edits", headers=headers,
|
||
data={"model": settings.ai_model, "prompt": prompt, "size": size, "n": "1"},
|
||
files={"image[]": (f"drawing.{ext}", img_bytes, mime)})
|
||
if body.mode == "photo":
|
||
img_bytes = base64.b64decode(body.photo_b64.split(",")[-1])
|
||
mime = "image/jpeg" if img_bytes[:2] == b"\xff\xd8" else "image/png"
|
||
ext = "jpg" if mime == "image/jpeg" else "png"
|
||
return await client.post(
|
||
f"{base}/v1/images/edits", headers=headers,
|
||
data={"model": settings.ai_model, "prompt": prompt, "size": size, "n": "1"},
|
||
files={"image[]": (f"photo.{ext}", img_bytes, mime)})
|
||
return await client.post(
|
||
f"{base}/v1/images/generations", headers=headers,
|
||
json={"model": settings.ai_model, "prompt": prompt, "n": 1, "size": size})
|
||
|
||
b64 = None
|
||
last_err = ""
|
||
# 中转偶发超时/5xx 自动重试一次(生图耗时长, 失败重试显著提升成功率)
|
||
for attempt in range(2):
|
||
try:
|
||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||
resp = await _post_image(client)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
item = (data.get("data") or [{}])[0]
|
||
b64 = item.get("b64_json")
|
||
if b64:
|
||
break
|
||
last_err = f"生图服务返回异常: {str(data)[:200]}"
|
||
except httpx.TimeoutException:
|
||
last_err = "AI 生图超时，请稍后重试"
|
||
except httpx.HTTPStatusError as e:
|
||
detail = e.response.text[:200] if e.response is not None else ""
|
||
last_err = f"生图服务错误({e.response.status_code if e.response else '?'}): {detail}"
|
||
# 4xx(如鉴权/参数)重试无意义
|
||
if e.response is not None and e.response.status_code < 500:
|
||
break
|
||
except Exception as e: # noqa: BLE001
|
||
last_err = f"生图失败: {str(e)[:150]}"
|
||
if attempt == 0 and b64 is None:
|
||
stage_wait = 3 # 重试前短暂等待
|
||
import asyncio
|
||
await asyncio.sleep(stage_wait)
|
||
if not b64:
|
||
return fail(last_err or "生成失败")
|
||
return ok(data={"image_b64": b64, "prompt": prompt, "size": size,
|
||
"features": {"width_mm": info["width_mm"], "height_mm": info["height_mm"],
|
||
"summary": info["summary"]}},
|
||
message="生成成功")
|