| from __future__ import annotations |
|
|
| from pathlib import Path |
| from typing import Any |
|
|
| from renderer.core.config import Settings |
| from renderer.core.models import TaskResult |
| from renderer.core.utils import now, safe_filename, write_json |
| from renderer.studio.capabilities import ( |
| AI_ASSISTANTS, |
| AI_EDITING_FEATURES, |
| IMAGE_GENERATION_PROVIDERS, |
| MUSIC_PROVIDERS, |
| VIDEO_GENERATION_PROVIDERS, |
| VOICE_PROVIDERS, |
| ) |
|
|
|
|
| class StudioTaskProcessor: |
| """Manifest-producing async handlers for optional AI providers and studio automation.""" |
|
|
| def __init__(self, settings: Settings | None = None, log=None) -> None: |
| self.settings = settings or Settings() |
| self.settings.ensure_dirs() |
| self._logs: list[str] = [] |
| self._log = log |
|
|
| def caption_generate(self, payload: dict[str, Any], job_id: str) -> TaskResult: |
| text = str(payload.get("text") or payload.get("transcript") or "") |
| captions = payload.get("events") if isinstance(payload.get("events"), list) else _captions_from_text(text) |
| manifest = { |
| "type": "caption_generation", |
| "status": "ready", |
| "engine": payload.get("engine", "whisper"), |
| "media": payload.get("media") or payload.get("audio"), |
| "template": payload.get("template", "capcut"), |
| "language": payload.get("language"), |
| "features": { |
| "word_timestamps": bool(payload.get("word_timestamps", True)), |
| "sentence_timestamps": True, |
| "emoji_insertion": bool(payload.get("emoji_insertion", False)), |
| "speaker_detection": bool(payload.get("speaker_detection", False)), |
| "karaoke": bool(payload.get("karaoke", True)), |
| "animated": bool(payload.get("animated", True)), |
| }, |
| "captions": captions, |
| } |
| output = self._json_artifact(job_id, "captions", manifest) |
| return self._result(output, {"task": "caption_generate", "caption_count": len(captions)}) |
|
|
| def music_generate(self, payload: dict[str, Any], job_id: str) -> TaskResult: |
| provider = _provider(payload.get("provider"), MUSIC_PROVIDERS, "musicgen") |
| prompt = str(payload.get("prompt") or payload.get("style") or "background music") |
| duration = float(payload.get("duration", 30)) |
| manifest = { |
| "type": "music_generation", |
| "status": "provider_required", |
| "provider": provider, |
| "prompt": prompt, |
| "style": payload.get("style", "background_music"), |
| "duration": duration, |
| "bpm": payload.get("bpm"), |
| "license": payload.get("license", "user_configured"), |
| "next_step": "Configure provider credentials or connect this manifest to a local MusicGen runner.", |
| } |
| output = self._json_artifact(job_id, "music_request", manifest) |
| return self._result(output, {"task": "music_generate", "provider": provider, "duration": duration}) |
|
|
| def voice_generate(self, payload: dict[str, Any], job_id: str) -> TaskResult: |
| provider = _provider(payload.get("provider"), VOICE_PROVIDERS, "kokoro") |
| text = str(payload.get("text") or "") |
| manifest = { |
| "type": "voice_generation", |
| "status": "provider_required", |
| "provider": provider, |
| "text": text, |
| "voice": payload.get("voice", "default"), |
| "emotion": payload.get("emotion"), |
| "speed": float(payload.get("speed", 1.0)), |
| "pitch": float(payload.get("pitch", 1.0)), |
| "clone_reference": payload.get("clone_reference"), |
| "multi_speaker": payload.get("speakers", []), |
| "next_step": "Configure the selected TTS backend to render audio for this manifest.", |
| } |
| output = self._json_artifact(job_id, "voice_request", manifest) |
| return self._result(output, {"task": "voice_generate", "provider": provider, "characters": len(text)}) |
|
|
| def image_generate(self, payload: dict[str, Any], job_id: str) -> TaskResult: |
| provider = _provider(payload.get("provider"), IMAGE_GENERATION_PROVIDERS, "flux") |
| manifest = { |
| "type": "image_generation", |
| "status": "provider_required", |
| "provider": provider, |
| "prompt": payload.get("prompt", ""), |
| "negative_prompt": payload.get("negative_prompt", ""), |
| "mode": payload.get("mode", "text_to_image"), |
| "control_image": payload.get("control_image"), |
| "source_image": payload.get("source_image"), |
| "size": payload.get("size", "1024x1024"), |
| "features": { |
| "background_replacement": bool(payload.get("background_replacement", False)), |
| "object_removal": bool(payload.get("object_removal", False)), |
| "upscaling": bool(payload.get("upscaling", False)), |
| }, |
| } |
| output = self._json_artifact(job_id, "image_request", manifest) |
| return self._result(output, {"task": "image_generate", "provider": provider}) |
|
|
| def video_generate(self, payload: dict[str, Any], job_id: str) -> TaskResult: |
| provider = _provider(payload.get("provider"), VIDEO_GENERATION_PROVIDERS, "ltx_video") |
| manifest = { |
| "type": "video_generation", |
| "status": "provider_required", |
| "provider": provider, |
| "prompt": payload.get("prompt", ""), |
| "mode": payload.get("mode", "text_to_video"), |
| "image": payload.get("image"), |
| "duration": float(payload.get("duration", 5)), |
| "fps": int(payload.get("fps", 24)), |
| "size": payload.get("size", "1280x720"), |
| "next_step": "Connect Wan, LTX Video, Hunyuan Video, or Veo credentials/runtime to execute this request.", |
| } |
| output = self._json_artifact(job_id, "video_request", manifest) |
| return self._result(output, {"task": "video_generate", "provider": provider}) |
|
|
| def ai_tool(self, tool: str, payload: dict[str, Any], job_id: str) -> TaskResult: |
| tool = _canonical(tool) |
| if tool not in AI_EDITING_FEATURES: |
| raise ValueError(f"Unsupported AI editing tool: {tool}") |
| manifest = { |
| "type": "ai_editing", |
| "tool": tool, |
| "status": "ready", |
| "media": payload.get("media"), |
| "project_id": payload.get("project_id"), |
| "platform": payload.get("platform", "tiktok"), |
| "result": _ai_result(tool, payload), |
| "created_at": now(), |
| } |
| output = self._json_artifact(job_id, tool, manifest) |
| return self._result(output, {"task": tool}) |
|
|
| def assistant_tool(self, tool: str, payload: dict[str, Any], job_id: str) -> TaskResult: |
| tool = _canonical(tool) |
| if tool not in AI_ASSISTANTS: |
| raise ValueError(f"Unsupported assistant: {tool}") |
| text = str(payload.get("topic") or payload.get("transcript") or payload.get("prompt") or "") |
| manifest = { |
| "type": "assistant", |
| "tool": tool, |
| "status": "ready", |
| "input": text, |
| "platform": payload.get("platform", "general"), |
| "result": _assistant_result(tool, text, payload), |
| "created_at": now(), |
| } |
| output = self._json_artifact(job_id, tool, manifest) |
| return self._result(output, {"task": tool, "characters": len(text)}) |
|
|
| def _json_artifact(self, job_id: str, name: str, payload: dict[str, Any]) -> Path: |
| output = self.settings.exports_dir / f"{job_id}_{safe_filename(name)}.json" |
| write_json(output, payload) |
| self._message(f"Wrote {name} manifest") |
| return output |
|
|
| def _result(self, output: Path, metrics: dict[str, Any]) -> TaskResult: |
| return TaskResult(output_path=output, commands=[], metrics=metrics, logs=list(self._logs)) |
|
|
| def _message(self, message: str) -> None: |
| self._logs.append(message) |
| if self._log: |
| self._log(message) |
|
|
|
|
| def _provider(value: Any, supported: list[str], default: str) -> str: |
| provider = _canonical(str(value or default)) |
| return provider if provider in supported else default |
|
|
|
|
| def _canonical(value: str) -> str: |
| return value.strip().lower().replace("-", "_").replace(" ", "_") |
|
|
|
|
| def _captions_from_text(text: str) -> list[dict[str, Any]]: |
| if not text: |
| return [] |
| words = text.split() |
| chunks: list[list[str]] = [] |
| while words: |
| chunks.append(words[:8]) |
| words = words[8:] |
| captions = [] |
| cursor = 0.0 |
| for chunk in chunks: |
| duration = max(1.2, len(chunk) * 0.34) |
| captions.append({"start": round(cursor, 2), "end": round(cursor + duration, 2), "text": " ".join(chunk)}) |
| cursor += duration |
| return captions |
|
|
|
|
| def _ai_result(tool: str, payload: dict[str, Any]) -> dict[str, Any]: |
| platform = str(payload.get("platform") or "tiktok") |
| if tool == "auto_highlight_detection": |
| return {"highlights": [{"start": 0, "end": 8, "reason": "opening hook"}]} |
| if tool == "auto_scene_detection": |
| return {"scenes": [{"start": 0, "end": 5, "label": "intro"}, {"start": 5, "end": 12, "label": "body"}]} |
| if tool in {"auto_reframe", "auto_crop", "auto_platform_optimization"}: |
| return {"platform": platform, "safe_zone": "vertical_center", "aspect_ratio": "9:16"} |
| if tool == "auto_viral_score": |
| return {"score": 74, "signals": ["short duration", "caption-ready", platform]} |
| if tool == "auto_hook_detection": |
| return {"hook": str(payload.get("transcript") or payload.get("text") or "")[:120], "score": 68} |
| if tool == "auto_thumbnail_selection": |
| return {"frames": [{"timestamp": 2.0, "score": 82}, {"timestamp": 6.5, "score": 75}]} |
| return {"plan": f"{tool} plan generated", "confidence": "heuristic", "platform": platform} |
|
|
|
|
| def _assistant_result(tool: str, text: str, payload: dict[str, Any]) -> dict[str, Any]: |
| subject = text.strip() or "your video" |
| short = " ".join(subject.split()[:12]) |
| if tool == "script_writer": |
| return {"script": f"Hook: {short}\nValue: show the clearest proof.\nCTA: invite viewers to take the next step."} |
| if tool == "hook_generator": |
| return {"hooks": [f"Stop scrolling if you care about {short}", f"Nobody tells you this about {short}"]} |
| if tool == "title_generator": |
| return {"titles": [short.title(), f"How {short.title()} Changes Everything"]} |
| if tool == "description_generator": |
| return {"description": f"{subject}\n\nBuilt with Ava2lon Studio AI."} |
| if tool == "hashtag_generator": |
| tags = [word.strip(".,!?").lower() for word in subject.split() if len(word.strip(".,!?")) > 3] |
| return {"hashtags": ["#" + tag for tag in tags[:8]] or ["#video", "#creator"]} |
| if tool == "storyboard_generator": |
| return {"beats": [{"scene": 1, "goal": "hook"}, {"scene": 2, "goal": "proof"}, {"scene": 3, "goal": "CTA"}]} |
| if tool == "b_roll_planner": |
| return {"shots": [{"type": "close_up", "description": short}, {"type": "screen_recording", "description": "show the result"}]} |
| if tool == "thumbnail_prompt_generator": |
| return {"prompt": f"High contrast thumbnail for {short}, expressive face, bold text, clean background"} |
| if tool == "seo_optimizer": |
| return {"keywords": [word.strip(".,!?").lower() for word in subject.split()[:10]], "score": 72} |
| return {"result": subject, "options": payload} |
|
|
