Validate final template output artifacts

app/templates/executor.py

@@ -0,0 +1,504 @@
+from __future__ import annotations
+
+import os
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass
+from pathlib import Path
+from types import MappingProxyType
+from typing import Literal
+
+from app.core.exceptions import (
+    MediaAPIError,
+    TemplateExecutionError,
+    TemplateValidationError,
+)
+from app.core.logger import get_logger
+from app.core.response import SuccessResponse
+from app.models.media import InputMedia, MediaSource, OperationResult, ResolvedRequest
+from app.operations.compress import compress_video, normalize_audio, normalize_video
+from app.operations.concat import (
+    concat_audio,
+    concat_video,
+    image_sequence,
+    image_slideshow,
+    image_to_video,
+)
+from app.operations.convert import convert_audio, convert_image, convert_video
+from app.operations.crop import crop_image, crop_video
+from app.operations.extract_audio import (
+    extract_audio,
+    fade_audio,
+    mute_video,
+    noise_reduction,
+    remove_audio,
+    remove_silence,
+    replace_audio,
+    set_volume,
+)
+from app.operations.merge import merge_audio, merge_video
+from app.operations.mix_audio import duck_background_music
+from app.operations.resize import pad_video, resize_image, resize_video, scale_video
+from app.operations.rotate import (
+    change_bitrate,
+    change_fps,
+    change_speed,
+    reverse_video,
+    rotate_video,
+    slow_motion,
+)
+from app.operations.subtitles import burn_subtitles, soft_subtitles
+from app.operations.thumbnails import (
+    blur_video,
+    denoise_video,
+    extract_frames,
+    generate_gif,
+    sharpen_video,
+    thumbnail,
+)
+from app.operations.trim import trim_audio, trim_video
+from app.operations.watermark import (
+    overlay_image,
+    overlay_video,
+    watermark_image,
+    watermark_video,
+)
+from app.services.media_service import MediaProcessor, Operation
+from app.templates.models import PreparedTemplate, ResolvedPipelineStep
+from app.templates.registry import TemplateRegistry
+
+logger = get_logger(__name__)
+
+OperationKind = Literal["ffmpeg", "whisper", "passthrough"]
+InputMode = Literal["current", "all"]
+
+
+@dataclass(frozen=True, slots=True)
+class OperationBinding:
+    """Binding from a safe YAML operation name to shared application behavior."""
+
+    kind: OperationKind
+    handler: Operation | None = None
+    input_mode: InputMode = "current"
+    whisper_task: Literal["transcribe", "translate"] = "transcribe"
+    aliases: tuple[tuple[str, str], ...] = ()
+
+
+def _ffmpeg(
+    handler: Operation,
+    *,
+    input_mode: InputMode = "current",
+    aliases: tuple[tuple[str, str], ...] = (),
+) -> OperationBinding:
+    return OperationBinding(kind="ffmpeg", handler=handler, input_mode=input_mode, aliases=aliases)
+
+
+OPERATION_BINDINGS: Mapping[str, OperationBinding] = MappingProxyType(
+    {
+        "compress": _ffmpeg(compress_video),
+        "compress_video": _ffmpeg(compress_video),
+        "resize": _ffmpeg(resize_video),
+        "resize_video": _ffmpeg(resize_video),
+        "crop": _ffmpeg(crop_video),
+        "crop_video": _ffmpeg(crop_video),
+        "trim": _ffmpeg(trim_video),
+        "trim_video": _ffmpeg(trim_video),
+        "rotate": _ffmpeg(rotate_video),
+        "reverse": _ffmpeg(reverse_video),
+        "merge": _ffmpeg(merge_video, input_mode="all"),
+        "merge_videos": _ffmpeg(merge_video, input_mode="all"),
+        "concat": _ffmpeg(concat_video, input_mode="all"),
+        "concat_videos": _ffmpeg(concat_video, input_mode="all"),
+        "convert": _ffmpeg(convert_video),
+        "convert_video": _ffmpeg(convert_video),
+        "overlay": _ffmpeg(overlay_video, input_mode="all"),
+        "overlay_video": _ffmpeg(overlay_video, input_mode="all"),
+        "watermark": _ffmpeg(watermark_video, input_mode="all"),
+        "watermark_video": _ffmpeg(watermark_video, input_mode="all"),
+        "extract_frames": _ffmpeg(extract_frames),
+        "generate_gif": _ffmpeg(generate_gif),
+        "thumbnail": _ffmpeg(thumbnail),
+        "replace_audio": _ffmpeg(replace_audio, input_mode="all"),
+        "remove_audio": _ffmpeg(remove_audio),
+        "mute": _ffmpeg(mute_video),
+        "speed": _ffmpeg(change_speed),
+        "slow_motion": _ffmpeg(slow_motion),
+        "fps": _ffmpeg(change_fps, aliases=(("value", "fps"),)),
+        "bitrate": _ffmpeg(change_bitrate, aliases=(("value", "bitrate"),)),
+        "burn_subtitles": _ffmpeg(burn_subtitles, input_mode="all"),
+        "soft_subtitles": _ffmpeg(soft_subtitles, input_mode="all"),
+        "scale": _ffmpeg(scale_video),
+        "pad": _ffmpeg(pad_video),
+        "blur": _ffmpeg(blur_video),
+        "sharpen": _ffmpeg(sharpen_video),
+        "denoise": _ffmpeg(denoise_video),
+        "normalize_video": _ffmpeg(normalize_video),
+        "extract_audio": _ffmpeg(extract_audio),
+        "convert_audio": _ffmpeg(convert_audio),
+        "normalize_audio": _ffmpeg(normalize_audio),
+        "trim_audio": _ffmpeg(trim_audio),
+        "merge_audio": _ffmpeg(merge_audio, input_mode="all"),
+        "duck_audio": _ffmpeg(duck_background_music, input_mode="all"),
+        "concat_audio": _ffmpeg(concat_audio, input_mode="all"),
+        "fade_audio": _ffmpeg(fade_audio),
+        "volume": _ffmpeg(set_volume),
+        "remove_silence": _ffmpeg(remove_silence),
+        "noise_reduction": _ffmpeg(noise_reduction),
+        "resize_image": _ffmpeg(resize_image),
+        "crop_image": _ffmpeg(crop_image),
+        "convert_image": _ffmpeg(convert_image),
+        "slideshow": _ffmpeg(image_slideshow, input_mode="all"),
+        "image_sequence": _ffmpeg(image_sequence, input_mode="all"),
+        "image_to_video": _ffmpeg(image_to_video),
+        "watermark_image": _ffmpeg(watermark_image, input_mode="all"),
+        "overlay_image": _ffmpeg(overlay_image, input_mode="all"),
+        "transcribe": OperationBinding(kind="whisper", whisper_task="transcribe"),
+        "translate": OperationBinding(kind="whisper", whisper_task="translate"),
+        "download": OperationBinding(kind="passthrough"),
+    }
+)
+
+
+class OperationExecutor:
+    """Executes allow-listed YAML operations through existing implementations."""
+
+    def __init__(self, processor: MediaProcessor) -> None:
+        self.processor = processor
+
+    @property
+    def supported_operations(self) -> set[str]:
+        """Return operation names that are safe for template YAML."""
+        return set(OPERATION_BINDINGS)
+
+    def input_mode(self, operation: str) -> InputMode:
+        """Return the default input selection mode for an operation."""
+        return self._binding(operation).input_mode
+
+    async def execute(
+        self,
+        operation: str,
+        inputs: list[InputMedia],
+        parameters: dict[str, object],
+        output_dir: Path,
+    ) -> OperationResult:
+        """Execute one pipeline step without publishing its intermediate output."""
+        binding = self._binding(operation)
+        normalized_parameters = dict(parameters)
+        for source, destination in binding.aliases:
+            if source in normalized_parameters and destination not in normalized_parameters:
+                normalized_parameters[destination] = normalized_parameters[source]
+        if binding.kind == "passthrough":
+            if not inputs:
+                raise TemplateExecutionError("Download step requires an input")
+            media = inputs[0]
+            return OperationResult(
+                path=media.temp_path,
+                filename=media.filename,
+                mime_type=media.mime_type,
+                metadata={"operation": "download", **media.metadata},
+            )
+        if binding.kind == "whisper":
+            normalized_parameters["task"] = binding.whisper_task
+            return await self.processor.transcribe_result(inputs, normalized_parameters, output_dir)
+        if binding.handler is None:  # pragma: no cover - guarded by static bindings
+            raise TemplateExecutionError("Template operation has no implementation")
+        return await binding.handler(
+            self.processor.ffmpeg, inputs, normalized_parameters, output_dir
+        )
+
+    @staticmethod
+    def _binding(operation: str) -> OperationBinding:
+        binding = OPERATION_BINDINGS.get(operation)
+        if binding is None:
+            raise TemplateValidationError(f"Unsupported template operation '{operation}'")
+        return binding
+
+
+class TemplateExecutor:
+    """Runs validated template pipelines over normalized InputMedia instances."""
+
+    def __init__(
+        self,
+        registry: TemplateRegistry,
+        operation_executor: OperationExecutor,
+        processor: MediaProcessor,
+    ) -> None:
+        self.registry = registry
+        self.operation_executor = operation_executor
+        self.processor = processor
+
+    async def execute_request(self, resolved: ResolvedRequest) -> SuccessResponse:
+        """Execute template controls parsed by the shared InputResolver."""
+        reference = resolved.params.get("template")
+        if not isinstance(reference, str) or not reference.strip():
+            raise TemplateValidationError("A non-empty 'template' reference is required")
+        parameters = resolved.params.get("parameters", {})
+        if not isinstance(parameters, dict):
+            raise TemplateValidationError("Template 'parameters' must be an object")
+        return await self.execute(resolved, reference, parameters)
+
+    async def execute(
+        self,
+        resolved: ResolvedRequest,
+        template_reference: str,
+        parameters: dict[str, object] | None = None,
+    ) -> SuccessResponse:
+        """Execute a versioned template and publish only its final artifact."""
+        started = time.monotonic()
+        cpu_started = time.process_time()
+        prepared: PreparedTemplate | None = None
+        operations: list[str] = []
+        try:
+            prepared = self.registry.prepare(template_reference, parameters)
+            workspace = await self.processor.cleanup.create_workspace(resolved.request_id)
+            initial_metadata = await self.processor.probe_inputs(resolved.inputs)
+            originals = list(resolved.inputs)
+            current = originals[0]
+            artifacts: dict[str, InputMedia] = {}
+            for index, step in enumerate(prepared.pipeline):
+                if not step.enabled:
+                    continue
+                selected = self._select_inputs(step, current, originals, artifacts)
+                step_dir = workspace.outputs / f"{index + 1:03d}_{step.operation}"
+                result = await self.operation_executor.execute(
+                    step.operation, selected, step.parameters, step_dir
+                )
+                current = self._result_media(result)
+                if self._probeable(current):
+                    await self.processor.probe_inputs([current])
+                if step.save_as:
+                    artifacts[step.save_as] = current
+                operations.append(step.operation)
+            if not operations:
+                raise TemplateExecutionError("All template pipeline operations were disabled")
+            self._validate_output(prepared, current)
+            output_validation = self._output_validation(current)
+            result = OperationResult(
+                path=current.temp_path,
+                filename=prepared.output.filename or current.filename,
+                mime_type=current.mime_type,
+                metadata={**current.metadata, "output_validation": output_validation},
+            )
+            response = await self.processor.finish_result(
+                resolved,
+                f"template.{prepared.definition.id}",
+                result,
+                started,
+                initial_metadata,
+                extra_metadata={
+                    "template": {
+                        "id": prepared.definition.id,
+                        "version": prepared.definition.version,
+                        "reference": (f"{prepared.definition.id}@{prepared.definition.version}"),
+                        "category": prepared.definition.category,
+                    },
+                    "parameters": prepared.parameters,
+                    "operations": operations,
+                },
+            )
+            logger.info(
+                "template execution completed",
+                extra=self._log_data(
+                    prepared,
+                    resolved,
+                    operations,
+                    started,
+                    cpu_started,
+                    response.metadata.get("output_size", 0),
+                    template_reference,
+                    parameters,
+                ),
+            )
+            return response
+        except MediaAPIError as exc:
+            logger.warning(
+                "template execution failed",
+                extra={
+                    **self._log_data(
+                        prepared,
+                        resolved,
+                        operations,
+                        started,
+                        cpu_started,
+                        0,
+                        template_reference,
+                        parameters,
+                    ),
+                    "error_code": exc.code,
+                    "error": exc.message,
+                },
+            )
+            raise
+        except Exception:
+            logger.exception(
+                "unexpected template execution error",
+                extra=self._log_data(
+                    prepared,
+                    resolved,
+                    operations,
+                    started,
+                    cpu_started,
+                    0,
+                    template_reference,
+                    parameters,
+                ),
+            )
+            raise
+        finally:
+            try:
+                await self.processor.cleanup.complete(resolved.request_id)
+            except Exception:
+                logger.exception(
+                    "template workspace completion failed",
+                    extra={"request_id": resolved.request_id},
+                )
+
+    def _select_inputs(
+        self,
+        step: ResolvedPipelineStep,
+        current: InputMedia,
+        originals: list[InputMedia],
+        artifacts: dict[str, InputMedia],
+    ) -> list[InputMedia]:
+        if step.inputs is None:
+            if self.operation_executor.input_mode(step.operation) == "all":
+                return [current, *originals[1:]]
+            return [current]
+        selected: list[InputMedia] = []
+        for selector in step.inputs:
+            if selector == "current":
+                selected.append(current)
+            elif selector == "original":
+                selected.append(originals[0])
+            elif selector == "originals":
+                selected.extend(originals)
+            elif selector.startswith("original:"):
+                index = int(selector.split(":", 1)[1])
+                try:
+                    selected.append(originals[index])
+                except IndexError as exc:
+                    raise TemplateExecutionError(
+                        f"Template requires original input index {index}"
+                    ) from exc
+            else:
+                name = selector.split(":", 1)[1]
+                try:
+                    selected.append(artifacts[name])
+                except KeyError as exc:  # pragma: no cover - definition validation guards order
+                    raise TemplateExecutionError(
+                        f"Template artifact '{name}' is unavailable"
+                    ) from exc
+        if not selected:
+            raise TemplateExecutionError("Template operation selected no inputs")
+        return selected
+
+    def _result_media(self, result: OperationResult) -> InputMedia:
+        if result.path is None or not result.path.is_file():
+            raise TemplateExecutionError("A template operation did not produce a file artifact")
+        filename = result.filename or result.path.name
+        mime_type = result.mime_type or self.processor.validator.infer_mime(result.path)
+        return InputMedia(
+            source=MediaSource.LOCAL_PATH,
+            filename=filename,
+            mime_type=mime_type,
+            temp_path=result.path,
+            size=result.path.stat().st_size,
+            metadata=result.metadata,
+        )
+
+    @staticmethod
+    def _probeable(media: InputMedia) -> bool:
+        return media.mime_type.startswith(("video/", "audio/", "image/"))
+
+    @staticmethod
+    def _validate_output(prepared: PreparedTemplate, media: InputMedia) -> None:
+        if not media.temp_path.is_file():
+            raise TemplateExecutionError("Template output artifact is missing")
+        try:
+            size = media.temp_path.stat().st_size
+        except OSError as exc:
+            raise TemplateExecutionError("Unable to inspect template output artifact") from exc
+        if size <= 0:
+            raise TemplateExecutionError("Template output artifact is empty")
+        if prepared.output.filename:
+            filename = prepared.output.filename
+            if Path(filename).name != filename or len(filename) > 255:
+                raise TemplateValidationError("Template output filename must be a safe basename")
+        expected = prepared.output.format.lower().lstrip(".")
+        if expected == "source":
+            return
+        actual = media.temp_path.suffix.lower().lstrip(".")
+        aliases = {"jpeg": "jpg", "m4a": "m4a"}
+        if aliases.get(actual, actual) != aliases.get(expected, expected):
+            raise TemplateExecutionError(
+                "Template output did not match its declared format",
+                details={"expected": expected, "actual": actual},
+            )
+
+    @staticmethod
+    def _output_validation(media: InputMedia) -> dict[str, object]:
+        """Summarize checks performed on the final artifact for automation clients."""
+        metadata = media.metadata
+        size = media.temp_path.stat().st_size
+        video_streams = metadata.get("video_streams")
+        audio_streams = metadata.get("audio_streams")
+        if media.mime_type.startswith("video/") and video_streams is not None and not video_streams:
+            raise TemplateExecutionError(
+                "Template output is declared as video but contains no video stream"
+            )
+        if media.mime_type.startswith("audio/") and audio_streams is not None and not audio_streams:
+            raise TemplateExecutionError(
+                "Template output is declared as audio but contains no audio stream"
+            )
+        duration = metadata.get("duration")
+        if media.mime_type.startswith(("video/", "audio/")) and duration is not None:
+            try:
+                valid_duration = float(duration) > 0
+            except (TypeError, ValueError):
+                valid_duration = False
+            if not valid_duration:
+                raise TemplateExecutionError("Template output has an invalid media duration")
+        return {
+            "valid": True,
+            "size_bytes": size,
+            "mime_type": media.mime_type,
+            "container": metadata.get("container"),
+            "duration_seconds": duration,
+            "video_stream_count": len(video_streams) if isinstance(video_streams, list) else None,
+            "audio_stream_count": len(audio_streams) if isinstance(audio_streams, list) else None,
+            "resolution": metadata.get("resolution"),
+        }
+
+    @staticmethod
+    def _log_data(
+        prepared: PreparedTemplate | None,
+        resolved: ResolvedRequest,
+        operations: list[str],
+        started: float,
+        cpu_started: float,
+        output_size: object,
+        template_reference: str,
+        supplied_parameters: dict[str, object] | None,
+    ) -> dict[str, object]:
+        memory: int | None = None
+        cpu_percent: float | None = None
+        try:
+            import psutil
+
+            memory = psutil.Process(os.getpid()).memory_info().rss
+            cpu_percent = psutil.cpu_percent(interval=None)
+        except ImportError:
+            pass
+        return {
+            "template_id": prepared.definition.id if prepared else template_reference,
+            "template_version": prepared.definition.version if prepared else None,
+            "template_reference": template_reference,
+            "parameters": prepared.parameters if prepared else supplied_parameters or {},
+            "request_id": resolved.request_id,
+            "operations": operations,
+            "duration": round(time.monotonic() - started, 4),
+            "cpu_time": round(time.process_time() - cpu_started, 6),
+            "cpu_percent": cpu_percent,
+            "memory_bytes": memory,
+            "output_size": output_size,
+        }