auto-save 2026-05-14 06:55 (+1, ~3)

api/main.py

@@ -23,6 +23,10 @@ load_dotenv()
 JOBS_DIR = Path(os.getenv("JOBS_DIR", "./jobs")).resolve()
 JOBS_DIR.mkdir(parents=True, exist_ok=True)
 CORS_ORIGINS = [o.strip() for o in os.getenv("CORS_ORIGINS", "http://localhost:4290").split(",") if o.strip()]
+PRODUCT_LIBRARY_DIR = Path(
+    os.getenv("PRODUCT_LIBRARY_DIR", Path(__file__).resolve().parent / "product_library" / "skg-products")
+).resolve()
+PRODUCT_LIBRARY_MANIFEST = PRODUCT_LIBRARY_DIR / "manifest.json"
 
 LLM_BASE_URL = os.getenv("LLM_BASE_URL", "").strip()
 LLM_API_KEY = os.getenv("LLM_API_KEY", "").strip()
@@ -214,6 +218,24 @@ class SubjectAsset(BaseModel):
     created_at: float = 0.0
 
 
+class ProductLibraryItem(BaseModel):
+    id: str
+    handle: str
+    title: str
+    product_type: str = ""
+    image_type: str = "gallery"
+    image_index: int = 0
+    filename: str
+    url: str
+    width: int = 0
+    height: int = 0
+    source_path: str = ""
+    white_score: float = 0.0
+    near_white_score: float = 0.0
+    has_people: bool = False
+    tags: list[str] = Field(default_factory=list)
+
+
 class KeyElement(BaseModel):
     """关键帧里识别 / 用户提取的元素 · 多次提取累积多张图，让用户挑选满意的"""
     id: str             # uuid hex 8
@@ -368,6 +390,35 @@ def storyboard_ref_path(job_id: str, ref: dict | None) -> Path | None:
     return None
 
 
+def load_product_library_items() -> list[ProductLibraryItem]:
+    if not PRODUCT_LIBRARY_MANIFEST.exists():
+        return []
+    try:
+        data = json.loads(PRODUCT_LIBRARY_MANIFEST.read_text(encoding="utf-8"))
+        return [ProductLibraryItem(**item) for item in data.get("items", [])]
+    except Exception as e:
+        raise HTTPException(500, f"product library manifest invalid: {e}")
+
+
+def find_product_library_item(product_id: str) -> ProductLibraryItem:
+    product_id = product_id.strip()
+    for item in load_product_library_items():
+        if item.id == product_id:
+            return item
+    raise HTTPException(404, "product library item not found")
+
+
+def product_library_file(item: ProductLibraryItem) -> Path:
+    p = (PRODUCT_LIBRARY_DIR / item.filename).resolve()
+    try:
+        p.relative_to(PRODUCT_LIBRARY_DIR)
+    except ValueError:
+        raise HTTPException(400, "invalid product library path")
+    if not p.exists():
+        raise HTTPException(404, "product library image missing")
+    return p
+
+
 def storyboard_ref_url(job_id: str, ref: dict | None) -> str:
     if not ref:
         return ""
@@ -2783,6 +2834,25 @@ def get_storyboard_video(job_id: str, video_id: str):
     return FileResponse(p, media_type="video/mp4")
 
 
+class CopyProductLibraryAssetReq(BaseModel):
+    product_id: str
+
+
+@app.get("/product-library/skg", response_model=list[ProductLibraryItem])
+def list_skg_product_library() -> list[ProductLibraryItem]:
+    """内置 SKG 白底产品图库。来源是本地筛选后的产品图 manifest。"""
+    return load_product_library_items()
+
+
+@app.get("/product-library/skg/images/{filename}")
+def get_skg_product_library_image(filename: str):
+    items = load_product_library_items()
+    item = next((x for x in items if Path(x.filename).name == filename), None)
+    if not item:
+        raise HTTPException(404, "product library image not found")
+    return FileResponse(product_library_file(item), media_type="image/jpeg")
+
+
 @app.post("/jobs/{job_id}/assets")
 async def upload_storyboard_asset(job_id: str, file: UploadFile = File(...)) -> dict:
     if job_id not in JOBS:
@@ -2813,6 +2883,32 @@ async def upload_storyboard_asset(job_id: str, file: UploadFile = File(...)) ->
     }
 
 
+@app.post("/jobs/{job_id}/assets/product-library")
+def copy_product_library_asset(job_id: str, req: CopyProductLibraryAssetReq) -> dict:
+    if job_id not in JOBS:
+        raise HTTPException(404, "job not found")
+    item = find_product_library_item(req.product_id)
+    src = product_library_file(item)
+    asset_id = uuid.uuid4().hex[:12]
+    out_dir = job_dir(job_id) / "assets"
+    out_dir.mkdir(parents=True, exist_ok=True)
+    out = out_dir / f"{asset_id}.jpg"
+    try:
+        img = Image.open(src).convert("RGB")
+        img.thumbnail((1600, 1600), Image.Resampling.LANCZOS)
+        img.save(out, "JPEG", quality=94)
+    except Exception as e:
+        raise HTTPException(400, f"product library copy failed: {e}")
+    label = f"产品融合 · {item.title} #{item.image_index}"
+    return {
+        "kind": "asset",
+        "frame_idx": -1,
+        "element_id": asset_id,
+        "cutout_id": asset_id,
+        "label": label,
+    }
+
+
 @app.get("/jobs/{job_id}/assets/{asset_id}.jpg")
 def get_storyboard_asset(job_id: str, asset_id: str):
     p = job_dir(job_id) / "assets" / f"{asset_id}.jpg"