Upload 52 files

models/tokenizer.py

@@ -80,7 +80,7 @@ class Tokenizer:
         Returns:
             str: A formatted string describing the monetary value.
         """
-        m = m.group()
+        m = match.group()
         currency = 'dollar' if m[0] == '$' else 'pound'
 
         # Handle whole amounts (e.g., "$10", "£20")

renderer/automation/__init__.py

@@ -0,0 +1,7 @@
+from renderer.automation.templates import (
+    automation_template_catalog,
+    list_automation_templates,
+    resolve_render_template,
+)
+
+__all__ = ["automation_template_catalog", "list_automation_templates", "resolve_render_template"]

renderer/automation/templates.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import re
+from copy import deepcopy
+from typing import Any
+
+
+PLACEHOLDER = re.compile(r"{{\s*([A-Za-z_][A-Za-z0-9_.-]*)\s*}}")
+
+AUTOMATION_TEMPLATES: dict[str, dict[str, Any]] = {
+    "vertical-captioned-clip": {
+        "preset": "tiktok_9_16_fast",
+        "output_name": "{{output_name}}",
+        "scenes": "{{scenes}}",
+        "voiceover": "{{voiceover}}",
+        "background_music": "{{background_music}}",
+        "auto_subtitles": True,
+        "audio_normalize": True,
+    },
+    "podcast-clip": {
+        "preset": "podcast_square",
+        "output_name": "{{output_name}}",
+        "scenes": "{{scenes}}",
+        "auto_subtitles": True,
+        "audio_normalize": True,
+    },
+    "product-promo": {
+        "preset": "capcut_product_launch",
+        "output_name": "{{output_name}}",
+        "scenes": "{{scenes}}",
+        "watermark": "{{watermark}}",
+        "background_music": "{{background_music}}",
+    },
+}
+
+
+def list_automation_templates() -> list[str]:
+    return sorted(AUTOMATION_TEMPLATES)
+
+
+def automation_template_catalog() -> dict[str, dict[str, Any]]:
+    return deepcopy(AUTOMATION_TEMPLATES)
+
+
+def resolve_render_template(
+    template: str | dict[str, Any],
+    variables: dict[str, Any],
+    overrides: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    if isinstance(template, str):
+        if template not in AUTOMATION_TEMPLATES:
+            raise ValueError(f"Unknown automation template: {template}")
+        source = deepcopy(AUTOMATION_TEMPLATES[template])
+    elif isinstance(template, dict):
+        source = deepcopy(template)
+    else:
+        raise ValueError("Template must be a registered name or JSON object")
+
+    rendered = _substitute(source, variables)
+    if not isinstance(rendered, dict):
+        raise ValueError("Rendered template must produce a JSON object")
+    return _deep_merge(rendered, overrides or {})
+
+
+def _substitute(value: Any, variables: dict[str, Any]) -> Any:
+    if isinstance(value, dict):
+        return {key: _substitute(item, variables) for key, item in value.items()}
+    if isinstance(value, list):
+        return [_substitute(item, variables) for item in value]
+    if not isinstance(value, str):
+        return value
+
+    exact = PLACEHOLDER.fullmatch(value)
+    if exact:
+        return deepcopy(_lookup(variables, exact.group(1)))
+
+    return PLACEHOLDER.sub(lambda match: str(_lookup(variables, match.group(1))), value)
+
+
+def _lookup(variables: dict[str, Any], key: str) -> Any:
+    value: Any = variables
+    for part in key.split("."):
+        if not isinstance(value, dict) or part not in value:
+            raise ValueError(f"Missing template variable: {key}")
+        value = value[part]
+    return value
+
+
+def _deep_merge(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
+    output = deepcopy(base)
+    for key, value in overrides.items():
+        if isinstance(value, dict) and isinstance(output.get(key), dict):
+            output[key] = _deep_merge(output[key], value)
+        else:
+            output[key] = deepcopy(value)
+    return output

renderer/core/config.py

@@ -19,6 +19,7 @@ class Settings:
     metadata_cache: Path = Path(os.getenv("METADATA_CACHE", str(DEFAULT_ROOT / "temp" / "metadata_cache.json")))
     signing_secret: str = os.getenv("AVA2LON_SIGNING_SECRET", os.getenv("BASYX_SIGNING_SECRET", "dev-secret-change-me"))
     api_key: str = os.getenv("AVA2LON_API_KEY", os.getenv("BASYX_API_KEY", ""))
+    public_base_url: str = os.getenv("PUBLIC_BASE_URL", "").rstrip("/")
     font_path: Path = Path(os.getenv("FONT_PATH", "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"))
     output_width: int = int(os.getenv("OUTPUT_WIDTH", "1080"))
     output_height: int = int(os.getenv("OUTPUT_HEIGHT", "1920"))
@@ -32,6 +33,8 @@ class Settings:
     whisper_device: str = os.getenv("WHISPER_DEVICE", "cpu")
     whisper_model_dir: Path = Path(os.getenv("WHISPER_MODEL_DIR", str(DEFAULT_ROOT / "models")))
     max_retries: int = int(os.getenv("MAX_RETRIES", "3"))
+    callback_max_retries: int = int(os.getenv("CALLBACK_MAX_RETRIES", "3"))
+    callback_retry_seconds: float = float(os.getenv("CALLBACK_RETRY_SECONDS", "1"))
     max_workers: int = int(os.getenv("MAX_RENDER_WORKERS", "1"))
     job_retention_seconds: int = int(os.getenv("JOB_RETENTION_SECONDS", str(24 * 3600)))
     crf: int = int(os.getenv("OUTPUT_CRF", "23"))

renderer/core/models.py

@@ -71,6 +71,8 @@ class AIReelsRequest:
     music_start: float = 0.0
     music_ducking: bool = True
     voice_volume: float = 1.0
+    callback_url: str | None = None
+    export_target: str | None = None
 
 
 @dataclass
@@ -112,12 +114,17 @@ class JobRecord:
     state: JobState
     created_at: float
     updated_at: float
+    job_type: str = "task"
     output_path: str | None = None
     download_token: str | None = None
     callback_url: str | None = None
     export_target: str | None = None
     export_path: str | None = None
     failure_reason: str | None = None
+    callback_attempts: int = 0
+    callback_delivered_at: float | None = None
+    callback_last_error: str | None = None
     commands: list[list[str]] = field(default_factory=list)
     logs: list[str] = field(default_factory=list)
     metrics: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict)

renderer/core/render_engine.py

@@ -73,6 +73,7 @@ class RenderEngine:
         metrics = {
             "render_time_seconds": round(time.time() - started, 3),
             "output_size_bytes": output.stat().st_size,
+            "duration_seconds": round(timeline.total_duration, 3),
             "scene_count": len(request.scenes),
             "creative_style": request.creative_style or request.metadata.get("creative_style"),
             "scene_effects": [scene.effect for scene in request.scenes if scene.effect],
@@ -177,6 +178,7 @@ class RenderEngine:
         metrics = {
             "render_time_seconds": round(time.time() - started, 3),
             "output_size_bytes": output.stat().st_size,
+            "duration_seconds": round(timeline.total_duration, 3),
             "scene_count": len(request.scenes),
             "creative_style": request.creative_style or request.metadata.get("creative_style"),
             "scene_effects": [scene.effect for scene in request.scenes if scene.effect],

renderer/jobs/manager.py

@@ -1,10 +1,14 @@
 from __future__ import annotations
 
 import concurrent.futures
+import hashlib
+import hmac
 import http.client
 import json
+import mimetypes
 import shutil
 import threading
+import time
 import urllib.request
 from urllib.parse import urlparse
 from dataclasses import asdict
@@ -30,10 +34,17 @@ class JobManager:
             lambda job_id, log: RenderEngine(self.settings, log=log).render(request, job_id),
             callback_url=request.callback_url,
             export_target=request.export_target,
+            metadata=request.metadata,
+            job_type="render",
         )
 
     def submit_ai_reels(self, request: AIReelsRequest) -> str:
-        return self._submit(lambda job_id, log: RenderEngine(self.settings, log=log).ai_reels(request, job_id))
+        return self._submit(
+            lambda job_id, log: RenderEngine(self.settings, log=log).ai_reels(request, job_id),
+            callback_url=request.callback_url,
+            export_target=request.export_target,
+            job_type="render",
+        )
 
     def submit_task(
         self,
@@ -41,8 +52,10 @@ class JobManager:
         *,
         callback_url: str | None = None,
         export_target: str | None = None,
+        metadata: dict | None = None,
+        job_type: str = "task",
     ) -> str:
-        return self._submit(handler, callback_url=callback_url, export_target=export_target)
+        return self._submit(handler, callback_url=callback_url, export_target=export_target, metadata=metadata, job_type=job_type)
 
     def submit_batch(self, requests: list[RenderRequest]) -> list[str]:
         job_ids: list[str] = []
@@ -112,11 +125,24 @@ class JobManager:
             "max_workers": self.settings.max_workers,
         }
 
+    def group(self, group_id: str) -> list[JobRecord]:
+        records: list[JobRecord] = []
+        for path in self.settings.jobs_dir.glob("*.json"):
+            try:
+                record = JobRecord(**read_json(path, {}))
+            except Exception:
+                continue
+            if record.metadata.get("group_id") == group_id:
+                records.append(record)
+        return sorted(records, key=lambda item: item.created_at)
+
     def _submit(
         self,
         handler: Callable[[str, Callable[[str], None]], object],
         callback_url: str | None = None,
         export_target: str | None = None,
+        metadata: dict | None = None,
+        job_type: str = "task",
     ) -> str:
         job_id = new_id()
         record = JobRecord(
@@ -124,9 +150,11 @@ class JobManager:
             state="PENDING",
             created_at=now(),
             updated_at=now(),
+            job_type=job_type,
             download_token=create_download_token(self.settings.signing_secret, job_id),
             callback_url=callback_url,
             export_target=export_target,
+            metadata=metadata or {},
         )
         self._save(record)
         self.executor.submit(self._run_with_retries, job_id, handler)
@@ -134,7 +162,8 @@ class JobManager:
 
     def _run_with_retries(self, job_id: str, handler: Callable[[str, Callable[[str], None]], object]) -> None:
         attempts = 0
-        while attempts < self.settings.max_retries:
+        max_attempts = max(1, self.settings.max_retries)
+        while attempts < max_attempts:
             attempts += 1
             if self.get(job_id).state == "CANCELLED":
                 self._send_callback(job_id)
@@ -149,6 +178,9 @@ class JobManager:
                 record = self.get(job_id)
                 output_path = getattr(result, "output_path", None)
                 export_path = self._export_copy(output_path, record.export_target, job_id) if output_path else None
+                result_metrics = getattr(result, "metrics", {}) or {}
+                if output_path:
+                    result_metrics = result_metrics | {"artifact": _artifact_metadata(Path(output_path))}
                 self._update(
                     job_id,
                     state="COMPLETED",
@@ -156,13 +188,13 @@ class JobManager:
                     export_path=export_path,
                     commands=getattr(result, "commands", []),
                     logs=getattr(result, "logs", []),
-                    metrics=getattr(result, "metrics", {}) | {"attempt": attempts},
+                    metrics=result_metrics | {"attempt": attempts},
                 )
                 self._send_callback(job_id)
                 return
             except Exception as exc:
                 self.append_log(job_id, f"Attempt {attempts} failed: {exc}")
-                if attempts >= self.settings.max_retries:
+                if attempts >= max_attempts:
                     self._update(job_id, state="FAILED", failure_reason=str(exc), metrics={"attempt": attempts})
                     self._send_callback(job_id)
 
@@ -211,17 +243,54 @@ class JobManager:
             return
         if not record.callback_url:
             return
-        payload = json.dumps(asdict(record), default=str).encode("utf-8")
-        request = urllib.request.Request(
-            record.callback_url,
-            data=payload,
-            headers={"Content-Type": "application/json", "User-Agent": "ava2lon-studio-callback/2.0"},
-            method="POST",
+        event = _event_name(record.job_type, record.state)
+        data = asdict(record)
+        data.update(
+            {
+                "event": event,
+                "event_id": f"{job_id}:{record.state.lower()}:{int(record.updated_at)}",
+                "occurred_at": record.updated_at,
+                "status_url": self._public_url(f"/status/{job_id}"),
+                "download_url": self._public_url(f"/download/{job_id}?token={record.download_token}"),
+            }
         )
-        try:
-            urllib.request.urlopen(request, timeout=10).read()
-        except Exception as exc:
-            self.append_log(job_id, f"Callback delivery failed: {exc}")
+        payload = json.dumps(data, default=str).encode("utf-8")
+        signature = hmac.new(self.settings.signing_secret.encode("utf-8"), payload, hashlib.sha256).hexdigest()
+        attempts = max(1, self.settings.callback_max_retries)
+        last_error: str | None = None
+        for attempt in range(1, attempts + 1):
+            request = urllib.request.Request(
+                record.callback_url,
+                data=payload,
+                headers={
+                    "Content-Type": "application/json",
+                    "User-Agent": "ava2lon-studio-callback/2.0",
+                    "X-Ava2lon-Event": event,
+                    "X-Ava2lon-Signature": f"sha256={signature}",
+                },
+                method="POST",
+            )
+            try:
+                with urllib.request.urlopen(request, timeout=10) as response:
+                    response.read()
+                    if response.status >= 400:
+                        raise RuntimeError(f"HTTP {response.status}")
+                self._update(
+                    job_id,
+                    callback_attempts=attempt,
+                    callback_delivered_at=now(),
+                    callback_last_error=None,
+                )
+                return
+            except Exception as exc:
+                last_error = str(exc)
+                self._update(job_id, callback_attempts=attempt, callback_last_error=last_error)
+                if attempt < attempts:
+                    time.sleep(max(0.0, self.settings.callback_retry_seconds) * (2 ** (attempt - 1)))
+        self.append_log(job_id, f"Callback delivery failed after {attempts} attempts: {last_error}")
+
+    def _public_url(self, path: str) -> str:
+        return f"{self.settings.public_base_url}{path}" if self.settings.public_base_url else path
 
 
 def _put_file(url: str, path: Path) -> None:
@@ -251,3 +320,20 @@ def _put_file(url: str, path: Path) -> None:
     connection.close()
     if response.status >= 400:
         raise RuntimeError(f"Export upload failed with HTTP {response.status}: {body[:500]!r}")
+
+
+def _event_name(job_type: str, state: str) -> str:
+    return f"{job_type}.{state.lower()}"
+
+
+def _artifact_metadata(path: Path) -> dict:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return {
+        "filename": path.name,
+        "size_bytes": path.stat().st_size,
+        "mime_type": mimetypes.guess_type(path.name)[0] or "application/octet-stream",
+        "sha256": digest.hexdigest(),
+    }

renderer/platform/processor.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 import math
 import mimetypes
+import re
 import shutil
 import zipfile
 from pathlib import Path
@@ -14,7 +15,7 @@ from renderer.core.models import TaskResult
 from renderer.core.utils import safe_filename, temp_workdir, write_json
 from renderer.ffmpeg.assets import AssetProbe
 from renderer.ffmpeg.command import FFmpegCommand
-from renderer.ffmpeg.runner import FFmpegRunner
+from renderer.ffmpeg.runner import CommandResult, FFmpegRunner
 from renderer.templates import get_platform_profile
 
 
@@ -66,6 +67,13 @@ TOOLKIT_TASKS = {
     "chroma_key",
     "blue_screen",
     "ai_background_removal",
+    "inspect",
+    "loudness_analyze",
+    "silence_detect",
+    "black_detect",
+    "scene_detect",
+    "contact_sheet",
+    "hls",
 }
 
 
@@ -225,6 +233,25 @@ class PlatformProcessor:
             final = self._export(target, job_id, "thumbnail.jpg")
         return self._result(final, {"task": "thumbnail", "timestamp": seek})
 
+    def inspect_media(self, media: str, job_id: str) -> TaskResult:
+        with temp_workdir(self.settings.temp_dir, f"{job_id}_inspect") as work:
+            source = self.ingest.resolve(media, Path(work) / "inputs", "media")
+            metadata = self.assets.probe(source).__dict__
+            metadata["source"] = media
+            output = self._json_artifact(job_id, "media_inspection", metadata)
+        return self._result(output, {"task": "inspect", "duration": metadata.get("duration", 0)})
+
+    def media_analysis(self, task: str, media: str, job_id: str, params: dict[str, Any]) -> TaskResult:
+        with temp_workdir(self.settings.temp_dir, f"{job_id}_{task}") as work:
+            source = self.ingest.resolve(media, Path(work) / "inputs", "media")
+            command = _analysis_command(task, source, params)
+            result = self._run(command)
+            analysis = _parse_analysis(task, result.stderr, params)
+            analysis.update({"task": task, "source": media})
+            output = self._json_artifact(job_id, task, analysis)
+        count = len(analysis.get("events", []))
+        return self._result(output, {"task": task, "event_count": count})
+
     def toolkit(self, payload: dict[str, Any], job_id: str) -> TaskResult:
         task = str(payload.get("task") or payload.get("operation") or "").strip()
         if task not in TOOLKIT_TASKS:
@@ -234,6 +261,11 @@ class PlatformProcessor:
         if task == "split":
             clips = payload.get("clips") if isinstance(payload.get("clips"), list) else payload.get("params", {}).get("clips")
             return self.clips(str(payload.get("input") or payload.get("media")), job_id, clips)
+        if task == "inspect":
+            return self.inspect_media(str(payload.get("input") or payload.get("media")), job_id)
+        if task in {"loudness_analyze", "silence_detect", "black_detect", "scene_detect"}:
+            params = payload.get("params") if isinstance(payload.get("params"), dict) else payload
+            return self.media_analysis(task, str(payload.get("input") or payload.get("media")), job_id, params)
 
         with temp_workdir(self.settings.temp_dir, f"{job_id}_{task}") as work:
             workdir = Path(work)
@@ -246,6 +278,8 @@ class PlatformProcessor:
                 output = output.with_suffix(".mp3")
             if task == "gif" and output.suffix.lower() != ".gif":
                 output = output.with_suffix(".gif")
+            if task == "contact_sheet" and output.suffix.lower() not in {".jpg", ".jpeg", ".png"}:
+                output = output.with_suffix(".jpg")
             command = self._toolkit_command(task, source, output, params, workdir)
             self._run(command)
             if task == "frames":
@@ -253,6 +287,11 @@ class PlatformProcessor:
                 with zipfile.ZipFile(final, "w", zipfile.ZIP_DEFLATED) as archive:
                     for frame in sorted(workdir.glob("frame_*.jpg")):
                         archive.write(frame, frame.name)
+            elif task == "hls":
+                final = self.settings.exports_dir / f"{job_id}_hls.zip"
+                with zipfile.ZipFile(final, "w", zipfile.ZIP_DEFLATED) as archive:
+                    for path in sorted(workdir.glob("hls_*")):
+                        archive.write(path, path.name)
             else:
                 final = self._export(output, job_id, output.name)
         return self._result(final, {"task": task})
@@ -284,6 +323,30 @@ class PlatformProcessor:
                 lines.append(f"file '{str(media).replace(chr(39), chr(39) + chr(92) + chr(39) + chr(39))}'")
             concat_file.write_text("\n".join(lines), encoding="utf-8")
             return FFmpegCommand().add("-hide_banner", "-f", "concat", "-safe", "0").input(concat_file).add("-c", "copy").overwrite().add(output).build()
+        if task == "contact_sheet":
+            columns = max(1, min(10, int(params.get("columns", 4))))
+            rows = max(1, min(10, int(params.get("rows", 4))))
+            interval = max(0.1, float(params.get("interval_seconds", 5)))
+            width = max(80, min(1920, int(params.get("thumbnail_width", 320))))
+            vf = f"fps=1/{interval},scale={width}:-1,tile={columns}x{rows}:padding=4:margin=4"
+            return cmd.add("-vf", vf, "-frames:v", 1, "-q:v", 2).overwrite().add(output).build()
+        if task == "hls":
+            segment_seconds = max(1, min(30, int(params.get("segment_seconds", 6))))
+            playlist_type = str(params.get("playlist_type", "vod")).lower()
+            if playlist_type not in {"event", "vod"}:
+                raise ValueError("HLS playlist_type must be 'event' or 'vod'")
+            playlist = workdir / "hls_playlist.m3u8"
+            segments = workdir / "hls_segment_%05d.ts"
+            return (
+                cmd.add(
+                    "-c:v", "libx264", "-preset", self.settings.preset, "-crf", int(params.get("crf", self.settings.crf)),
+                    "-c:a", "aac", "-f", "hls", "-hls_time", segment_seconds, "-hls_playlist_type", playlist_type,
+                    "-hls_segment_filename", segments,
+                )
+                .overwrite()
+                .add(playlist)
+                .build()
+            )
 
         duration = params.get("duration")
         if task in {"cut", "trim", "gif", "loop", "freeze_frame"} and duration is not None:
@@ -354,10 +417,11 @@ class PlatformProcessor:
         shutil.copy2(source, target)
         return target
 
-    def _run(self, command: list[str]) -> None:
+    def _run(self, command: list[str]) -> CommandResult:
         result = self.runner.run(command)
         if result.stderr:
             self._logs.append(result.stderr[-4000:])
+        return result
 
     def _record_command(self, command: list[str]) -> None:
         self._commands.append(command)
@@ -600,4 +664,60 @@ def _default_output_name(task: str) -> str:
         return "clip.gif"
     if task == "thumbnail":
         return "thumbnail.jpg"
+    if task == "contact_sheet":
+        return "contact_sheet.jpg"
+    if task == "hls":
+        return "stream.zip"
     return f"{task}.mp4"
+
+
+def _analysis_command(task: str, source: Path, params: dict[str, Any]) -> list[str]:
+    command = FFmpegCommand().add("-hide_banner").input(source)
+    if task == "loudness_analyze":
+        return command.add("-vn", "-af", "ebur128=framelog=verbose", "-f", "null", "-").build()
+    if task == "silence_detect":
+        noise = float(params.get("noise_db", -35))
+        duration = float(params.get("duration_seconds", 0.5))
+        return command.add("-af", f"silencedetect=noise={noise}dB:d={duration}", "-f", "null", "-").build()
+    if task == "black_detect":
+        threshold = max(0.0, min(1.0, float(params.get("pixel_threshold", 0.1))))
+        duration = max(0.1, float(params.get("duration_seconds", 0.1)))
+        return command.add("-vf", f"blackdetect=d={duration}:pic_th={threshold}", "-an", "-f", "null", "-").build()
+    if task == "scene_detect":
+        threshold = max(0.01, min(1.0, float(params.get("threshold", 0.4))))
+        return command.add("-vf", f"select='gt(scene,{threshold})',showinfo", "-an", "-f", "null", "-").build()
+    raise ValueError(f"Unsupported analysis task: {task}")
+
+
+def _parse_analysis(task: str, stderr: str, params: dict[str, Any]) -> dict[str, Any]:
+    events: list[dict[str, Any]] = []
+    if task == "loudness_analyze":
+        integrated = re.findall(r"I:\s*(-?\d+(?:\.\d+)?)\s*LUFS", stderr)
+        loudness_range = re.findall(r"LRA:\s*(-?\d+(?:\.\d+)?)\s*LU", stderr)
+        true_peak = re.findall(r"Peak:\s*(-?\d+(?:\.\d+)?)\s*dBFS", stderr)
+        return {
+            "integrated_lufs": float(integrated[-1]) if integrated else None,
+            "loudness_range_lu": float(loudness_range[-1]) if loudness_range else None,
+            "true_peak_dbfs": float(true_peak[-1]) if true_peak else None,
+            "target_lufs": float(params.get("target_lufs", -14)),
+        }
+    if task == "silence_detect":
+        starts = re.finditer(r"silence_start:\s*([0-9.]+)", stderr)
+        ends = list(re.finditer(r"silence_end:\s*([0-9.]+).*?silence_duration:\s*([0-9.]+)", stderr))
+        end_index = 0
+        for match in starts:
+            if end_index < len(ends):
+                end = ends[end_index]
+                if float(end.group(1)) >= float(match.group(1)):
+                    events.append({"start": float(match.group(1)), "end": float(end.group(1)), "duration": float(end.group(2))})
+                    end_index += 1
+            else:
+                events.append({"start": float(match.group(1))})
+        return {"events": events, "noise_db": float(params.get("noise_db", -35))}
+    if task == "black_detect":
+        for match in re.finditer(r"black_start:([0-9.]+)\s+black_end:([0-9.]+)\s+black_duration:([0-9.]+)", stderr):
+            events.append({"start": float(match.group(1)), "end": float(match.group(2)), "duration": float(match.group(3))})
+        return {"events": events, "pixel_threshold": float(params.get("pixel_threshold", 0.1))}
+    for match in re.finditer(r"pts_time:([0-9.]+)", stderr):
+        events.append({"time": float(match.group(1))})
+    return {"events": events, "threshold": float(params.get("threshold", 0.4))}

tests/test_n8n_automation.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+import time
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from threading import Thread
+
+from renderer.automation.templates import resolve_render_template
+from renderer.core.config import Settings
+from renderer.core.models import TaskResult
+from renderer.jobs.manager import JobManager
+from renderer.platform.processor import _analysis_command, _parse_analysis
+
+
+class N8NAutomationTests(unittest.TestCase):
+    def test_template_variables_preserve_json_types(self) -> None:
+        result = resolve_render_template(
+            {"scenes": "{{scenes}}", "title": "Campaign {{campaign.name}}", "enabled": True},
+            {"scenes": [{"start": 0, "duration": 4}], "campaign": {"name": "July"}},
+            {"enabled": False},
+        )
+        self.assertEqual(result["scenes"], [{"start": 0, "duration": 4}])
+        self.assertEqual(result["title"], "Campaign July")
+        self.assertFalse(result["enabled"])
+
+    def test_analysis_parsers_return_structured_events(self) -> None:
+        silence = _parse_analysis(
+            "silence_detect",
+            "silence_start: 1.25\nsilence_end: 2.75 | silence_duration: 1.50",
+            {},
+        )
+        black = _parse_analysis("black_detect", "black_start:0 black_end:1.2 black_duration:1.2", {})
+        loudness = _parse_analysis(
+            "loudness_analyze",
+            "I: -20.0 LUFS\nLRA: 4.0 LU\nPeak: -1.2 dBFS",
+            {},
+        )
+        self.assertEqual(silence["events"][0]["duration"], 1.5)
+        self.assertEqual(black["events"][0]["end"], 1.2)
+        self.assertEqual(loudness["integrated_lufs"], -20.0)
+
+    def test_hls_command_keeps_playlist_segment_prefix(self) -> None:
+        from renderer.platform.processor import PlatformProcessor
+
+        with tempfile.TemporaryDirectory() as directory:
+            settings = Settings(temp_dir=Path(directory) / "temp", exports_dir=Path(directory) / "exports")
+            processor = PlatformProcessor(settings)
+            command = processor._toolkit_command(
+                "hls", Path("input.mp4"), Path(directory) / "stream.zip", {}, Path(directory)
+            )
+        self.assertTrue(any(value.endswith("hls_segment_%05d.ts") for value in command))
+        self.assertTrue(any(value.endswith("hls_playlist.m3u8") for value in command))
+
+    def test_job_records_artifact_and_group_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            settings = Settings(
+                temp_dir=root / "temp",
+                exports_dir=root / "exports",
+                jobs_dir=root / "jobs",
+                storage_dir=root / "storage",
+                metadata_cache=root / "temp" / "metadata.json",
+                whisper_model_dir=root / "models",
+                max_workers=1,
+                max_retries=1,
+            )
+            manager = JobManager(settings)
+
+            def handler(job_id: str, log) -> TaskResult:
+                output = settings.exports_dir / f"{job_id}.json"
+                output.write_text("automation", encoding="utf-8")
+                return TaskResult(output_path=output, metrics={"task": "test"})
+
+            job_id = manager.submit_task(handler, metadata={"group_id": "group_test", "variant_name": "shorts"})
+            deadline = time.time() + 5
+            while time.time() < deadline and manager.get(job_id).state in {"PENDING", "RUNNING"}:
+                time.sleep(0.02)
+            record = manager.get(job_id)
+            manager.executor.shutdown(wait=True)
+
+        self.assertEqual(record.state, "COMPLETED")
+        self.assertEqual(record.metadata["group_id"], "group_test")
+        artifact = record.metrics["artifact"]
+        self.assertEqual(artifact["sha256"], hashlib.sha256(b"automation").hexdigest())
+
+    def test_webhook_contains_event_and_signature(self) -> None:
+        received: list[dict] = []
+
+        class Handler(BaseHTTPRequestHandler):
+            def do_POST(self) -> None:
+                body = self.rfile.read(int(self.headers["Content-Length"]))
+                received.append({"body": json.loads(body), "signature": self.headers.get("X-Ava2lon-Signature")})
+                self.send_response(204)
+                self.end_headers()
+
+            def log_message(self, format: str, *args) -> None:
+                return
+
+        server = HTTPServer(("127.0.0.1", 0), Handler)
+        thread = Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            with tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                settings = Settings(
+                    temp_dir=root / "temp",
+                    exports_dir=root / "exports",
+                    jobs_dir=root / "jobs",
+                    storage_dir=root / "storage",
+                    metadata_cache=root / "temp" / "metadata.json",
+                    whisper_model_dir=root / "models",
+                    max_workers=1,
+                    max_retries=1,
+                    callback_max_retries=1,
+                )
+                manager = JobManager(settings)
+
+                def handler(job_id: str, log) -> TaskResult:
+                    output = settings.exports_dir / f"{job_id}.txt"
+                    output.write_text("done", encoding="utf-8")
+                    return TaskResult(output_path=output)
+
+                callback = f"http://127.0.0.1:{server.server_port}/callback"
+                job_id = manager.submit_task(handler, callback_url=callback)
+                deadline = time.time() + 5
+                while time.time() < deadline and manager.get(job_id).callback_delivered_at is None:
+                    time.sleep(0.02)
+                record = manager.get(job_id)
+                manager.executor.shutdown(wait=True)
+        finally:
+            server.shutdown()
+            server.server_close()
+
+        self.assertIsNotNone(record.callback_delivered_at)
+        self.assertEqual(received[0]["body"]["event"], "task.completed")
+        self.assertTrue(received[0]["signature"].startswith("sha256="))
+
+
+if __name__ == "__main__":
+    unittest.main()