HartOMat/backend/app/domains/rendering/tasks.py

"""Rendering domain tasks — Celery tasks for Blender-based rendering.

These tasks run on the `asset_pipeline` queue in the render-worker
container, which has Blender and cadquery available.

Phase A2: Initial implementation replacing the blender-renderer HTTP service.
Phase B: This module will be expanded as part of the Domain-Driven restructure.
"""
import logging
from pathlib import Path

from app.tasks.celery_app import celery_app
from app.core.task_logs import log_task_event

logger = logging.getLogger(__name__)


def _update_workflow_run_status(order_line_id: str, status: str, error: str | None = None) -> None:
    """Update the most recent WorkflowRun for an order_line after task completion."""
    try:
        import asyncio
        from datetime import datetime as _dt

        async def _run():
            from app.database import AsyncSessionLocal
            from app.domains.rendering.models import WorkflowRun
            from sqlalchemy import select as _sel
            async with AsyncSessionLocal() as db:
                res = await db.execute(
                    _sel(WorkflowRun)
                    .where(WorkflowRun.order_line_id == order_line_id)
                    .order_by(WorkflowRun.created_at.desc())
                    .limit(1)
                )
                run = res.scalar_one_or_none()
                if run and run.status == "pending":
                    run.status = status
                    run.completed_at = _dt.utcnow()
                    if error:
                        run.error_message = error[:2000]
                    await db.commit()

        asyncio.get_event_loop().run_until_complete(_run())
    except Exception as _exc:
        logger.warning("Failed to update WorkflowRun status for line %s: %s", order_line_id, _exc)


@celery_app.task(
    bind=True,
    name="app.domains.rendering.tasks.render_still_task",
    queue="asset_pipeline",
    max_retries=2,
)
def render_still_task(
    self,
    step_path: str,
    output_path: str,
    engine: str = "cycles",
    samples: int = 256,
    smooth_angle: int = 30,
    cycles_device: str = "auto",
    width: int = 512,
    height: int = 512,
    transparent_bg: bool = False,
    template_path: str | None = None,
    target_collection: str = "Product",
    material_library_path: str | None = None,
    material_map: dict | None = None,
    part_names_ordered: list | None = None,
    lighting_only: bool = False,
    shadow_catcher: bool = False,
    rotation_x: float = 0.0,
    rotation_y: float = 0.0,
    rotation_z: float = 0.0,
    noise_threshold: str = "",
    denoiser: str = "",
    denoising_input_passes: str = "",
    denoising_prefilter: str = "",
    denoising_quality: str = "",
    denoising_use_gpu: str = "",
    mesh_attributes: dict | None = None,
) -> dict:
    """Render a STEP file to a still PNG via Blender subprocess.

    Returns render metadata dict on success.
    Retries up to 2 times on failure (30s countdown).
    """
    log_task_event(self.request.id, f"Starting render_still_task: {Path(step_path).name}", "info")
    try:
        from app.services.render_blender import render_still
        result = render_still(
            step_path=Path(step_path),
            output_path=Path(output_path),
            engine=engine,
            samples=samples,
            smooth_angle=smooth_angle,
            cycles_device=cycles_device,
            width=width,
            height=height,
            transparent_bg=transparent_bg,
            template_path=template_path,
            target_collection=target_collection,
            material_library_path=material_library_path,
            material_map=material_map,
            part_names_ordered=part_names_ordered,
            lighting_only=lighting_only,
            shadow_catcher=shadow_catcher,
            rotation_x=rotation_x,
            rotation_y=rotation_y,
            rotation_z=rotation_z,
            noise_threshold=noise_threshold,
            denoiser=denoiser,
            denoising_input_passes=denoising_input_passes,
            denoising_prefilter=denoising_prefilter,
            denoising_quality=denoising_quality,
            denoising_use_gpu=denoising_use_gpu,
            mesh_attributes=mesh_attributes or {},
        )
        log_task_event(self.request.id, f"Completed successfully in {result.get('total_duration_s', 0):.1f}s", "done")
        logger.info(
            "render_still_task completed: %s → %s in %.1fs",
            Path(step_path).name, Path(output_path).name,
            result.get("total_duration_s", 0),
        )
        try:
            from app.core.websocket import publish_event_sync
            publish_event_sync(None, {
                "type": "render.still.completed",
                "step_path": Path(step_path).name,
                "output": Path(output_path).name,
            })
        except Exception:
            pass
        return result
    except Exception as exc:
        log_task_event(self.request.id, f"Failed: {exc}", "error")
        logger.error("render_still_task failed for %s: %s", step_path, exc)
        try:
            from app.core.websocket import publish_event_sync
            publish_event_sync(None, {
                "type": "render.still.failed",
                "step_path": Path(step_path).name,
                "error": str(exc),
            })
        except Exception:
            pass
        raise self.retry(exc=exc, countdown=30)


@celery_app.task(
    bind=True,
    name="app.domains.rendering.tasks.render_turntable_task",
    queue="asset_pipeline",
    max_retries=2,
)
def render_turntable_task(
    self,
    step_path: str,
    output_dir: str,
    output_name: str = "turntable",
    engine: str = "cycles",
    samples: int = 64,
    smooth_angle: int = 30,
    cycles_device: str = "auto",
    width: int = 1920,
    height: int = 1080,
    frame_count: int = 120,
    fps: int = 30,
    turntable_degrees: float = 360.0,
    turntable_axis: str = "world_z",
    bg_color: str = "",
    template_path: str | None = None,
    target_collection: str = "Product",
    material_library_path: str | None = None,
    material_map: dict | None = None,
    part_names_ordered: list | None = None,
    lighting_only: bool = False,
    shadow_catcher: bool = False,
    camera_orbit: bool = True,
    rotation_x: float = 0.0,
    rotation_y: float = 0.0,
    rotation_z: float = 0.0,
) -> dict:
    """Render a STEP file as a turntable animation (frames + FFmpeg composite).

    Returns render metadata dict on success.
    """
    log_task_event(self.request.id, f"Starting render_turntable_task: {Path(step_path).name}", "info")
    import json
    import os
    import shutil
    import subprocess
    import sys
    from app.services.render_blender import find_blender

    blender_bin = find_blender()
    if not blender_bin:
        raise RuntimeError("Blender binary not found in render-worker container")

    step = Path(step_path)
    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    scripts_dir = Path(os.environ.get("RENDER_SCRIPTS_DIR", "/render-scripts"))
    turntable_script = scripts_dir / "turntable_render.py"

    # GLB generation via OCC — deflection from admin settings (scene_linear/angular_deflection)
    from app.config import settings as app_settings
    from sqlalchemy import create_engine as _create_engine, text as _text
    from sqlalchemy.orm import Session as _Session
    _db_engine = _create_engine(app_settings.database_url_sync)
    with _Session(_db_engine) as _s:
        _rows = _s.execute(_text("SELECT key, value FROM system_settings")).fetchall()
        _sett = {r[0]: r[1] for r in _rows}
    _db_engine.dispose()
    linear_deflection = float(_sett.get("scene_linear_deflection", "0.1"))
    angular_deflection = float(_sett.get("scene_angular_deflection", "0.1"))
    glb_path = step.parent / f"{step.stem}_thumbnail.glb"
    if not glb_path.exists() or glb_path.stat().st_size == 0:
        occ_script = scripts_dir / "export_step_to_gltf.py"
        occ_cmd = [
            sys.executable, str(occ_script),
            "--step_path", str(step),
            "--output_path", str(glb_path),
            "--linear_deflection", str(linear_deflection),
            "--angular_deflection", str(angular_deflection),
        ]
        occ_result = subprocess.run(occ_cmd, capture_output=True, text=True, timeout=120)
        if occ_result.returncode != 0:
            raise RuntimeError(
                f"export_step_to_gltf.py failed:\n{occ_result.stderr[-500:]}"
            )
        logger.info("render_turntable_task: GLB generated: %s", glb_path.name)

    # Build turntable render arguments
    frames_dir = out_dir / "frames"
    frames_dir.mkdir(exist_ok=True)

    cmd = [
        blender_bin, "--background",
        "--python", str(turntable_script),
        "--",
        str(glb_path),
        str(frames_dir),
        output_name,
        str(width), str(height),
        engine, str(samples), str(smooth_angle), cycles_device,
        str(frame_count), str(fps), str(turntable_degrees), turntable_axis,
        template_path or "",
        target_collection,
        material_library_path or "",
        json.dumps(material_map) if material_map else "{}",
        json.dumps(part_names_ordered) if part_names_ordered else "[]",
        "1" if lighting_only else "0",
        "1" if shadow_catcher else "0",
        "1" if camera_orbit else "0",
        str(rotation_x), str(rotation_y), str(rotation_z),
    ]

    try:
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=3600
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"Blender turntable exited {result.returncode}:\n{result.stdout[-2000:]}"
            )
    except Exception as exc:
        log_task_event(self.request.id, f"Failed: {exc}", "error")
        logger.error("render_turntable_task failed: %s", exc)
        try:
            from app.core.websocket import publish_event_sync
            publish_event_sync(None, {
                "type": "render.turntable.failed",
                "step_path": Path(step_path).name,
                "error": str(exc),
            })
        except Exception:
            pass
        raise self.retry(exc=exc, countdown=60)

    # FFmpeg composite: frames → MP4 with optional background
    output_mp4 = out_dir / f"{output_name}.mp4"
    ffmpeg_cmd = _build_ffmpeg_cmd(
        frames_dir, output_mp4, fps=fps, bg_color=bg_color
    )
    try:
        subprocess.run(ffmpeg_cmd, check=True, capture_output=True, text=True, timeout=300)
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(f"FFmpeg composite failed: {exc.stderr[-500:]}")

    log_task_event(self.request.id, "Completed successfully", "done")
    try:
        from app.core.websocket import publish_event_sync
        publish_event_sync(None, {
            "type": "render.turntable.completed",
            "step_path": Path(step_path).name,
            "output": Path(output_mp4).name,
        })
    except Exception:
        pass
    return {
        "output_mp4": str(output_mp4),
        "frame_count": frame_count,
        "fps": fps,
    }


@celery_app.task(
    name="rendering.publish_asset",
    queue="step_processing",
)
def publish_asset(
    order_line_id: str,
    asset_type: str,
    storage_key: str,
    render_config: dict | None = None,
) -> str | None:
    """Create a MediaAsset record after a successful render."""
    import asyncio

    async def _run() -> str | None:
        from app.database import AsyncSessionLocal
        from app.domains.media.models import MediaAsset, MediaAssetType
        from app.domains.orders.models import OrderLine
        from app.domains.products.models import Product
        from sqlalchemy import select

        async with AsyncSessionLocal() as db:
            res = await db.execute(select(OrderLine).where(OrderLine.id == order_line_id))
            line = res.scalar_one_or_none()
            if not line:
                return None

            # Resolve cad_file_id from the linked product
            cad_file_id = None
            if line.product_id:
                prod_res = await db.execute(select(Product).where(Product.id == line.product_id))
                product = prod_res.scalar_one_or_none()
                if product:
                    cad_file_id = product.cad_file_id

            asset = MediaAsset(
                tenant_id=getattr(line, "tenant_id", None),
                order_line_id=line.id,
                product_id=line.product_id,
                cad_file_id=cad_file_id,
                asset_type=MediaAssetType(asset_type),
                storage_key=storage_key,
                render_config=render_config,
            )
            db.add(asset)
            await db.commit()
            return str(asset.id)

    return asyncio.get_event_loop().run_until_complete(_run())


def _resolve_step_path_for_order_line(order_line_id: str) -> tuple[str | None, str | None]:
    """Sync helper: resolves (step_path, cad_file_id) from an OrderLine via DB."""
    import asyncio

    async def _inner() -> tuple[str | None, str | None]:
        from app.database import AsyncSessionLocal
        from app.domains.orders.models import OrderLine
        from app.domains.products.models import Product
        from app.models.cad_file import CadFile
        from sqlalchemy import select
        from sqlalchemy.orm import selectinload

        async with AsyncSessionLocal() as db:
            res = await db.execute(
                select(OrderLine)
                .options(selectinload(OrderLine.product))
                .where(OrderLine.id == order_line_id)
            )
            line = res.scalar_one_or_none()
            if not line or not line.product or not line.product.cad_file_id:
                return None, None
            cad_res = await db.execute(
                select(CadFile).where(CadFile.id == line.product.cad_file_id)
            )
            cad = cad_res.scalar_one_or_none()
            if not cad or not cad.stored_path:
                return None, None
            return cad.stored_path, str(line.product.cad_file_id)

    return asyncio.get_event_loop().run_until_complete(_inner())


@celery_app.task(
    bind=True,
    name="app.domains.rendering.tasks.render_order_line_still_task",
    queue="asset_pipeline",
    max_retries=2,
)
def render_order_line_still_task(self, order_line_id: str, **params) -> dict:
    """Render a still image for an order line, resolving STEP path from DB.

    Wraps render_still_task logic but accepts order_line_id instead of step_path.
    On success, creates a MediaAsset record via publish_asset.
    """
    import asyncio
    from app.domains.rendering.job_document import RenderJobDocument, JobState
    from app.core.process_steps import StepName

    log_task_event(self.request.id, f"Starting render_order_line_still_task: order_line={order_line_id}", "info")

    # Initialise job document and store real Celery task ID
    job_doc = RenderJobDocument.new(order_line_id=order_line_id, celery_task_id=self.request.id)
    job_doc.set_state(JobState.RUNNING)

    def _save_job_doc():
        async def _run():
            from app.database import AsyncSessionLocal
            from app.domains.orders.models import OrderLine
            from sqlalchemy import update as _upd
            async with AsyncSessionLocal() as db:
                await db.execute(
                    _upd(OrderLine)
                    .where(OrderLine.id == order_line_id)
                    .values(render_job_doc=job_doc.to_dict())
                )
                await db.commit()
        try:
            asyncio.get_event_loop().run_until_complete(_run())
        except Exception as _exc:
            logger.debug("_save_job_doc failed: %s", _exc)

    _save_job_doc()

    job_doc.begin_step(StepName.RESOLVE_STEP_PATH)
    step_path_str, cad_file_id = _resolve_step_path_for_order_line(order_line_id)
    if not step_path_str:
        job_doc.fail_step(StepName.RESOLVE_STEP_PATH, "product missing or has no linked CAD file")
        job_doc.set_state(JobState.FAILED, error="Cannot resolve STEP path")
        _save_job_doc()
        log_task_event(self.request.id, f"Failed: cannot resolve STEP path for order_line {order_line_id}", "error")
        raise RuntimeError(
            f"Cannot resolve STEP path for order_line {order_line_id}: "
            "product missing or has no linked CAD file"
        )
    job_doc.finish_step(StepName.RESOLVE_STEP_PATH, output={"step_path": step_path_str})

    step = Path(step_path_str)
    output_dir = step.parent / "renders"
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / f"line_{order_line_id}.png"

    try:
        job_doc.begin_step(StepName.BLENDER_STILL)
        from app.services.render_blender import render_still
        result = render_still(
            step_path=step,
            output_path=output_path,
            **params,
        )
        job_doc.finish_step(
            StepName.BLENDER_STILL,
            output={"output_path": str(output_path), "duration_s": result.get("total_duration_s")},
        )
        job_doc.set_state(JobState.COMPLETED, result={
            "output_path": str(output_path),
            "duration_s": result.get("total_duration_s"),
            "engine_used": result.get("engine_used"),
        })
        _save_job_doc()

        publish_asset.delay(
            order_line_id,
            "still",
            str(output_path),
            render_config=result,
        )
        log_task_event(self.request.id, f"Completed successfully in {result.get('total_duration_s', 0):.1f}s", "done")
        logger.info(
            "render_order_line_still_task completed for line %s in %.1fs",
            order_line_id, result.get("total_duration_s", 0),
        )
        try:
            from app.core.websocket import publish_event_sync
            publish_event_sync(None, {
                "type": "render.order_line.completed",
                "order_line_id": order_line_id,
            })
        except Exception:
            pass
        _update_workflow_run_status(order_line_id, "completed")
        return result
    except Exception as exc:
        job_doc.fail_step(StepName.BLENDER_STILL, str(exc))
        job_doc.set_state(JobState.FAILED, error=str(exc))
        _save_job_doc()
        log_task_event(self.request.id, f"Failed: {exc}", "error")
        logger.error("render_order_line_still_task failed for %s: %s", order_line_id, exc)
        try:
            from app.core.websocket import publish_event_sync
            publish_event_sync(None, {
                "type": "render.order_line.failed",
                "order_line_id": order_line_id,
                "error": str(exc),
            })
        except Exception:
            pass
        _update_workflow_run_status(order_line_id, "failed", str(exc))
        raise self.retry(exc=exc, countdown=30)


@celery_app.task(
    bind=True,
    name="app.domains.rendering.tasks.export_gltf_for_order_line_task",
    queue="asset_pipeline",
    max_retries=1,
)
def export_gltf_for_order_line_task(self, order_line_id: str) -> dict:
    """Export a geometry GLB directly from STEP via OCC (no STL intermediary).

    Publishes a MediaAsset with asset_type='gltf_geometry'.
    """
    import os
    import subprocess
    import sys

    step_path_str, cad_file_id = _resolve_step_path_for_order_line(order_line_id)
    if not step_path_str:
        raise RuntimeError(f"Cannot resolve STEP path for order_line {order_line_id}")

    step = Path(step_path_str)
    output_path = step.parent / f"{step.stem}_geometry.glb"
    scripts_dir = Path(os.environ.get("RENDER_SCRIPTS_DIR", "/render-scripts"))
    occ_script = scripts_dir / "export_step_to_gltf.py"

    if not occ_script.exists():
        raise RuntimeError(f"export_step_to_gltf.py not found at {occ_script}")

    try:
        cmd = [
            sys.executable, str(occ_script),
            "--step_path", str(step),
            "--output_path", str(output_path),
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        if result.returncode != 0:
            raise RuntimeError(
                f"export_step_to_gltf.py exited {result.returncode}:\n{result.stderr[-500:]}"
            )
        publish_asset.delay(order_line_id, "gltf_geometry", str(output_path))
        logger.info("export_gltf_for_order_line_task completed via OCC: %s", output_path.name)
        return {"glb_path": str(output_path), "method": "occ"}
    except Exception as exc:
        logger.error("export_gltf_for_order_line_task failed for %s: %s", order_line_id, exc)
        raise self.retry(exc=exc, countdown=15)


@celery_app.task(
    bind=True,
    name="app.domains.rendering.tasks.export_blend_for_order_line_task",
    queue="asset_pipeline",
    max_retries=1,
)
def export_blend_for_order_line_task(self, order_line_id: str) -> dict:
    """Export a production .blend file via Blender + asset library (export_blend.py).

    Publishes a MediaAsset with asset_type='blend_production'.
    Requires Blender + the render-scripts directory.
    """
    import json
    import os
    import subprocess

    step_path_str, cad_file_id = _resolve_step_path_for_order_line(order_line_id)
    if not step_path_str:
        raise RuntimeError(f"Cannot resolve STEP path for order_line {order_line_id}")

    step = Path(step_path_str)
    # Use geometry GLB as input (generate if missing)
    glb_path = step.parent / f"{step.stem}_geometry.glb"
    if not glb_path.exists():
        import subprocess as _sp
        import sys as _sys
        scripts_dir_tmp = Path(os.environ.get("RENDER_SCRIPTS_DIR", "/render-scripts"))
        occ_cmd = [
            _sys.executable, str(scripts_dir_tmp / "export_step_to_gltf.py"),
            "--step_path", str(step),
            "--output_path", str(glb_path),
        ]
        occ_res = _sp.run(occ_cmd, capture_output=True, text=True, timeout=120)
        if occ_res.returncode != 0:
            raise RuntimeError(f"GLB generation failed:\n{occ_res.stderr[-500:]}")

    output_path = step.parent / f"{step.stem}_production.blend"
    scripts_dir = Path(os.environ.get("RENDER_SCRIPTS_DIR", "/render-scripts"))
    export_script = scripts_dir / "export_blend.py"

    from app.services.render_blender import find_blender
    blender_bin = find_blender()
    if not blender_bin:
        raise RuntimeError("Blender binary not found — cannot run export_blend task")

    # Resolve asset library path and material map from DB
    asset_lib_path = ""
    mat_map: dict = {}
    try:
        from sqlalchemy import create_engine, select as sql_select
        from sqlalchemy.orm import Session
        from app.config import settings as app_settings
        from app.domains.orders.models import OrderLine
        from app.domains.products.models import Product

        engine = create_engine(app_settings.database_url_sync)
        with Session(engine) as s:
            line = s.execute(sql_select(OrderLine).where(OrderLine.id == order_line_id)).scalar_one_or_none()
            if line:
                product = s.execute(sql_select(Product).where(Product.id == line.product_id)).scalar_one_or_none()
                if product:
                    mat_map = {
                        m.get("part_name", ""): m.get("material", "")
                        for m in (product.cad_part_materials or [])
                    }
    except Exception as exc:
        logger.warning("export_blend_for_order_line_task: DB resolution error (non-fatal): %s", exc)

    try:
        cmd = [
            blender_bin, "--background",
            "--python", str(export_script),
            "--",
            "--glb_path", str(glb_path),
            "--output_path", str(output_path),
            "--asset_library_blend", asset_lib_path,
            "--material_map", json.dumps(mat_map),
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
        if result.returncode != 0:
            raise RuntimeError(
                f"export_blend.py exited {result.returncode}:\n{result.stderr[-500:]}"
            )
        publish_asset.delay(order_line_id, "blend_production", str(output_path))
        logger.info("export_blend_for_order_line_task completed: %s", output_path.name)
        return {"blend_path": str(output_path)}
    except Exception as exc:
        logger.error("export_blend_for_order_line_task failed for %s: %s", order_line_id, exc)
        raise self.retry(exc=exc, countdown=30)


@celery_app.task(
    bind=True,
    name="app.domains.rendering.tasks.apply_asset_library_materials_task",
    queue="asset_pipeline",
    max_retries=1,
)
def apply_asset_library_materials_task(self, order_line_id: str, asset_library_id: str) -> dict:
    """Apply Blender asset library materials to a render via the asset_library.py script."""
    import json
    import os
    import subprocess
    from pathlib import Path
    from app.services.render_blender import find_blender

    blender_bin = find_blender()
    if not blender_bin:
        raise RuntimeError("Blender not available")

    # Resolve paths from DB
    def _inner():
        from sqlalchemy import create_engine, select as sql_select
        from sqlalchemy.orm import Session
        from app.config import settings
        from app.domains.orders.models import OrderLine
        from app.domains.products.models import CadFile, Product

        engine = create_engine(settings.database_url_sync)
        with Session(engine) as s:
            line = s.execute(sql_select(OrderLine).where(OrderLine.id == order_line_id)).scalar_one_or_none()
            if not line:
                return None, None, None
            product = s.execute(sql_select(Product).where(Product.id == line.product_id)).scalar_one_or_none()
            if not product or not product.cad_file_id:
                return None, None, None
            cad = s.execute(sql_select(CadFile).where(CadFile.id == product.cad_file_id)).scalar_one_or_none()
            glb_path = str(Path(cad.stored_path).parent / f"{Path(cad.stored_path).stem}_geometry.glb") if cad else None

            # Resolve asset library blend path
            try:
                from app.domains.materials.models import AssetLibrary
                lib = s.execute(sql_select(AssetLibrary).where(AssetLibrary.id == asset_library_id)).scalar_one_or_none()
                blend_path = lib.blend_file_path if lib else None
            except Exception:
                blend_path = None

            mat_map = {m.get("part_name", ""): m.get("material", "") for m in (product.cad_part_materials or [])}
            return glb_path, blend_path, mat_map

    result = _inner()
    if result is None or result[0] is None:
        logger.warning("apply_asset_library_materials_task: could not resolve paths for %s", order_line_id)
        return {"status": "skipped"}

    glb_path, blend_path, mat_map = result
    if not glb_path or not Path(glb_path).exists():
        logger.warning("Geometry GLB not found for %s", order_line_id)
        return {"status": "skipped", "reason": "glb_not_found"}

    scripts_dir = Path(os.environ.get("RENDER_SCRIPTS_DIR", "/render-scripts"))
    script = scripts_dir / "asset_library.py"

    cmd = [
        blender_bin, "--background", "--python", str(script), "--",
        "--glb_path", glb_path,
        "--asset_library_blend", blend_path or "",
        "--material_map", json.dumps(mat_map),
    ]

    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        if proc.returncode != 0:
            raise RuntimeError(f"asset_library.py failed: {proc.stderr[-500:]}")
    except Exception as exc:
        logger.error("apply_asset_library_materials_task failed for %s: %s", order_line_id, exc)
        raise self.retry(exc=exc, countdown=15)

    return {"status": "applied", "order_line_id": order_line_id}


def _build_ffmpeg_cmd(
    frames_dir: Path, output_mp4: Path, fps: int = 30, bg_color: str = ""
) -> list:
    """Build FFmpeg command for compositing turntable frames to MP4."""
    import shutil as _shutil
    ffmpeg = _shutil.which("ffmpeg") or "ffmpeg"
    frame_pattern = str(frames_dir / "%04d.png")

    if bg_color:
        # Overlay transparent frames onto solid color background
        r = int(bg_color[1:3], 16) if bg_color.startswith("#") else 255
        g = int(bg_color[3:5], 16) if bg_color.startswith("#") else 255
        b = int(bg_color[5:7], 16) if bg_color.startswith("#") else 255
        color_str = f"color=c=0x{r:02x}{g:02x}{b:02x}:s=1920x1080:r={fps}"
        return [
            ffmpeg, "-y",
            "-f", "lavfi", "-i", color_str,
            "-framerate", str(fps), "-i", frame_pattern,
            "-filter_complex", "[0:v][1:v]overlay=0:0",
            "-c:v", "libx264", "-pix_fmt", "yuv420p",
            "-movflags", "+faststart",
            str(output_mp4),
        ]
    else:
        return [
            ffmpeg, "-y",
            "-framerate", str(fps), "-i", frame_pattern,
            "-c:v", "libx264", "-pix_fmt", "yuv420p",
            "-movflags", "+faststart",
            str(output_mp4),
        ]