feat: redesign creative studio entry

api/main.py

@@ -4275,6 +4275,32 @@ class TranslateReq(BaseModel):
     target: Literal["en", "zh"] = "en"
 
 
+class CreativeCopyReq(BaseModel):
+    goal: str
+    product: str = ""
+    audience: str = ""
+    platform: str = "TikTok / Reels"
+    tone: str = "direct"
+    seconds: int = 20
+    source_text: str = ""
+
+
+class CreativeCopyVariant(BaseModel):
+    title: str = ""
+    hook_zh: str = ""
+    script_zh: str = ""
+    script_en: str = ""
+    image_prompt_en: str = ""
+    video_prompt_en: str = ""
+    caption_zh: str = ""
+    hashtags: list[str] = Field(default_factory=list)
+
+
+class CreativeCopyResp(BaseModel):
+    model: str
+    variants: list[CreativeCopyVariant]
+
+
 class ScriptRewriteSegmentReq(BaseModel):
     index: int
     start: float = 0.0
@@ -4339,6 +4365,74 @@ def _ensure_english(text: str) -> str:
         return text
 
 
+def _creative_copy_fallback(req: CreativeCopyReq) -> CreativeCopyResp:
+    goal = req.goal.strip() or "展示 SKG 产品的核心卖点"
+    product = req.product.strip() or "SKG 健康科技产品"
+    seconds = max(6, min(60, int(req.seconds or 20)))
+    script_zh = (
+        f"开场 0-3 秒：直接展示{product}和使用场景，提出一个具体痛点。\n"
+        f"中段 3-{max(4, seconds - 5)} 秒：用三个连续镜头说明{goal}，画面保持产品清晰可见。\n"
+        f"结尾 {max(4, seconds - 5)}-{seconds} 秒：给出一句明确行动口播，收在产品近景。"
+    )
+    script_en = _ensure_english(script_zh)
+    image_prompt = _ensure_english(
+        f"{product}, premium health-tech product advertising image, clean lifestyle scene, clear product visibility, natural lighting, vertical composition"
+    )
+    video_prompt = _ensure_english(
+        f"{seconds}-second vertical short video ad for {product}. {goal}. Start with the product in use, show one clear benefit, keep camera motion smooth, realistic lifestyle lighting, no medical treatment claims."
+    )
+    return CreativeCopyResp(
+        model="fallback",
+        variants=[
+            CreativeCopyVariant(
+                title="快速成片版",
+                hook_zh=f"{product}，把一个日常痛点变成一个清楚的使用理由。",
+                script_zh=script_zh,
+                script_en=script_en,
+                image_prompt_en=image_prompt,
+                video_prompt_en=video_prompt,
+                caption_zh=f"{product}｜{goal}",
+                hashtags=["#SKG", "#健康科技", "#短视频广告"],
+            )
+        ],
+    )
+
+
+def _parse_creative_copy_response(raw: str, req: CreativeCopyReq) -> CreativeCopyResp:
+    text = (raw or "").strip()
+    text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.I).strip()
+    text = re.sub(r"\s*```$", "", text).strip()
+    match = re.search(r"\{[\s\S]*\}", text)
+    json_text = match.group(0) if match else text
+    try:
+        data = json.loads(json_text)
+    except Exception:
+        return _creative_copy_fallback(req)
+    raw_items = data.get("variants") if isinstance(data, dict) else None
+    if not isinstance(raw_items, list):
+        return _creative_copy_fallback(req)
+    variants: list[CreativeCopyVariant] = []
+    for item in raw_items[:3]:
+        if not isinstance(item, dict):
+            continue
+        hashtags = item.get("hashtags") or []
+        if not isinstance(hashtags, list):
+            hashtags = []
+        variants.append(CreativeCopyVariant(
+            title=str(item.get("title") or "").strip()[:80],
+            hook_zh=str(item.get("hook_zh") or "").strip()[:180],
+            script_zh=str(item.get("script_zh") or "").strip()[:900],
+            script_en=_ensure_english(str(item.get("script_en") or item.get("script_zh") or "").strip())[:1200],
+            image_prompt_en=_ensure_english(str(item.get("image_prompt_en") or "").strip())[:1200],
+            video_prompt_en=_ensure_english(str(item.get("video_prompt_en") or "").strip())[:1400],
+            caption_zh=str(item.get("caption_zh") or "").strip()[:240],
+            hashtags=[str(tag).strip()[:40] for tag in hashtags if str(tag).strip()][:8],
+        ))
+    if not variants:
+        return _creative_copy_fallback(req)
+    return CreativeCopyResp(model=REWRITE_MODEL if LLM_API_KEY else "fallback", variants=variants)
+
+
 @app.post("/translate")
 def translate_text(req: TranslateReq) -> dict:
     """单条文本翻译（给生图自定义提取元素 zh→en 用）"""
@@ -4374,6 +4468,44 @@ def translate_text(req: TranslateReq) -> dict:
         raise HTTPException(500, f"translate failed: {e}")
 
 
+@app.post("/creative/copy", response_model=CreativeCopyResp)
+def generate_creative_copy(req: CreativeCopyReq) -> CreativeCopyResp:
+    goal = req.goal.strip()
+    if not goal:
+        raise HTTPException(400, "goal required")
+    if not LLM_API_KEY:
+        return _creative_copy_fallback(req)
+    seconds = max(6, min(60, int(req.seconds or 20)))
+    prompt = (
+        "You are creating practical short-form ad material for an SKG AI creative tool. "
+        "Return strict JSON only. Create 3 distinct variants that can be pasted directly into image/video generation. "
+        "Avoid medical treatment claims; describe comfort, relaxation, daily use, visual proof, and product clarity instead. "
+        "Every variant must include title, hook_zh, script_zh, script_en, image_prompt_en, video_prompt_en, caption_zh, hashtags.\n\n"
+        f"Goal: {goal}\n"
+        f"Product: {req.product.strip() or 'SKG health-tech product'}\n"
+        f"Audience: {req.audience.strip() or 'short-form shoppers'}\n"
+        f"Platform: {req.platform.strip() or 'TikTok / Reels'}\n"
+        f"Tone: {req.tone.strip() or 'direct'}\n"
+        f"Length: {seconds}s\n"
+        f"Source/reference text:\n{req.source_text.strip()[:1500]}"
+    )
+    try:
+        resp = llm().chat.completions.create(
+            model=REWRITE_MODEL,
+            messages=[
+                {"role": "system", "content": "Return valid JSON only. No markdown. No explanation."},
+                {"role": "user", "content": prompt},
+            ],
+            response_format={"type": "json_object"},
+            temperature=0.72,
+            max_tokens=2200,
+        )
+        return _parse_creative_copy_response(resp.choices[0].message.content or "", req)
+    except Exception as e:
+        print(f"[creative copy fallback] {e}", flush=True)
+        return _creative_copy_fallback(req)
+
+
 def _fallback_script_rewrite_item(segment: ScriptRewriteSegmentReq, author_intent: str = "") -> dict:
     source = (segment.source or "").strip()
     intent = _ensure_english(author_intent or "")
@@ -4663,6 +4795,54 @@ async def create_job_from_upload(bg: BackgroundTasks, file: UploadFile = File(..
     return job
 
 
+def _write_creative_reference_frame(job_id: str, file_bytes: bytes | None = None) -> tuple[int, int]:
+    frames_dir = job_dir(job_id) / "frames"
+    frames_dir.mkdir(parents=True, exist_ok=True)
+    out = frames_dir / "000.jpg"
+    if file_bytes:
+        try:
+            with Image.open(io.BytesIO(file_bytes)) as raw:
+                im = ImageOps.exif_transpose(raw).convert("RGB")
+                im.thumbnail((1600, 1600), Image.LANCZOS)
+                width, height = im.size
+                im.save(out, "JPEG", quality=92)
+                return width, height
+        except Exception as e:
+            raise HTTPException(400, f"invalid image file: {e}")
+    im = Image.new("RGB", (1024, 1024), (246, 248, 246))
+    im.save(out, "JPEG", quality=92)
+    return im.size
+
+
+@app.post("/creative/jobs/image", response_model=Job)
+async def create_creative_image_job(file: UploadFile | None = File(default=None)) -> Job:
+    job_id = uuid.uuid4().hex[:12]
+    file_bytes: bytes | None = None
+    source_label = "blank"
+    if file and file.filename:
+        ext = Path(file.filename).suffix.lower()
+        if ext not in {".jpg", ".jpeg", ".png", ".webp"}:
+            raise HTTPException(400, f"unsupported image format: {ext}")
+        file_bytes = await file.read()
+        source_label = file.filename
+    width, height = _write_creative_reference_frame(job_id, file_bytes)
+    frame = KeyFrame(index=0, timestamp=0, url=f"/jobs/{job_id}/frames/0.jpg")
+    job = Job(
+        id=job_id,
+        url=f"creative://{source_label}",
+        status="frames_extracted",
+        progress=100,
+        message="创作任务已就绪",
+        width=width,
+        height=height,
+        duration=0,
+        frames=[frame],
+    )
+    JOBS[job_id] = job
+    save_state(job)
+    return job
+
+
 @app.post("/jobs/{job_id}/analyze", response_model=Job)
 async def trigger_analyze(
     job_id: str,