zlei6/OpenMontage
1
0
Fork 0
mirror of https://github.com/calesthio/OpenMontage.git synced 2026-08-05 15:20:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge main into Sora provider branch

Browse Source
This commit is contained in:
calesthio
2026-07-05 07:03:26 -07:00
parent b3ff734b5a ce11f6a24f
commit 0eaef99902
165 changed files with 10350 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

386
tools/analysis/dashscope_asr.py Normal file
View File

@@ -0,0 +1,386 @@
"""DashScope (Alibaba Cloud Bailian) ASR with word-level timestamps.
Uses the DashScope-native async transcription endpoint with
X-DashScope-Async: enable header. The model qwen3-asr-flash-filetrans is the
ONLY DashScope path that returns word-level timestamps (the sync
qwen3-asr-flash via /chat/completions does not).
Pattern: submit (POST) -> poll (GET /tasks/{task_id}) -> download
transcription_url -> parse transcripts[].sentences[].words[].
This tool replaces the broken `whisperx` slot for subtitle-aligned
transcription. Word timestamps are normalized from milliseconds to seconds.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class DashscopeAsr(BaseTool):
name = "dashscope_asr"
version = "0.1.0"
tier = ToolTier.ANALYZE
capability = "analysis"
provider = "dashscope"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.ASYNC
determinism = Determinism.DETERMINISTIC
runtime = ToolRuntime.API
dependencies = []
install_instructions = (
"Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n"
" Get one at https://dashscope.aliyun.com/"
)
fallback = "transcriber"
fallback_tools = ["transcriber"]
agent_skills = ["dashscope"]
capabilities = [
"speech_to_text",
"word_timestamps",
"multilingual",
]
supports = {
"word_timestamps": True,
"multilingual": True,
"offline": False,
}
best_for = [
"word-level timestamp transcription for subtitle alignment",
"Mandarin and English speech recognition",
"replacing whisperx when word-level granularity is needed",
]
not_good_for = [
"real-time transcription",
"local/offline transcription",
]
input_schema = {
"type": "object",
"required": ["audio_url"],
"properties": {
"audio_url": {
"type": "string",
"description": (
"Publicly accessible URL of the audio file to transcribe. "
"Must be reachable by DashScope servers — local paths "
"are not supported."
),
},
"model": {
"type": "string",
"enum": ["qwen3-asr-flash-filetrans"],
"default": "qwen3-asr-flash-filetrans",
},
"language_hints": {
"type": "array",
"items": {"type": "string"},
"default": ["zh", "en"],
"description": (
"Language hints to improve accuracy. "
'Examples: ["zh", "en", "ja"].'
),
},
"enable_words": {
"type": "boolean",
"default": True,
"description": (
"Enable word-level timestamps. Required for subtitle "
"alignment."
),
},
"output_path": {"type": "string"},
"poll_interval_seconds": {
"type": "number",
"default": 5.0,
"minimum": 1.0,
},
"timeout_seconds": {
"type": "integer",
"default": 300,
"minimum": 30,
},
},
}
resource_profile = ResourceProfile(
cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=20, network_required=True
)
retry_policy = RetryPolicy(
max_retries=2,
backoff_seconds=2.0,
retryable_errors=["timeout", "rate_limit"],
)
idempotency_key_fields = ["audio_url", "model", "enable_words", "language_hints"]
side_effects = [
"writes transcription JSON to output_path",
"calls DashScope (Alibaba Cloud) ASR API (async submit + poll)",
]
user_visible_verification = [
"Check transcription text for accuracy",
"Verify word-level timestamps before building subtitles",
]
SUBMIT_URL = (
"https://dashscope.aliyuncs.com/api/v1/services/audio/asr/"
"transcription"
)
POLL_URL_TEMPLATE = (
"https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}"
)
def get_status(self) -> ToolStatus:
if os.environ.get("DASHSCOPE_API_KEY"):
return ToolStatus.AVAILABLE
return ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
# DashScope ASR pricing is per-minute; check console for actual cost.
return 0.0
def execute(self, inputs: dict[str, Any]) -> ToolResult:
api_key = os.environ.get("DASHSCOPE_API_KEY")
if not api_key:
return ToolResult(
success=False,
error="DASHSCOPE_API_KEY not set. " + self.install_instructions,
)
audio_url = inputs.get("audio_url", "").strip()
if not audio_url:
return ToolResult(
success=False, error="audio_url is required."
)
if not self._is_public_url(audio_url):
return ToolResult(
success=False,
error=(
"audio_url must be a publicly accessible URL (http/https). "
"DashScope servers fetch the file; local paths are not "
"supported. Upload the audio to a public location first."
),
)
# DashScope ASR rejects http:// URLs with InvalidParameter.MalformedURL;
# upgrade to https:// before submitting. Note: signed OSS URLs with
# query params (Expires, Signature) may also be rejected — prefer clean
# public file URLs when possible.
if audio_url.startswith("http://"):
audio_url = "https://" + audio_url[len("http://"):]
inputs = {**inputs, "audio_url": audio_url}
start = time.time()
try:
result = self._transcribe(inputs, api_key=api_key)
except Exception as exc:
return ToolResult(
success=False,
error=f"DashScope ASR failed: {self._safe_error(exc)}",
)
result.duration_seconds = round(time.time() - start, 2)
return result
def _transcribe(
self, inputs: dict[str, Any], *, api_key: str
) -> ToolResult:
import json
import requests
payload = self._build_payload(inputs)
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-DashScope-Async": "enable",
}
# Submit
submit_resp = requests.post(
self.SUBMIT_URL, headers=headers, json=payload, timeout=(10, 60)
)
submit_data = self._json_or_raise(submit_resp)
self._raise_for_error(submit_resp.status_code, submit_data)
task_id = submit_data.get("output", {}).get("task_id")
if not task_id:
raise RuntimeError(
"DashScope ASR submit succeeded but did not return "
"output.task_id"
)
# Poll
poll_data = self._poll_task(
requests_module=requests,
api_key=api_key,
task_id=task_id,
poll_interval=float(inputs.get("poll_interval_seconds", 5.0)),
timeout_seconds=int(inputs.get("timeout_seconds", 300)),
)
# qwen3-asr-flash-filetrans returns output.result.transcription_url
# (singular "result", NOT "results" array like paraformer-v2)
result = poll_data.get("output", {}).get("result", {})
transcription_url = result.get("transcription_url")
if not transcription_url:
raise RuntimeError(
"DashScope ASR task succeeded but "
"result.transcription_url missing"
)
# Download transcription JSON
trans_resp = requests.get(transcription_url, timeout=120)
trans_resp.raise_for_status()
transcription = trans_resp.json()
# Save full transcription
output_path = Path(
inputs.get("output_path", "dashscope_asr.json")
)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(
json.dumps(transcription, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
# Parse word-level timestamps (normalize ms -> seconds)
words = self._extract_words(transcription)
transcripts = transcription.get("transcripts", [])
return ToolResult(
success=True,
data={
"provider": "dashscope",
"model": payload["model"],
"audio_url": inputs["audio_url"],
"task_id": task_id,
"transcripts": transcripts,
"words": words,
"word_count": len(words),
"output": str(output_path),
},
artifacts=[str(output_path)],
cost_usd=self.estimate_cost(inputs),
model=payload["model"],
)
def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
return {
"model": inputs.get(
"model", "qwen3-asr-flash-filetrans"
),
"input": {
"file_url": inputs["audio_url"],
},
"parameters": {
"enable_words": bool(inputs.get("enable_words", True)),
"language_hints": inputs.get(
"language_hints", ["zh", "en"]
),
},
}
def _poll_task(
self,
*,
requests_module: Any,
api_key: str,
task_id: str,
poll_interval: float,
timeout_seconds: int,
) -> dict[str, Any]:
deadline = time.time() + timeout_seconds
headers = {"Authorization": f"Bearer {api_key}"}
while time.time() < deadline:
time.sleep(poll_interval)
resp = requests_module.get(
self.POLL_URL_TEMPLATE.format(task_id=task_id),
headers=headers,
timeout=(10, 60),
)
data = self._json_or_raise(resp)
self._raise_for_error(resp.status_code, data)
status = data.get("output", {}).get("task_status")
if status == "SUCCEEDED":
return data
if status == "FAILED":
msg = data.get("output", {}).get(
"message", "unknown error"
)
raise RuntimeError(
f"DashScope ASR task failed: {msg}"
)
raise TimeoutError(
f"DashScope ASR task {task_id} did not finish within "
f"{timeout_seconds}s"
)
@staticmethod
def _is_public_url(url: str) -> bool:
return url.startswith("http://") or url.startswith("https://")
@staticmethod
def _extract_words(
transcription: dict[str, Any]
) -> list[dict[str, Any]]:
"""Extract flat word list with timestamps normalized to seconds."""
words: list[dict[str, Any]] = []
for transcript in transcription.get("transcripts", []):
for sentence in transcript.get("sentences", []):
for word in sentence.get("words", []):
words.append(
{
"text": word.get("text", ""),
"begin_time_seconds": round(
word.get("begin_time", 0) / 1000.0, 3
),
"end_time_seconds": round(
word.get("end_time", 0) / 1000.0, 3
),
}
)
return words
@staticmethod
def _json_or_raise(response: Any) -> dict[str, Any]:
try:
return response.json()
except ValueError as exc:
raise RuntimeError(
f"Non-JSON response from DashScope API: "
f"HTTP {response.status_code}"
) from exc
def _raise_for_error(
self, http_status: int, payload: dict[str, Any]
) -> None:
if http_status < 400:
return
code = payload.get("code")
message = payload.get("message", "unknown error")
raise RuntimeError(
f"DashScope API error: HTTP {http_status}, "
f"code {code}: {message}"
)
@staticmethod
def _safe_error(exc: Exception) -> str:
return str(exc).replace(
os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]"
)

41
tools/audio/audio_mixer.py
View File

@@ -492,17 +492,22 @@ class AudioMixer(BaseTool):
duck_enabled = ducking.get("enabled", True) if isinstance(ducking, dict) else bool(ducking)
if duck_enabled and speech_tracks and music_tracks:
# Mix speech tracks together first
# Build ONE speech stream, then split it into two independent
# branches: one feeds the sidechain compressor as the ducking key,
# the other is mixed into the final output. A filtergraph label may
# only be consumed once, so reusing the same speech label for both
# the sidechain key and the output mix is invalid on stricter ffmpeg
# builds (e.g. the Linux ffmpeg on CI). asplit makes the fork explicit.
speech_indices = list(range(len(speech_tracks)))
speech_labels = "".join(f"[a{i}]" for i in speech_indices)
if len(speech_tracks) > 1:
filter_parts.append(
f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_mix]"
f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_all]"
)
speech_out = "[speech_mix]"
else:
speech_out = f"[a{speech_indices[0]}]"
filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_all]")
filter_parts.append("[speech_all]asplit=2[speech_key][speech_out]")
# Mix music tracks together
music_start = len(speech_tracks)
@@ -517,42 +522,20 @@ class AudioMixer(BaseTool):
else:
music_in = f"[a{music_indices[0]}]"
# Apply sidechain ducking
# Apply sidechain ducking — music is compressed, [speech_key] is the key
duck_params = ducking if isinstance(ducking, dict) else {}
attack = duck_params.get("attack_ms", 200) / 1000
release = duck_params.get("release_ms", 500) / 1000
music_vol = duck_params.get("music_volume_during_speech", 0.15)
filter_parts.append(
f"{music_in}{speech_out}sidechaincompress="
f"{music_in}[speech_key]sidechaincompress="
f"threshold=0.02:ratio=9:attack={attack}:release={release}:"
f"level_sc=1:mix=0.9[ducked_music];"
f"[ducked_music]volume={music_vol * 3}[music_out]"
)
# Duplicate speech for final mix (sidechain consumes it as key)
filter_parts.append(
f"{speech_out}acopy[speech_dup]" if speech_out.startswith("[a") else ""
)
# Re-mix speech path: we need speech audio in the output too
# Simpler approach: use amix on original speech and ducked music
# Reset: use a cleaner approach — amerge the speech mix and ducked music
# Actually, let's rebuild. The sidechain approach above uses speech as
# the key signal but doesn't consume it from the output chain.
# FFmpeg sidechaincompress: input 0 = audio to compress, input 1 = key signal
# So music is compressed, speech signal is the key. We need to mix them.
# Remove the last filter_part (the acopy that may be empty)
if filter_parts and filter_parts[-1] == "":
filter_parts.pop()
# Build speech mix for output separately
if len(speech_tracks) > 1:
# speech_mix already exists, make a copy for output
filter_parts.append(f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_out]")
else:
filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_out]")
# Final mix: speech_out + music_out
# Final mix: the other speech branch + ducked music
mix_label = "[speech_out][music_out]amix=inputs=2:duration=longest[premix]"
# Add SFX if present

243
tools/audio/dashscope_tts.py Normal file
View File

@@ -0,0 +1,243 @@
"""DashScope (Alibaba Cloud Bailian) text-to-speech via Qwen-TTS models.
Uses the DashScope-native multimodal-generation endpoint (same as image gen).
The response contains a temporary audio URL (WAV, valid ~24h) that must be
downloaded separately — unlike OpenAI TTS which returns raw audio bytes.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class DashscopeTTS(BaseTool):
name = "dashscope_tts"
version = "0.1.0"
tier = ToolTier.VOICE
capability = "tts"
provider = "dashscope"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.SYNC
determinism = Determinism.STOCHASTIC
runtime = ToolRuntime.API
dependencies = []
install_instructions = (
"Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n"
" Get one at https://dashscope.aliyun.com/"
)
fallback = "piper_tts"
fallback_tools = [
"doubao_tts",
"elevenlabs_tts",
"openai_tts",
"piper_tts",
]
agent_skills = ["dashscope"]
capabilities = [
"text_to_speech",
"voice_selection",
"multilingual",
]
supports = {
"voice_cloning": False,
"multilingual": True,
"offline": False,
"native_audio": True,
}
best_for = [
"natural Mandarin and multilingual narration via Qwen-TTS",
"cost-effective TTS via Alibaba Cloud",
"Chinese-language voiceover production",
]
not_good_for = [
"fully offline production",
"voice clone matching",
]
input_schema = {
"type": "object",
"required": ["text"],
"properties": {
"text": {
"type": "string",
"description": (
"Text to convert to speech "
"(max 600 chars for qwen3-tts-flash)."
),
},
"model": {
"type": "string",
"enum": [
"qwen3-tts-flash",
"qwen3-tts-instruct-flash",
"qwen-tts-2025-05-22",
],
"default": "qwen3-tts-flash",
},
"voice": {
"type": "string",
"default": "Cherry",
"description": (
'DashScope voice name. Examples: "Cherry", "Ethan", '
'"Chelsie".'
),
},
"language_type": {
"type": "string",
"default": "Auto",
"enum": ["Auto", "Chinese", "English", "Japanese", "Korean"],
"description": "Language hint for the TTS model.",
},
"instructions": {
"type": "string",
"description": (
"Natural language delivery instructions "
"(only for qwen3-tts-instruct-flash)."
),
},
"output_path": {"type": "string"},
},
}
resource_profile = ResourceProfile(
cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True
)
retry_policy = RetryPolicy(
max_retries=2, retryable_errors=["rate_limit", "timeout"]
)
idempotency_key_fields = ["text", "voice", "model", "language_type", "instructions"]
side_effects = [
"writes audio file to output_path",
"calls DashScope (Alibaba Cloud) TTS API",
]
user_visible_verification = [
"Listen to generated audio for naturalness and pacing"
]
ENDPOINT = (
"https://dashscope.aliyuncs.com/api/v1/services/aigc/"
"multimodal-generation/generation"
)
def get_status(self) -> ToolStatus:
if os.environ.get("DASHSCOPE_API_KEY"):
return ToolStatus.AVAILABLE
return ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
# Conservative per-character estimate; DashScope bills by character.
return round(len(inputs.get("text", "")) * 0.000015, 4)
def execute(self, inputs: dict[str, Any]) -> ToolResult:
api_key = os.environ.get("DASHSCOPE_API_KEY")
if not api_key:
return ToolResult(
success=False,
error="DASHSCOPE_API_KEY not set. " + self.install_instructions,
)
import requests
from tools.analysis.audio_probe import probe_duration
start = time.time()
try:
payload = self._build_payload(inputs)
response = requests.post(
self.ENDPOINT,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json=payload,
timeout=120,
)
response.raise_for_status()
data = response.json()
audio_info = data.get("output", {}).get("audio", {})
audio_url = audio_info.get("url")
if not audio_url:
return ToolResult(
success=False,
error="DashScope TTS returned no audio URL",
)
# Download the audio from the temporary URL (valid ~24h).
download = requests.get(audio_url, timeout=120)
download.raise_for_status()
output_path = Path(
inputs.get("output_path", "dashscope_tts.wav")
)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(download.content)
audio_duration = probe_duration(output_path)
usage = data.get("usage", {})
except Exception as e:
return ToolResult(
success=False,
error=f"DashScope TTS failed: {self._safe_error(e)}",
)
return ToolResult(
success=True,
data={
"provider": "dashscope",
"model": payload["model"],
"voice": payload["input"]["voice"],
"language_type": payload["input"].get("language_type", "Auto"),
"text_length": len(inputs["text"]),
"audio_duration_seconds": (
round(audio_duration, 2) if audio_duration else None
),
"output": str(output_path),
"audio_url": audio_url,
"usage": usage,
},
artifacts=[str(output_path)],
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - start, 2),
model=payload["model"],
)
def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
input_data: dict[str, Any] = {
"text": inputs["text"],
"voice": inputs.get("voice", "Cherry"),
"language_type": inputs.get("language_type", "Auto"),
}
if inputs.get("instructions"):
input_data["instructions"] = inputs["instructions"]
input_data["optimize_instructions"] = True
return {
"model": inputs.get("model", "qwen3-tts-flash"),
"input": input_data,
}
@staticmethod
def _safe_error(exc: Exception) -> str:
return str(exc).replace(
os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]"
)

6
tools/audio/piper_tts.py
View File

@@ -98,11 +98,7 @@ class PiperTTS(BaseTool):
def get_status(self) -> ToolStatus:
if shutil.which("piper"):
return ToolStatus.AVAILABLE
try:
import piper # noqa: F401
return ToolStatus.AVAILABLE
except ImportError:
return ToolStatus.UNAVAILABLE
return ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
return 0.0

95
tools/base_tool.py
View File

@@ -6,6 +6,7 @@ interface for discovery, execution, cost estimation, and health reporting.
from __future__ import annotations
import functools
import hashlib
import inspect
import json
@@ -13,6 +14,7 @@ import os
import platform
import subprocess
import shutil
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
@@ -136,9 +138,102 @@ class ToolResult:
model: Optional[str] = None
import threading as _threading
# Shared nesting counter for instrumented execute() calls (thread-local so
# parallel tool threads don't see each other's depth).
_EXECUTE_DEPTH = _threading.local()
def _instrument_execute(fn: Callable) -> Callable:
"""Wrap a tool's execute() with Backlot event emission.
Appends start/finish/error entries to the owning project's events.jsonl
when the call can be attributed to a project (explicit project_dir input
or any path input under projects/). Powers the board's live activity
ticker and per-scene generating states with zero agent involvement.
Instrumentation is strictly non-fatal: any failure inside the event layer
is swallowed and the tool call proceeds untouched.
"""
if getattr(fn, "_backlot_instrumented", False):
return fn
depth_state = _EXECUTE_DEPTH # shared across all tools (selector → provider)
@functools.wraps(fn)
def wrapper(self, inputs: Any, *args: Any, **kwargs: Any):
# Event layer is fully optional: if it can't import, run untouched.
try:
from lib.events import emit_event, infer_project_dir
except Exception:
return fn(self, inputs, *args, **kwargs)
tool_name = getattr(self, "name", "") or self.__class__.__name__
scene_id = inputs.get("scene_id") if isinstance(inputs, dict) else None
output_path = inputs.get("output_path") if isinstance(inputs, dict) else None
# Nesting depth: selector tools delegate to provider tools' execute().
# Both emit (the ticker wants the provider name too), but depth lets
# consumers dedupe — e.g. sum cost_usd only at depth 0.
depth = getattr(depth_state, "value", 0)
depth_state.value = depth + 1
project_dir = infer_project_dir(inputs)
base = {
"tool": tool_name,
"scene_id": scene_id,
"depth": depth if depth else None,
}
if project_dir is not None:
emit_event(project_dir, {
**base, "event": "start",
"output_path": str(output_path) if output_path else None,
})
started = time.monotonic()
try:
result = fn(self, inputs, *args, **kwargs)
except Exception as exc:
if project_dir is not None:
emit_event(project_dir, {
**base, "event": "error",
"error": str(exc)[:300],
"duration_s": round(time.monotonic() - started, 2),
})
raise
finally:
depth_state.value = depth
if project_dir is None:
# The tool may have created its own project dir during execute
# (first call of a run) — attribute the finish if possible.
project_dir = infer_project_dir(inputs)
if project_dir is not None:
cost = getattr(result, "cost_usd", None)
emit_event(project_dir, {
**base, "event": "finish",
"output_path": str(output_path) if output_path else None,
"success": getattr(result, "success", None),
# NOTE: 0.0 is meaningful (ran for free) — only None is dropped.
"cost_usd": cost if isinstance(cost, (int, float)) else None,
"duration_s": round(time.monotonic() - started, 2),
})
return result
wrapper._backlot_instrumented = True # type: ignore[attr-defined]
return wrapper
class BaseTool(ABC):
"""Abstract base class for all OpenMontage tools."""
def __init_subclass__(cls, **kwargs: Any) -> None:
"""Auto-instrument every concrete execute() with Backlot events."""
super().__init_subclass__(**kwargs)
impl = cls.__dict__.get("execute")
if impl is not None and not getattr(impl, "__isabstractmethod__", False):
cls.execute = _instrument_execute(impl)
# --- Identity (override in subclasses) ---
name: str = ""
version: str = "0.1.0"

273
tools/graphics/dashscope_image.py Normal file
View File

@@ -0,0 +1,273 @@
"""DashScope (Alibaba Cloud Bailian) image generation via Qwen-Image models.
Uses the DashScope-native multimodal-generation endpoint (NOT OpenAI-compatible
mode, which only supports /chat/completions and /embeddings). The response
contains a temporary image URL (valid ~24h) that must be downloaded separately.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class DashscopeImage(BaseTool):
name = "dashscope_image"
version = "0.1.0"
tier = ToolTier.GENERATE
capability = "image_generation"
provider = "dashscope"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.SYNC
determinism = Determinism.STOCHASTIC
runtime = ToolRuntime.API
dependencies = []
install_instructions = (
"Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n"
" Get one at https://dashscope.aliyun.com/"
)
fallback = "grok_image"
fallback_tools = ["grok_image", "openai_image", "flux_image", "recraft_image"]
agent_skills = ["dashscope"]
capabilities = ["generate_image", "text_to_image"]
supports = {
"multiple_outputs": True,
"aspect_ratio": True,
"resolution": True,
"negative_prompt": True,
"seed": True,
}
best_for = [
"high-quality image generation with Qwen-Image models",
"Chinese-language prompt understanding",
"cost-effective image generation via Alibaba Cloud",
]
not_good_for = ["offline generation", "image editing (use grok_image edit mode)"]
input_schema = {
"type": "object",
"required": ["prompt"],
"properties": {
"prompt": {"type": "string"},
"model": {
"type": "string",
"enum": [
"qwen-image-2.0-pro",
"qwen-image-max",
"wan2.7-image",
"z-image-turbo",
],
"default": "qwen-image-2.0-pro",
},
"size": {
"type": "string",
"default": "1024*1024",
"description": (
'Image size as "W*H" (asterisk separator, NOT "x"). '
'Examples: "1024*1024", "2048*2048", "2688*1536".'
),
},
"n": {"type": "integer", "default": 1, "minimum": 1, "maximum": 6},
"negative_prompt": {
"type": "string",
"description": "Negative prompt (max 500 chars). Things to avoid in the image.",
},
"prompt_extend": {
"type": "boolean",
"default": True,
"description": "Enable DashScope prompt auto-rewrite for better results.",
},
"watermark": {"type": "boolean", "default": False},
"seed": {"type": "integer", "minimum": 0, "maximum": 2147483647},
"output_path": {"type": "string"},
},
}
resource_profile = ResourceProfile(
cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=100, network_required=True
)
retry_policy = RetryPolicy(
max_retries=2, retryable_errors=["rate_limit", "timeout"]
)
idempotency_key_fields = [
"prompt",
"model",
"size",
"n",
"negative_prompt",
"seed",
"prompt_extend",
"watermark",
]
side_effects = [
"writes image file to output_path",
"calls DashScope (Alibaba Cloud) image generation API",
]
user_visible_verification = [
"Inspect generated image for relevance and quality"
]
ENDPOINT = (
"https://dashscope.aliyuncs.com/api/v1/services/aigc/"
"multimodal-generation/generation"
)
def get_status(self) -> ToolStatus:
if os.environ.get("DASHSCOPE_API_KEY"):
return ToolStatus.AVAILABLE
return ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
# Conservative per-image estimate; DashScope bills per image.
# Check the DashScope console for actual pricing.
n = int(inputs.get("n", 1))
return n * 0.02
def execute(self, inputs: dict[str, Any]) -> ToolResult:
api_key = os.environ.get("DASHSCOPE_API_KEY")
if not api_key:
return ToolResult(
success=False,
error="DASHSCOPE_API_KEY not set. " + self.install_instructions,
)
import requests
start = time.time()
try:
payload = self._build_payload(inputs)
response = requests.post(
self.ENDPOINT,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json=payload,
timeout=180,
)
response.raise_for_status()
data = response.json()
image_urls = self._extract_image_urls(data)
if not image_urls:
return ToolResult(
success=False,
error="DashScope returned no image URLs",
)
# DashScope bills per image and URLs expire ~24h; save every one.
output_paths = self._resolve_output_paths(
inputs.get("output_path", "dashscope_image.png"),
count=len(image_urls),
)
for path, url in zip(output_paths, image_urls):
download = requests.get(url, timeout=120)
download.raise_for_status()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(download.content)
usage = data.get("usage", {})
n_generated = len(image_urls)
except Exception as e:
return ToolResult(
success=False,
error=f"DashScope image generation failed: {self._safe_error(e)}",
)
return ToolResult(
success=True,
data={
"provider": "dashscope",
"model": payload["model"],
"prompt": inputs["prompt"],
"size": payload["parameters"]["size"],
"output": str(output_paths[0]),
"outputs": [str(p) for p in output_paths],
"images_generated": n_generated,
"usage": usage,
},
artifacts=[str(p) for p in output_paths],
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - start, 2),
model=payload["model"],
)
@staticmethod
def _extract_image_urls(data: dict[str, Any]) -> list[str]:
"""Collect image URLs from every choice whose finish_reason is "stop".
Per Qwen Cloud docs, a multi-output task is SUCCEEDED if at least one
image is generated; failed choices carry finish_reason != "stop" and
must be skipped to avoid downloading partial/empty results.
"""
urls: list[str] = []
for choice in data.get("output", {}).get("choices", []):
if choice.get("finish_reason") != "stop":
continue
for item in choice.get("message", {}).get("content", []):
url = item.get("image")
if url:
urls.append(url)
return urls
@staticmethod
def _resolve_output_paths(base: str, count: int) -> list[Path]:
"""Derive distinct paths for `count` images. Single image keeps the
base path unchanged; multiple images insert an index before the
extension (foo.png -> foo_1.png, foo_2.png, ...)."""
base_path = Path(base)
if count <= 1:
return [base_path]
stem = base_path.stem
suffix = base_path.suffix
parent = base_path.parent
return [parent / f"{stem}_{i}{suffix}" for i in range(1, count + 1)]
def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
parameters: dict[str, Any] = {
"size": inputs.get("size", "1024*1024"),
"n": int(inputs.get("n", 1)),
"prompt_extend": bool(inputs.get("prompt_extend", True)),
"watermark": bool(inputs.get("watermark", False)),
}
if inputs.get("negative_prompt"):
parameters["negative_prompt"] = inputs["negative_prompt"]
if inputs.get("seed") is not None:
parameters["seed"] = int(inputs["seed"])
return {
"model": inputs.get("model", "qwen-image-2.0-pro"),
"input": {
"messages": [
{
"role": "user",
"content": [{"text": inputs["prompt"]}],
}
]
},
"parameters": parameters,
}
@staticmethod
def _safe_error(exc: Exception) -> str:
return str(exc).replace(
os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]"
)

12
tools/graphics/image_gen.py
View File

@@ -7,7 +7,7 @@
pexels_image, pixabay_image). This file is kept for backwards
compatibility and will be removed in a future release.
Supports cloud API providers (FLUX via fal.ai/Replicate, OpenAI DALL-E)
Supports cloud API providers (FLUX via fal.ai/Replicate, OpenAI GPT Image)
and local Stable Diffusion via diffusers. Reports unavailable with
install instructions when no provider is configured.
"""
@@ -43,12 +43,12 @@ class ImageGen(BaseTool):
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.SYNC
determinism = Determinism.SEEDED
runtime = ToolRuntime.HYBRID # API (DALL-E/FLUX) or local (diffusers)
runtime = ToolRuntime.HYBRID # API (GPT Image/FLUX) or local (diffusers)
dependencies = [] # checked dynamically based on provider
install_instructions = (
"Set one of these environment variables:\n"
" OPENAI_API_KEY — for DALL-E 3\n"
" OPENAI_API_KEY — for GPT Image 2\n"
" FAL_KEY — for FLUX via fal.ai\n"
"Or install diffusers for local generation:\n"
" pip install diffusers transformers accelerate torch"
@@ -121,7 +121,7 @@ class ImageGen(BaseTool):
def estimate_cost(self, inputs: dict[str, Any]) -> float:
provider = inputs.get("provider") or self._detect_provider()
if provider == "openai":
return 0.04 # DALL-E 3 standard
return 0.053 # gpt-image-2 medium at 1024x1024 (call uses auto quality)
if provider == "flux":
return 0.03
return 0.0 # local
@@ -159,14 +159,14 @@ class ImageGen(BaseTool):
client = OpenAI()
prompt = inputs["prompt"]
size = f"{inputs.get('width', 1024)}x{inputs.get('height', 1024)}"
model = inputs.get("model", "dall-e-3")
model = inputs.get("model", "gpt-image-2")
# GPT image models don't accept response_format; they always return b64
response = client.images.generate(
model=model,
prompt=prompt,
size=size,
n=1,
response_format="b64_json",
)
image_data = base64.b64decode(response.data[0].b64_json)

125
tools/graphics/math_animate.py
View File

@@ -6,6 +6,7 @@ using the Manim Community Edition engine. Free, local, no API key required.
from __future__ import annotations
import ast
import os
import shutil
import subprocess
@@ -28,6 +29,48 @@ from tools.base_tool import (
)
# --- Safety: caller-supplied scene_code is a local code-execution boundary ---
# math_animate runs Manim on Python supplied by the caller (often an LLM or
# prompt-influenced reference material). That is arbitrary local code execution
# (see issue #219). The static scan below is defense-in-depth: it blocks the
# constructs an attack needs — reading secrets/SSH material, opening network
# connections, spawning subprocesses — while leaving genuine math/animation
# scenes untouched. It is NOT a security sandbox: a determined attacker can
# evade a static denylist, so it is paired with an explicit `allow_unsafe_code`
# opt-out and a tool contract that names the boundary. A passing scan is not
# proof that code is safe to run.
_BLOCKED_IMPORTS = frozenset({
"os", "sys", "subprocess", "socket", "shutil", "requests", "urllib",
"http", "ftplib", "smtplib", "telnetlib", "ctypes", "pickle", "marshal",
"importlib", "builtins", "multiprocessing", "threading", "pty", "glob",
"resource", "signal", "tempfile", "webbrowser", "pathlib",
})
# Dangerous identifiers blocked wherever they appear as a bare name — not just
# as a direct call. This catches indirection like `__builtins__['open']`,
# `f = open`, or `getattr(x, '__class__')` that a call-target-only or
# attribute-only check would miss.
_BLOCKED_NAMES = frozenset({
"eval", "exec", "compile", "__import__", "open", "input", "breakpoint",
"__builtins__", "__loader__", "globals", "locals", "vars",
"getattr", "setattr", "delattr",
})
# Reflection via dunder attributes is the general escape hatch: `().__class__`,
# `print.__self__` (the builtins module), `x.__globals__`, `f.__reduce__`, etc.
# Enumerating dangerous dunders one by one is whack-a-mole, so block ALL dunder
# *attribute access* and allow only a tiny set that legitimate scenes use
# (`super().__init__(...)`, occasional `Type.__name__`). A dunder is any name
# that starts and ends with double underscores.
_ALLOWED_DUNDER_ATTRS = frozenset({"__init__", "__name__"})
def _is_blocked_dunder(attr: str) -> bool:
return (
attr.startswith("__")
and attr.endswith("__")
and attr not in _ALLOWED_DUNDER_ATTRS
)
# Quality presets mapping to Manim CLI flags
QUALITY_PRESETS = {
"low": {"flag": "-ql", "resolution": "854x480", "fps": 15},
@@ -76,7 +119,20 @@ class MathAnimate(BaseTool):
"description": (
"Python code defining a Manim scene. Must contain a class "
"inheriting from Scene with a construct() method. "
"Import 'from manim import *' is auto-added if missing."
"Import 'from manim import *' is auto-added if missing. "
"SECURITY: this code is EXECUTED on the host by Manim. It is "
"scanned for dangerous constructs (system/network/subprocess "
"access) and rejected by default; treat scene_code as trusted "
"input only."
),
},
"allow_unsafe_code": {
"type": "boolean",
"default": False,
"description": (
"Bypass the scene_code safety scan. Only set this for code "
"you fully trust — it permits arbitrary local code execution "
"(filesystem, network, subprocess). See issue #219."
),
},
"scene_name": {
@@ -117,7 +173,13 @@ class MathAnimate(BaseTool):
)
retry_policy = RetryPolicy(max_retries=1, retryable_errors=["timeout"])
idempotency_key_fields = ["scene_code", "scene_name", "quality"]
side_effects = ["writes video/image file to output_path", "creates temp files"]
side_effects = [
"EXECUTES caller-supplied Python (Manim scene_code) on the host — this "
"is a local code-execution boundary; scene_code is scanned and rejected "
"by default unless allow_unsafe_code=true (see issue #219)",
"writes video/image file to output_path",
"creates temp files",
]
user_visible_verification = [
"Watch the animation for correctness and visual quality",
"Verify math formulas render correctly (requires LaTeX)",
@@ -160,6 +222,48 @@ class MathAnimate(BaseTool):
result.duration_seconds = round(time.time() - start, 2)
return result
@staticmethod
def _scan_scene_code(code: str) -> list[str]:
"""Static safety scan of caller-supplied Manim scene code (issue #219).
Returns a de-duplicated list of disallowed constructs (dangerous
imports, builtins, and sandbox-escape dunders). Empty list means the
scan found nothing to block — which is NOT a guarantee the code is safe.
A syntax error is left for Manim to report, so it returns no violations.
"""
try:
tree = ast.parse(code)
except SyntaxError:
return []
violations: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
root = alias.name.split(".")[0]
if root in _BLOCKED_IMPORTS:
violations.append(f"import '{alias.name}'")
elif isinstance(node, ast.ImportFrom):
root = (node.module or "").split(".")[0]
if root in _BLOCKED_IMPORTS:
violations.append(f"from '{node.module}' import ...")
elif isinstance(node, ast.Name):
# Blocks direct calls (eval(...)) and indirection alike:
# `__builtins__['open']`, `f = open`, `getattr(o, '__class__')`.
if node.id in _BLOCKED_NAMES:
violations.append(f"use of '{node.id}'")
elif isinstance(node, ast.Attribute):
if _is_blocked_dunder(node.attr):
violations.append(f"dunder attribute access '.{node.attr}'")
seen: set[str] = set()
deduped: list[str] = []
for v in violations:
if v not in seen:
seen.add(v)
deduped.append(v)
return deduped
def _render(self, inputs: dict[str, Any]) -> ToolResult:
scene_code = inputs["scene_code"]
scene_name = inputs.get("scene_name")
@@ -174,6 +278,23 @@ class MathAnimate(BaseTool):
if "from manim import" not in scene_code:
scene_code = "from manim import *\n\n" + scene_code
# Safety gate: scene_code is executed on the host by Manim. Reject
# dangerous constructs unless the caller explicitly opts out. (issue #219)
if not inputs.get("allow_unsafe_code", False):
violations = self._scan_scene_code(scene_code)
if violations:
return ToolResult(
success=False,
error=(
"scene_code blocked by the math_animate safety scan. This "
"tool executes caller-supplied Python on the host; the "
"following constructs are disallowed by default:\n - "
+ "\n - ".join(violations)
+ "\nIf you fully trust this code and require them, pass "
"allow_unsafe_code=true. See issue #219."
),
)
# Auto-detect scene name if not provided
if not scene_name:
scene_name = self._detect_scene_name(scene_code)

60
tools/graphics/openai_image.py
View File

@@ -1,4 +1,4 @@
"""OpenAI GPT Image generation (gpt-image-1 / DALL-E 3)."""
"""OpenAI GPT Image generation (gpt-image-2)."""
from __future__ import annotations
@@ -60,20 +60,17 @@ class OpenAIImage(BaseTool):
"prompt": {"type": "string"},
"model": {
"type": "string",
"enum": ["gpt-image-1", "dall-e-3"],
"default": "gpt-image-1",
"enum": ["gpt-image-2"],
"default": "gpt-image-2",
},
"size": {
"type": "string",
"enum": [
"1024x1024", "1536x1024", "1024x1536", "auto",
"1024x1792", "1792x1024", # dall-e-3 only
],
"enum": ["1024x1024", "1536x1024", "1024x1536", "auto"],
"default": "1024x1024",
},
"quality": {
"type": "string",
"enum": ["low", "medium", "high", "auto", "standard", "hd"],
"enum": ["low", "medium", "high", "auto"],
"default": "high",
},
"output_format": {
@@ -100,15 +97,12 @@ class OpenAIImage(BaseTool):
return ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
model = inputs.get("model", "gpt-image-1")
# gpt-image-2 per-image pricing at 1024x1024 (non-square sizes run
# slightly cheaper): https://developers.openai.com/api/docs/guides/image-generation
quality = inputs.get("quality", "high")
n = inputs.get("n", 1)
if model == "gpt-image-1":
cost_map = {"low": 0.011, "medium": 0.042, "high": 0.167, "auto": 0.042}
return cost_map.get(quality, 0.042) * n
# dall-e-3 fallback pricing
quality_map = {"standard": 0.04, "hd": 0.08}
return quality_map.get(quality, 0.04) * n
cost_map = {"low": 0.006, "medium": 0.053, "high": 0.211, "auto": 0.053}
return cost_map.get(quality, 0.053) * n
def execute(self, inputs: dict[str, Any]) -> ToolResult:
if not os.environ.get("OPENAI_API_KEY"):
@@ -121,36 +115,22 @@ class OpenAIImage(BaseTool):
start = time.time()
client = OpenAI()
model = inputs.get("model", "gpt-image-1")
model = inputs.get("model", "gpt-image-2")
prompt = inputs["prompt"]
size = inputs.get("size", "1024x1024")
n = inputs.get("n", 1)
try:
if model == "gpt-image-1":
quality = inputs.get("quality", "high")
output_format = inputs.get("output_format", "png")
response = client.images.generate(
model=model,
prompt=prompt,
size=size,
quality=quality,
output_format=output_format,
n=n,
)
else:
# dall-e-3 path
quality = inputs.get("quality", "standard")
if quality in ("low", "medium", "high", "auto"):
quality = "standard" # map to dall-e-3 quality options
response = client.images.generate(
model=model,
prompt=prompt,
size=size,
quality=quality,
n=1, # dall-e-3 only supports n=1
response_format="b64_json",
)
quality = inputs.get("quality", "high")
output_format = inputs.get("output_format", "png")
response = client.images.generate(
model=model,
prompt=prompt,
size=size,
quality=quality,
output_format=output_format,
n=n,
)
image_data = base64.b64decode(response.data[0].b64_json)
ext = inputs.get("output_format", "png")

307
tools/publishers/export_bundle.py Normal file
View File

@@ -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/<project_name>'.",
},
"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/<name>/renders/...`` (the OpenMontage
convention), default the bundle to ``projects/<name>/exports/`` alongside
``artifacts/``, ``assets/`` and ``renders/``. Otherwise fall back to a
top-level ``exports/<project_name>/``.
"""
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/<name>/renders/final.mp4 -> <name>; fall back to the file stem.
parents = video_path.resolve().parents
if len(parents) >= 2:
return parents[1].name
return video_path.stem

206
tools/video/direct_clip_search.py
View File

@@ -33,6 +33,7 @@ No CLIP model. No embeddings. No corpus index. Just files on disk.
"""
from __future__ import annotations
from contextlib import contextmanager
import subprocess
import time
import urllib.parse
@@ -53,6 +54,10 @@ from tools.base_tool import (
)
class _DeadlineExceeded(TimeoutError):
"""Raised when the direct-clip-search wall-clock deadline is exhausted."""
class DirectClipSearch(BaseTool):
name = "direct_clip_search"
version = "0.1.0"
@@ -178,6 +183,16 @@ class DirectClipSearch(BaseTool):
"default": True,
"description": "Skip download if a file with the same clip_id already exists.",
},
"timeout_seconds": {
"type": "number",
"default": 600,
"minimum": 1,
"description": (
"Overall wall-clock deadline for search, download, and thumbnail "
"work. Defaults to 10 minutes. On timeout, returns partial progress "
"instead of relying on an external process interrupt."
),
},
},
}
@@ -245,6 +260,8 @@ class DirectClipSearch(BaseTool):
clips_per_query = int(inputs.get("clips_per_query", 3))
extract_thumbs = bool(inputs.get("extract_thumbnails", True))
skip_existing = bool(inputs.get("skip_existing", True))
timeout_seconds = float(inputs.get("timeout_seconds", 600))
deadline = start + timeout_seconds
clips_dir = output_dir / "clips"
thumbs_dir = output_dir / "thumbnails"
@@ -295,9 +312,53 @@ class DirectClipSearch(BaseTool):
errors: list[dict] = []
skipped = 0
per_source_counts: dict[str, int] = {s.name: 0 for s in sources}
queries_started = 0
def timeout_result(
*,
phase: str,
query: str = "",
source: str = "",
clip_id: str = "",
) -> ToolResult:
elapsed = time.time() - start
return ToolResult(
success=False,
error=(
f"Direct clip search timed out after {timeout_seconds:.1f}s "
f"during {phase}."
),
data={
"timed_out": True,
"phase": phase,
"query": query,
"source": source,
"clip_id": clip_id,
"output_dir": str(output_dir),
"clips_downloaded": len([d for d in downloaded if not d.get("skipped_existing")]),
"clips_reused": skipped,
"total_clips": len(downloaded),
"per_source_counts": per_source_counts,
"queries_run": queries_started,
"resolved_sources": [s.name for s in sources],
"clips": downloaded,
"errors": errors[:25],
"elapsed_seconds": round(elapsed, 2),
"timeout_seconds": timeout_seconds,
},
cost_usd=0.0,
duration_seconds=round(elapsed, 2),
)
def timed_out() -> bool:
return time.time() >= deadline
for q_spec in queries:
if timed_out():
return timeout_result(phase="query", query=q_spec.get("query", ""))
query = q_spec["query"]
queries_started += 1
slot_id = q_spec.get("slot_id", "")
kind = q_spec.get("kind", "video")
collected_for_query = 0
@@ -312,11 +373,17 @@ class DirectClipSearch(BaseTool):
)
for src in sources:
if timed_out():
return timeout_result(phase="search", query=query, source=src.name)
if collected_for_query >= clips_per_query:
break
try:
candidates = src.search(query, filters)
with _requests_deadline(deadline):
candidates = src.search(query, filters)
except _DeadlineExceeded:
return timeout_result(phase="search", query=query, source=src.name)
except Exception as e:
errors.append({
"phase": "search",
@@ -327,6 +394,14 @@ class DirectClipSearch(BaseTool):
continue
for cand in candidates:
if timed_out():
return timeout_result(
phase="download",
query=query,
source=src.name,
clip_id=cand.clip_id,
)
if collected_for_query >= clips_per_query:
break
@@ -362,7 +437,15 @@ class DirectClipSearch(BaseTool):
# Download
try:
src.download(cand, clip_path)
with _requests_deadline(deadline):
src.download(cand, clip_path)
except _DeadlineExceeded:
return timeout_result(
phase="download",
query=query,
source=src.name,
clip_id=clip_id,
)
except Exception as e:
errors.append({
"phase": "download",
@@ -386,21 +469,7 @@ class DirectClipSearch(BaseTool):
pass
continue
# Extract thumbnail
thumb_path_str = ""
if extract_thumbs and cand.kind == "video":
thumb_path = thumbs_dir / f"{clip_id}.jpg"
try:
_extract_mid_thumbnail(clip_path, thumb_path)
if thumb_path.exists():
thumb_path_str = str(thumb_path)
except Exception:
pass # thumbnail failure is non-fatal
per_source_counts[src.name] = per_source_counts.get(src.name, 0) + 1
collected_for_query += 1
downloaded.append({
downloaded_record = {
"clip_id": clip_id,
"source": cand.source,
"source_id": cand.source_id,
@@ -409,7 +478,7 @@ class DirectClipSearch(BaseTool):
"slot_id": slot_id,
"kind": cand.kind,
"path": str(clip_path),
"thumbnail": thumb_path_str,
"thumbnail": "",
"duration": cand.duration,
"width": cand.width,
"height": cand.height,
@@ -417,7 +486,38 @@ class DirectClipSearch(BaseTool):
"license": cand.license,
"source_tags": cand.source_tags,
"skipped_existing": False,
})
}
downloaded.append(downloaded_record)
per_source_counts[src.name] = per_source_counts.get(src.name, 0) + 1
collected_for_query += 1
# Extract thumbnail
if extract_thumbs and cand.kind == "video":
if timed_out():
return timeout_result(
phase="thumbnail",
query=query,
source=src.name,
clip_id=clip_id,
)
thumb_path = thumbs_dir / f"{clip_id}.jpg"
try:
_extract_mid_thumbnail(
clip_path,
thumb_path,
timeout_seconds=remaining_seconds(deadline),
)
if thumb_path.exists():
downloaded_record["thumbnail"] = str(thumb_path)
except _DeadlineExceeded:
return timeout_result(
phase="thumbnail",
query=query,
source=src.name,
clip_id=clip_id,
)
except Exception:
pass # thumbnail failure is non-fatal
elapsed = time.time() - start
@@ -429,7 +529,7 @@ class DirectClipSearch(BaseTool):
"clips_reused": skipped,
"total_clips": len(downloaded),
"per_source_counts": per_source_counts,
"queries_run": len(queries),
"queries_run": queries_started,
"resolved_sources": [s.name for s in sources],
"clips": downloaded,
"errors": errors[:25],
@@ -462,7 +562,64 @@ def _guess_ext(cand) -> str:
return ".mp4" if cand.kind == "video" else ".jpg"
def _extract_mid_thumbnail(video_path: Path, thumb_path: Path) -> None:
def remaining_seconds(deadline: float) -> float:
remaining = deadline - time.time()
if remaining <= 0:
raise _DeadlineExceeded("direct_clip_search deadline exceeded")
return remaining
def _clamp_timeout(timeout: Any, remaining: float) -> Any:
if timeout is None:
return remaining
if isinstance(timeout, tuple):
return tuple(min(float(part), remaining) for part in timeout)
try:
return min(float(timeout), remaining)
except (TypeError, ValueError):
return remaining
@contextmanager
def _requests_deadline(deadline: float):
"""Clamp adapter requests calls to the direct-search deadline.
Stock-source adapters are intentionally simple and call `requests.get`
directly. Keeping the deadline wrapper here avoids widening every adapter
method signature while still preventing streaming downloads from running
past the tool-level budget.
"""
import requests
original_get = requests.get
def get_with_deadline(*args, **kwargs):
remaining = remaining_seconds(deadline)
kwargs["timeout"] = _clamp_timeout(kwargs.get("timeout"), remaining)
response = original_get(*args, **kwargs)
original_iter_content = getattr(response, "iter_content", None)
if callable(original_iter_content):
def iter_content_with_deadline(*iter_args, **iter_kwargs):
for chunk in original_iter_content(*iter_args, **iter_kwargs):
remaining_seconds(deadline)
yield chunk
response.iter_content = iter_content_with_deadline
return response
requests.get = get_with_deadline
try:
yield
finally:
requests.get = original_get
def _extract_mid_thumbnail(
video_path: Path,
thumb_path: Path,
*,
timeout_seconds: float = 15,
) -> None:
"""Extract a single frame from the middle of the video via ffmpeg.
This is deliberately simple — one frame, no CLIP, no motion score.
@@ -470,6 +627,7 @@ def _extract_mid_thumbnail(video_path: Path, thumb_path: Path) -> None:
clip is a good match.
"""
thumb_path.parent.mkdir(parents=True, exist_ok=True)
deadline = time.time() + timeout_seconds
# Probe duration first
probe_cmd = [
@@ -479,8 +637,9 @@ def _extract_mid_thumbnail(video_path: Path, thumb_path: Path) -> None:
str(video_path),
]
try:
probe_timeout = min(10, remaining_seconds(deadline))
result = subprocess.run(
probe_cmd, capture_output=True, text=True, timeout=10
probe_cmd, capture_output=True, text=True, timeout=probe_timeout
)
duration = float(result.stdout.strip() or "0")
except (ValueError, subprocess.TimeoutExpired, FileNotFoundError):
@@ -497,7 +656,8 @@ def _extract_mid_thumbnail(video_path: Path, thumb_path: Path) -> None:
"-q:v", "3",
str(thumb_path),
]
extract_timeout = min(15, remaining_seconds(deadline))
subprocess.run(
extract_cmd, capture_output=True, timeout=15,
extract_cmd, capture_output=True, timeout=extract_timeout,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)

124
tools/video/video_compose.py
View File

@@ -17,9 +17,11 @@ Routing is driven by `edit_decisions.render_runtime` (locked at proposal):
Authoring mode is orthogonal to runtime. Setting
`edit_decisions.composition_mode = "atelier"` (or `renderer_family="bespoke"`)
routes to a hand-authored, project-local Remotion composition that BYPASSES the
cut-schema and the stock scene-type registry entirely — the "hand-stitched
every time" path for hero/bespoke pieces. See `_render_via_atelier`.
means the composition is hand-authored rather than assembled from stock scene
components. Runtime still wins first: HyperFrames atelier routes through
`hyperframes_compose`, FFmpeg stays FFmpeg-only, and only Remotion atelier uses
`_render_via_atelier` for a project-local Remotion entry that bypasses the
cut-schema and stock scene-type registry.
Silent runtime swaps are forbidden by governance. If the chosen runtime is
unavailable or fails, this tool surfaces a structured blocker and waits for
@@ -187,6 +189,15 @@ class VideoCompose(BaseTool):
"codec": {"type": "string", "default": "libx264"},
"crf": {"type": "integer", "default": 23},
"preset": {"type": "string", "default": "medium"},
"remotion_timeout_ms": {
"type": "integer",
"description": (
"Remotion render timeout in milliseconds, passed through as "
"`--timeout` (governs headless-browser setup and delayRender). "
"Raise this when the browser is slow to start (e.g. restricted "
"networks). The subprocess timeout is widened to match."
),
},
},
}
@@ -1293,6 +1304,36 @@ class VideoCompose(BaseTool):
if not edit_decisions:
return ToolResult(success=False, error="edit_decisions required for render")
# --- Runtime routing: honor render_runtime locked at proposal ---
# Silent swaps are forbidden by governance. Resolve this before any
# composition-mode branching so `composition_mode="atelier"` cannot
# accidentally force the Remotion atelier path when HyperFrames or
# FFmpeg was approved.
render_runtime = (edit_decisions.get("render_runtime") or "").strip().lower()
if not render_runtime:
return ToolResult(
success=False,
error=(
"render_runtime is not set in edit_decisions. Per governance, "
"it MUST be locked at proposal stage (proposal_packet."
"production_plan.render_runtime) and carried forward through "
"edit_decisions.render_runtime. Valid values: 'remotion', "
"'hyperframes', 'ffmpeg'. Re-run the proposal stage with an "
"explicit runtime choice — do NOT default this field."
),
)
if render_runtime not in {"remotion", "hyperframes", "ffmpeg"}:
return ToolResult(
success=False,
error=(
f"Unknown render_runtime {render_runtime!r}. "
f"Valid values: remotion, hyperframes, ffmpeg. "
f"render_runtime must be set at proposal stage."
),
)
# --- Atelier (bespoke) mode -------------------------------------
# Hand-authored, project-local Remotion composition. Deliberately
# bypasses the cut-schema, the stock scene-type registry, and the
@@ -1301,8 +1342,11 @@ class VideoCompose(BaseTool):
# under remotion-composer/projects/<slug>/ and points this renderer at
# it. No reusable creative components; a new visual language per video.
# Triggered by composition_mode="atelier" (or renderer_family="bespoke").
if (edit_decisions.get("composition_mode") == "atelier"
or edit_decisions.get("renderer_family") == "bespoke"):
remotion_atelier_requested = (
edit_decisions.get("composition_mode") == "atelier"
or edit_decisions.get("renderer_family") == "bespoke"
)
if render_runtime == "remotion" and remotion_atelier_requested:
return self._render_via_atelier(inputs, edit_decisions)
if not asset_manifest:
@@ -1336,26 +1380,6 @@ class VideoCompose(BaseTool):
# Also accept profile as "output_profile" (skill convention) or "profile"
profile = inputs.get("profile") or inputs.get("output_profile")
# --- Runtime routing: honor render_runtime locked at proposal ---
# Silent swaps are forbidden by governance. If the chosen runtime
# is unavailable, surface a structured blocker rather than quietly
# picking a different engine. Missing render_runtime is itself a
# governance violation — edit_decisions.schema.json requires it.
render_runtime = (edit_decisions.get("render_runtime") or "").strip().lower()
if not render_runtime:
return ToolResult(
success=False,
error=(
"render_runtime is not set in edit_decisions. Per governance, "
"it MUST be locked at proposal stage (proposal_packet."
"production_plan.render_runtime) and carried forward through "
"edit_decisions.render_runtime. Valid values: 'remotion', "
"'hyperframes', 'ffmpeg'. Re-run the proposal stage with an "
"explicit runtime choice — do NOT default this field."
),
)
if render_runtime == "hyperframes":
return self._render_via_hyperframes(
inputs=inputs,
@@ -1374,16 +1398,6 @@ class VideoCompose(BaseTool):
output_path=output_path,
profile=profile,
)
if render_runtime != "remotion":
return ToolResult(
success=False,
error=(
f"Unknown render_runtime {render_runtime!r}. "
f"Valid values: remotion, hyperframes, ffmpeg. "
f"render_runtime must be set at proposal stage."
),
)
# --- Explicit Remotion path (render_runtime == 'remotion') ---
if self._needs_remotion(resolved_cuts):
remotion_inputs: dict[str, Any] = {
@@ -1392,6 +1406,11 @@ class VideoCompose(BaseTool):
}
if profile:
remotion_inputs["profile"] = profile
# Forward the creator-facing render timeout through the high-level
# render path (execute(operation="render") -> _render), otherwise it
# would only take effect on a direct _remotion_render() call.
if inputs.get("remotion_timeout_ms") is not None:
remotion_inputs["remotion_timeout_ms"] = inputs["remotion_timeout_ms"]
render_result = self._remotion_render(remotion_inputs)
# Governance: NEVER silently fall back to FFmpeg when Remotion fails.
@@ -1738,12 +1757,45 @@ class VideoCompose(BaseTool):
except (ImportError, ValueError):
pass
# Optional creator-facing render timeout. Remotion's `--timeout` (ms)
# governs headless-browser setup and delayRender(); on slow machines or
# restricted networks the default 30s browser setup times out with an
# opaque failure. Pass it through and give the subprocess enough headroom
# so run_command() does not kill Remotion before its own timeout fires.
remotion_timeout_ms = inputs.get("remotion_timeout_ms")
subprocess_timeout = 600
if remotion_timeout_ms:
try:
ms = int(remotion_timeout_ms)
cmd.append(f"--timeout={ms}")
subprocess_timeout = max(subprocess_timeout, ms // 1000 + 60)
except (TypeError, ValueError):
pass
try:
# Invoke from inside the composer dir so npx can resolve the
# local remotion binary via node_modules/.bin. Without this,
# Windows npx cannot locate the CLI and returns "could not
# determine executable to run".
self.run_command(cmd, timeout=600, cwd=composer_dir)
self.run_command(cmd, timeout=subprocess_timeout, cwd=composer_dir)
except subprocess.CalledProcessError as e:
# run_command uses check=True + capture_output, so the useful
# Remotion diagnostics live in stderr/stdout — surface the tail
# instead of the bare "returned non-zero exit status 1".
detail = (e.stderr or e.stdout or "").strip()
tail = "\n".join(detail.splitlines()[-25:]) if detail else "(no output captured)"
return ToolResult(
success=False,
error=f"Remotion render failed (exit {e.returncode}):\n{tail}",
)
except subprocess.TimeoutExpired as e:
return ToolResult(
success=False,
error=(
f"Remotion render timed out after {e.timeout}s. If the headless "
"browser is slow to start, raise remotion_timeout_ms (ms)."
),
)
except Exception as e:
return ToolResult(success=False, error=f"Remotion render failed: {e}")
finally: