beige/backend/app/services/fission_images.py

"""
app/services/fission_images.py — 第11环裂变 逐套生图（从 fission_pipeline 拆出）

复用单篇 generate_storyboard_images（codeproxy edits 带产品参考图），结果存
FissionNote.images_json。拆分原因：fission_pipeline.py 超 200 行红线上限。
🔴 生图引擎零改动；串行跑各套（套内 storyboard 已 asyncio.gather 并发）避免打爆中转站。
"""
import logging

from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)


def generate_fission_images(
    db: Session, clients, ft, product: dict, image_count: int, note_ids: list[int],
) -> None:
    """逐套生图，结果存 FissionNote.images_json。

    每套用该套自己的 note_json（含定制 imagePlan）当 note 喂生图引擎。
    产品参考图沿用源产品同一张，与单篇一致。
    """
    import asyncio
    import json as _json
    import os

    from app.models.fission import FissionNote
    from app.core.config import get_settings
    from app.services.ai_engine.image_gen import generate_storyboard_images
    from app.services.ai_engine.image_postprocessor import process_image
    from app.workers.pipeline_io import _resolve_image_path

    reference_images = None
    _img_path = _resolve_image_path(product.get("image_path", ""))
    if _img_path and os.path.isfile(_img_path):
        try:
            with open(_img_path, "rb") as f:
                reference_images = [f.read()]
        except Exception as e:  # noqa: BLE001
            logger.warning("裂变参考图读取失败，无参考图模式: %s", e)

    # R5多图：裂变同样按场景分组，生图按分镜 role 选对应图
    images_by_scene: dict[str, list[bytes]] = {}
    for _im in (product.get("images") or []):
        _p = _resolve_image_path(_im.get("path", ""))
        if _p and os.path.isfile(_p):
            try:
                with open(_p, "rb") as _f:
                    images_by_scene.setdefault(_im.get("scene") or "primary", []).append(_f.read())
            except Exception:  # noqa: BLE001
                pass
    if reference_images and not images_by_scene.get("primary"):
        images_by_scene.setdefault("primary", []).extend(reference_images)

    upload_base = get_settings().UPLOAD_ABS_ROOT
    for nid in note_ids:
        fn = db.query(FissionNote).filter(FissionNote.id == nid).first()
        if not fn:
            continue
        try:
            note = _json.loads(fn.note_json) if fn.note_json else {}
        except (ValueError, TypeError):
            note = {}
        try:
            results = asyncio.run(generate_storyboard_images(
                client=clients, note=note, product=product,
                image_count=image_count, reference_images=reference_images,
                images_by_scene=images_by_scene or None,
            ))
        except Exception as exc:  # noqa: BLE001
            logger.error("裂变套 seq=%s 生图失败: %s", fn.seq, exc)
            fn.status = "failed"; db.commit()
            continue

        img_dir = os.path.join(upload_base, str(ft.workspace_id), f"fission_{ft.id}", str(fn.seq))
        os.makedirs(img_dir, exist_ok=True)
        images: list[dict] = []
        for i, r in enumerate(results):
            if r.get("error") or not r.get("image_bytes"):
                images.append({"role": r.get("role", ""), "error": str(r.get("error", "生图失败"))[:64]})
                continue
            try:
                processed = process_image(r["image_bytes"], aspect_ratio="3:4", resample_strength=1)
            except Exception:  # noqa: BLE001
                processed = r["image_bytes"]
            fname = f"{i + 1:02d}_{r['role']}.jpg"
            with open(os.path.join(img_dir, fname), "wb") as f:
                f.write(processed)
            images.append({
                "role": r["role"], "seq": i + 1,
                "url": f"/uploads/{ft.workspace_id}/fission_{ft.id}/{fn.seq}/{fname}",
            })
        fn.images_json = _json.dumps(images, ensure_ascii=False)
        fn.status = "image_done"
        db.commit()