auto-save 2026-05-17 16:37 (~2)

api/main.py

@@ -4202,6 +4202,10 @@ class GenerateProductAngleAssetReq(BaseModel):
     note: str = ""
 
 
+class AnalyzeProductViewsReq(BaseModel):
+    refs: list[dict] = Field(default_factory=list)
+
+
 @app.get("/product-library/skg", response_model=list[ProductLibraryItem])
 def list_skg_product_library() -> list[ProductLibraryItem]:
     """内置 SKG 白底产品图库。来源是本地筛选后的产品图 manifest。"""
@@ -4260,6 +4264,90 @@ async def upload_storyboard_asset(job_id: str, file: UploadFile = File(...)) ->
     }
 
 
+PRODUCT_VIEW_VALUES = ["front", "left_45", "right_45", "side_thickness", "inner_contacts", "back_bottom"]
+PRODUCT_VIEW_LABELS = {
+    "front": "正面",
+    "left_45": "左 45",
+    "right_45": "右 45",
+    "side_thickness": "侧面厚度",
+    "inner_contacts": "内侧触点",
+    "back_bottom": "背面/底部",
+}
+
+
+def fallback_product_view(index: int) -> dict:
+    view = PRODUCT_VIEW_VALUES[min(index, len(PRODUCT_VIEW_VALUES) - 1)]
+    return {
+        "view": view,
+        "note": f"{PRODUCT_VIEW_LABELS.get(view, view)}参考；模型识别不可用时按上传顺序自动标注，请人工只检查备注。",
+        "confidence": 0.25,
+    }
+
+
+def analyze_product_view(ref_path: Path, index: int) -> dict:
+    if not LLM_API_KEY:
+        return fallback_product_view(index)
+    img_b64 = base64.b64encode(ref_path.read_bytes()).decode("ascii")
+    prompt = (
+        "You are inspecting a clean white-background product reference image for a SKG neck-and-shoulder wearable massage device. "
+        "Classify the camera/view angle into exactly one enum: front, left_45, right_45, side_thickness, inner_contacts, back_bottom. "
+        "Also write a concise Chinese note for video generation, focused on visible structure, asymmetry, thickness, inner massage contacts, buttons, opening width, and shoulder-neck wearing scale. "
+        "If uncertain, choose the closest useful view; do not ask the user. "
+        "Output strict JSON only: {\"view\":\"front|left_45|right_45|side_thickness|inner_contacts|back_bottom\", \"note\":\"中文备注\", \"confidence\":0.0}."
+    )
+    try:
+        resp = llm().chat.completions.create(
+            model=VISION_MODEL,
+            messages=[{"role": "user", "content": [
+                {"type": "text", "text": prompt},
+                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}},
+            ]}],
+            response_format={"type": "json_object"},
+            temperature=0.1,
+            max_tokens=700,
+        )
+        raw = (resp.choices[0].message.content or "").strip()
+        if not raw:
+            raw = (getattr(resp.choices[0].message, "reasoning_content", "") or "").strip()
+        match = re.search(r"\{[\s\S]*\}", raw)
+        data = json.loads(match.group(0) if match else raw)
+        view = str(data.get("view") or "").strip()
+        if view not in PRODUCT_VIEW_VALUES:
+            return fallback_product_view(index)
+        note = str(data.get("note") or "").strip() or f"{PRODUCT_VIEW_LABELS.get(view, view)}参考"
+        try:
+            confidence = max(0.0, min(1.0, float(data.get("confidence", 0.5))))
+        except Exception:
+            confidence = 0.5
+        return {"view": view, "note": note, "confidence": confidence}
+    except Exception as e:
+        fallback = fallback_product_view(index)
+        fallback["note"] = f"{fallback['note']} 识别失败：{str(e)[:80]}"
+        return fallback
+
+
+@app.post("/jobs/{job_id}/assets/product-views/analyze")
+def analyze_product_views(job_id: str, req: AnalyzeProductViewsReq) -> dict:
+    if job_id not in JOBS:
+        raise HTTPException(404, "job not found")
+    items = []
+    for index, ref in enumerate(req.refs[:6]):
+        ref_path = storyboard_ref_path(job_id, ref)
+        if not ref_path or not ref_path.exists():
+            result = fallback_product_view(index)
+        else:
+            result = analyze_product_view(ref_path, index)
+        items.append({
+            "index": index,
+            "view": result["view"],
+            "note": result["note"],
+            "confidence": result["confidence"],
+        })
+    used = {item["view"] for item in items}
+    missing = [view for view in PRODUCT_VIEW_VALUES if view not in used]
+    return {"items": items, "missing_views": missing}
+
+
 @app.post("/jobs/{job_id}/assets/product-angle")
 def generate_product_angle_asset(job_id: str, req: GenerateProductAngleAssetReq) -> dict:
     if job_id not in JOBS: