diff --git a/pipeline_defs/animated-explainer.yaml b/pipeline_defs/animated-explainer.yaml index fb6c9f84..0e067ab1 100644 --- a/pipeline_defs/animated-explainer.yaml +++ b/pipeline_defs/animated-explainer.yaml @@ -257,7 +257,7 @@ stages: - proposal_packet produces: - publish_log - tools_available: [] + tools_available: [export_bundle] checkpoint_required: true human_approval_default: true review_focus: diff --git a/skills/pipelines/explainer/publish-director.md b/skills/pipelines/explainer/publish-director.md index aaa2dcd6..16a12081 100644 --- a/skills/pipelines/explainer/publish-director.md +++ b/skills/pipelines/explainer/publish-director.md @@ -85,41 +85,53 @@ Each chapter maps to a script section's `start_seconds`. ### Step 5: Package Export -Create the export directory structure: +Use the `export_bundle` tool (capability `publish`) to do the packaging +deterministically — pass it the final `video_path` (from `render_report`), the +`title`, and the metadata you prepared (`description`, `tags`, `hashtags`, +`chapters`, optional `subtitles_path` and `thumbnail_path`/`thumbnail_concept`). +It lays out the export directory, writes the metadata files, and returns a +schema-valid `publish_log` (`status: "exported"`) in `data["publish_log"]` that +you persist as the stage artifact. + +It produces this structure: ``` exports/ / video/ - output.mp4 # Final rendered video + output.mp4 # Final rendered video (subtitles.srt alongside if provided) metadata/ metadata.json # All SEO metadata chapters.txt # Chapter markers - description.txt # Ready-to-paste description + description.txt # Ready-to-paste description (+ chapters) tags.txt # One tag per line thumbnails/ - concept.json # Thumbnail concept (or generated image) + concept.json # Thumbnail concept (or the copied thumbnail image) ``` +`export_bundle` is a local, offline packager — it does not upload. A networked +publisher (e.g. a YouTube uploader) would be a separate `publish`-capability +provider. + ### Step 6: Build Publish Log +`export_bundle` already returns a schema-valid `publish_log` in `data["publish_log"]` — persist that directly rather than hand-building one. Do **not** add extra entry fields (the schema sets `additionalProperties: false`; only `platform`, `status`, `url`, `video_id`, `visibility`, `export_path`, `timestamp`, `metadata_used`, `error` are allowed). The shape it returns: + ```json { "version": "1.0", "entries": [ { "platform": "youtube", - "status": "draft", - "timestamp": "2024-01-15T10:30:00Z", - "metadata": { + "status": "exported", + "export_path": "projects/vector-db-explainer/exports", + "timestamp": "2026-01-15T10:30:00+00:00", + "metadata_used": { "title": "Vector Databases Explained in 60 Seconds", - "description_length": 450, - "tags_count": 8, - "chapters_count": 6, - "thumbnail_ready": false - }, - "export_path": "exports/vector-db-explainer/", - "video_path": "renders/output.mp4" + "description": "What vector databases are and when to use them.", + "hashtags": ["#ai", "#vectordb"], + "chapters": [{ "start_seconds": 0, "title": "Introduction" }] + } } ] } diff --git a/tests/tools/test_export_bundle.py b/tests/tools/test_export_bundle.py new file mode 100644 index 00000000..74e30c49 --- /dev/null +++ b/tests/tools/test_export_bundle.py @@ -0,0 +1,155 @@ +"""Tests for the export_bundle publisher tool. + +Covers the tool contract, registry discovery, the export bundle layout, a +schema-valid publish_log, chapter formatting, and the missing-video error path. +""" + +import json +import sys +from pathlib import Path + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from tools.publishers.export_bundle import ExportBundle +from tools.base_tool import ToolStatus, ToolTier +from tools.tool_registry import ToolRegistry +from schemas.artifacts import validate_artifact + + +def _make_video(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"\x00\x00\x00\x18ftypmp42fakevideo") + + +def test_contract_metadata(): + tool = ExportBundle() + info = tool.get_info() + assert info["name"] == "export_bundle" + assert info["capability"] == "publish" + assert info["tier"] == ToolTier.PUBLISH.value + assert info["provider"] == "local" + assert info["resource_profile"]["network_required"] is False + assert tool.get_status() == ToolStatus.AVAILABLE + assert tool.estimate_cost({}) == 0.0 + + +def test_missing_video_errors(tmp_path): + result = ExportBundle().execute( + {"video_path": str(tmp_path / "nope.mp4"), "title": "X"} + ) + assert result.success is False + assert "not found" in (result.error or "") + + +def test_export_bundle_layout_and_publish_log(tmp_path): + video = tmp_path / "projects" / "demo" / "renders" / "final.mp4" + _make_video(video) + subs = tmp_path / "subs.srt" + subs.write_text("1\n00:00:00,000 --> 00:00:01,000\nhi\n", encoding="utf-8") + + result = ExportBundle().execute( + { + "video_path": str(video), + "title": "Vector Databases Explained in 60 Seconds", + "export_dir": str(tmp_path / "out"), + "description": "A quick explainer.", + "tags": ["vector db", "explainer"], + "hashtags": ["#ai", "#database"], + "chapters": [ + {"start_seconds": 0, "title": "Intro"}, + {"start_seconds": 75, "title": "How it works"}, + ], + "subtitles_path": str(subs), + "thumbnail_concept": {"text_overlay": "100x FASTER"}, + "platform": "youtube", + "visibility": "unlisted", + "timestamp": "2026-06-29T10:30:00+00:00", + } + ) + assert result.success is True + root = Path(result.data["export_path"]) + + # Layout + assert (root / "video" / "output.mp4").is_file() + assert (root / "video" / "subtitles.srt").is_file() + assert (root / "metadata" / "metadata.json").is_file() + assert (root / "metadata" / "description.txt").is_file() + assert (root / "metadata" / "tags.txt").is_file() + assert (root / "metadata" / "chapters.txt").is_file() + assert (root / "thumbnails" / "concept.json").is_file() + + # tags one-per-line + assert (root / "metadata" / "tags.txt").read_text().splitlines() == ["vector db", "explainer"] + # chapter formatting (75s -> 1:15) + assert "1:15 - How it works" in (root / "metadata" / "chapters.txt").read_text() + + # publish_log is schema-valid and shaped right + plog = result.data["publish_log"] + validate_artifact("publish_log", plog) + entry = plog["entries"][0] + assert entry["status"] == "exported" + assert entry["platform"] == "youtube" + assert entry["visibility"] == "unlisted" + assert entry["export_path"] == str(root) + assert entry["metadata_used"]["title"].startswith("Vector Databases") + + +def test_chapter_time_formatting_hours(tmp_path): + video = tmp_path / "p" / "renders" / "final.mp4" + _make_video(video) + result = ExportBundle().execute( + { + "video_path": str(video), + "title": "Long", + "export_dir": str(tmp_path / "out"), + "chapters": [{"time_seconds": 3725, "label": "Deep dive"}], # 1:02:05 + } + ) + assert result.success is True + txt = (Path(result.data["export_path"]) / "metadata" / "chapters.txt").read_text() + assert "1:02:05 - Deep dive" in txt + + +def test_infer_project_name(tmp_path): + video = tmp_path / "projects" / "my-cool-video" / "renders" / "final.mp4" + _make_video(video) + result = ExportBundle().execute( + {"video_path": str(video), "title": "T", "export_dir": str(tmp_path / "out")} + ) + # export still works; project name inference exercised via no-export_dir path below + assert result.success is True + + +def test_missing_optional_asset_errors(tmp_path): + video = tmp_path / "p" / "renders" / "final.mp4" + _make_video(video) + for key in ("subtitles_path", "thumbnail_path"): + result = ExportBundle().execute( + { + "video_path": str(video), + "title": "T", + "export_dir": str(tmp_path / "out"), + key: str(tmp_path / "does_not_exist.x"), + } + ) + assert result.success is False, key + assert key in (result.error or "") + + +def test_default_export_dir_inside_project_workspace(tmp_path): + # projects//renders/final.mp4 -> projects//exports (no export_dir given) + video = tmp_path / "projects" / "demo" / "renders" / "final.mp4" + _make_video(video) + result = ExportBundle().execute({"video_path": str(video), "title": "T"}) + assert result.success is True + assert Path(result.data["export_path"]) == (tmp_path / "projects" / "demo" / "exports").resolve() + + +def test_registry_discovers_export_bundle(): + reg = ToolRegistry() + reg.discover() + assert reg.get("export_bundle") is not None + assert reg.get_by_capability("publish")[0].name == "export_bundle" diff --git a/tools/publishers/export_bundle.py b/tools/publishers/export_bundle.py new file mode 100644 index 00000000..43cd21ef --- /dev/null +++ b/tools/publishers/export_bundle.py @@ -0,0 +1,307 @@ +"""Local export bundler — the first PUBLISH-tier tool. + +Every pipeline ends in a `publish` stage that produces a `publish_log` artifact, +but `tools/publishers/` shipped empty, so the mechanical packaging (copying the +render, writing metadata files, laying out the export directory, and emitting a +schema-valid `publish_log`) had to be hand-rolled by the agent each time. + +This tool does that packaging deterministically and locally — no external +account, no upload, no cost. It takes the final render path plus the SEO +metadata the publish-director skill prepares and writes a self-contained export +bundle a creator can hand to any platform, returning a validated `publish_log` +entry with `status: "exported"`. + +A networked publisher (e.g. a YouTube uploader) can be added later as a separate +`provider` under the same `publish` capability. +""" + +from __future__ import annotations + +import json +import shutil +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class ExportBundle(BaseTool): + name = "export_bundle" + version = "0.1.0" + tier = ToolTier.PUBLISH + capability = "publish" + provider = "local" + stability = ToolStability.BETA + execution_mode = ExecutionMode.SYNC + determinism = Determinism.DETERMINISTIC + runtime = ToolRuntime.LOCAL + + dependencies = [] # pure filesystem packaging + install_instructions = "No setup required — runs locally with the Python standard library." + + agent_skills = [] + + capabilities = ["package_export", "write_publish_log"] + supports = { + "local_offline": True, + "free": True, + "uploads": False, + } + best_for = [ + "packaging a finished render for hand-off to any platform", + "producing a schema-valid publish_log without an external account", + "offline / no-API-key publishing", + ] + not_good_for = [ + "uploading directly to YouTube/TikTok/etc. (no network publish)", + "generating SEO metadata or thumbnails (the publish-director prepares those)", + ] + + input_schema = { + "type": "object", + "required": ["video_path", "title"], + "properties": { + "video_path": { + "type": "string", + "description": "Path to the final rendered video (from render_report.outputs[].path).", + }, + "title": {"type": "string", "description": "Video title / SEO title."}, + "project_name": { + "type": "string", + "description": "Project name; used for the export folder. Defaults to the video's parent-of-parent dir name.", + }, + "export_dir": { + "type": "string", + "description": "Override the export root. Defaults to 'exports/'.", + }, + "description": {"type": "string"}, + "tags": {"type": "array", "items": {"type": "string"}}, + "hashtags": {"type": "array", "items": {"type": "string"}}, + "chapters": { + "type": "array", + "items": { + "type": "object", + "description": "Either {start_seconds, title} or {time, label}.", + }, + }, + "subtitles_path": {"type": "string"}, + "thumbnail_path": {"type": "string"}, + "thumbnail_concept": { + "type": "object", + "description": "Thumbnail concept JSON when no rendered thumbnail exists.", + }, + "platform": { + "type": "string", + "description": "Target platform label for the publish_log entry. Defaults to 'local'.", + }, + "visibility": {"type": "string", "enum": ["public", "private", "unlisted"]}, + "timestamp": { + "type": "string", + "description": "Override the ISO-8601 timestamp (mainly for deterministic tests).", + }, + }, + } + output_schema = { + "type": "object", + "properties": { + "publish_log": {"type": "object"}, + "export_path": {"type": "string"}, + "files_written": {"type": "array", "items": {"type": "string"}}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=128, vram_mb=0, disk_mb=0, network_required=False + ) + side_effects = ["writes an export bundle directory to disk"] + user_visible_verification = [ + "Open the export folder and confirm the video, metadata, and chapters are present and correct", + ] + + # ---- Helpers ---- + + @staticmethod + def _format_chapter_time(seconds: float) -> str: + seconds = int(round(seconds)) + h, rem = divmod(seconds, 3600) + m, s = divmod(rem, 60) + if h: + return f"{h}:{m:02d}:{s:02d}" + return f"{m}:{s:02d}" + + def _chapter_lines(self, chapters: list[dict[str, Any]]) -> list[str]: + lines: list[str] = [] + for ch in chapters: + label = ch.get("title") or ch.get("label") or "" + if "start_seconds" in ch or "time_seconds" in ch: + ts = self._format_chapter_time(ch.get("start_seconds", ch.get("time_seconds", 0))) + elif "time" in ch: + ts = str(ch["time"]) + else: + ts = "0:00" + lines.append(f"{ts} - {label}".rstrip(" -")) + return lines + + # ---- Execution ---- + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + video_path = Path(inputs["video_path"]).expanduser() + if not video_path.is_file(): + return ToolResult(success=False, error=f"video_path not found: {video_path}") + + title = inputs["title"] + project_name = inputs.get("project_name") or self._infer_project_name(video_path) + + # Explicitly-provided optional assets must exist — silently dropping them + # would ship a publish package missing part of an approved deliverable. + for key in ("subtitles_path", "thumbnail_path"): + val = inputs.get(key) + if val and not Path(val).expanduser().is_file(): + return ToolResult(success=False, error=f"{key} provided but not found: {val}") + + export_root = ( + Path(inputs["export_dir"]).expanduser() + if inputs.get("export_dir") + else self._default_export_dir(video_path, project_name) + ) + + video_dir = export_root / "video" + meta_dir = export_root / "metadata" + thumb_dir = export_root / "thumbnails" + for d in (video_dir, meta_dir, thumb_dir): + d.mkdir(parents=True, exist_ok=True) + + files_written: list[str] = [] + + # Video + out_video = video_dir / f"output{video_path.suffix or '.mp4'}" + shutil.copy2(video_path, out_video) + files_written.append(str(out_video)) + + # Subtitles (optional) + subs_in = inputs.get("subtitles_path") + if subs_in: + subs_in = Path(subs_in).expanduser() + if subs_in.is_file(): + out_subs = video_dir / f"subtitles{subs_in.suffix or '.srt'}" + shutil.copy2(subs_in, out_subs) + files_written.append(str(out_subs)) + + description = inputs.get("description", "") + tags = inputs.get("tags", []) or [] + hashtags = inputs.get("hashtags", []) or [] + chapters = inputs.get("chapters", []) or [] + chapter_lines = self._chapter_lines(chapters) + + # metadata.json + metadata = { + "title": title, + "description": description, + "tags": tags, + "hashtags": hashtags, + "chapters": chapters, + } + meta_json = meta_dir / "metadata.json" + meta_json.write_text(json.dumps(metadata, indent=2), encoding="utf-8") + files_written.append(str(meta_json)) + + # description.txt (description + chapters appended, ready to paste) + desc_parts = [description] if description else [] + if chapter_lines: + desc_parts.append("\n".join(chapter_lines)) + desc_txt = meta_dir / "description.txt" + desc_txt.write_text("\n\n".join(desc_parts) + ("\n" if desc_parts else ""), encoding="utf-8") + files_written.append(str(desc_txt)) + + # tags.txt (one per line) + if tags: + tags_txt = meta_dir / "tags.txt" + tags_txt.write_text("\n".join(tags) + "\n", encoding="utf-8") + files_written.append(str(tags_txt)) + + # chapters.txt + if chapter_lines: + chapters_txt = meta_dir / "chapters.txt" + chapters_txt.write_text("\n".join(chapter_lines) + "\n", encoding="utf-8") + files_written.append(str(chapters_txt)) + + # Thumbnail: real image if given, else concept JSON + thumb_in = inputs.get("thumbnail_path") + if thumb_in and Path(thumb_in).expanduser().is_file(): + thumb_in = Path(thumb_in).expanduser() + out_thumb = thumb_dir / f"thumbnail{thumb_in.suffix or '.png'}" + shutil.copy2(thumb_in, out_thumb) + files_written.append(str(out_thumb)) + elif inputs.get("thumbnail_concept"): + concept = thumb_dir / "concept.json" + concept.write_text(json.dumps(inputs["thumbnail_concept"], indent=2), encoding="utf-8") + files_written.append(str(concept)) + + timestamp = inputs.get("timestamp") or datetime.now(timezone.utc).isoformat() + entry: dict[str, Any] = { + "platform": inputs.get("platform", "local"), + "status": "exported", + "export_path": str(export_root), + "timestamp": timestamp, + "metadata_used": { + "title": title, + "description": description, + "hashtags": hashtags, + "chapters": chapters, + }, + } + if inputs.get("visibility"): + entry["visibility"] = inputs["visibility"] + + publish_log = {"version": "1.0", "entries": [entry]} + + # Validate against the canonical schema so a bad entry fails here, not at checkpoint. + try: + from schemas.artifacts import validate_artifact + + validate_artifact("publish_log", publish_log) + except Exception as exc: # pragma: no cover - defensive + return ToolResult(success=False, error=f"publish_log failed schema validation: {exc}") + + return ToolResult( + success=True, + data={ + "publish_log": publish_log, + "export_path": str(export_root), + "files_written": files_written, + }, + artifacts=[str(out_video)], + ) + + @staticmethod + def _default_export_dir(video_path: Path, project_name: str) -> Path: + """Keep run output inside the project workspace. + + When the render lives at ``projects//renders/...`` (the OpenMontage + convention), default the bundle to ``projects//exports/`` alongside + ``artifacts/``, ``assets/`` and ``renders/``. Otherwise fall back to a + top-level ``exports//``. + """ + resolved = video_path.resolve() + if resolved.parent.name == "renders": + return resolved.parent.parent / "exports" + return Path("exports") / project_name + + @staticmethod + def _infer_project_name(video_path: Path) -> str: + # projects//renders/final.mp4 -> ; fall back to the file stem. + parents = video_path.resolve().parents + if len(parents) >= 2: + return parents[1].name + return video_path.stem