From 77e99e97392e96dc538cba20bc7408bee1d644f7 Mon Sep 17 00:00:00 2001 From: Toran Bruce Richards Date: Fri, 27 Jun 2025 07:39:24 +0100 Subject: [PATCH] feat(blocks): Add more Revid.ai media generation blocks (#9931) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit

Why these changes are needed 🧐

Revid.ai offers several specialised, undocumented rendering flows beyond the basic “text-to-video” endpoint our platform already supported. to:

  1. Generate ads from copy plus product images (30-second vertical spots).

  2. Turn a single creative prompt into a fully AI-generated video (no multi-line script).

  3. Transform a screenshot into a narrated, avatar-driven clip, ideal for product-led demos.

Without first-class blocks for these flows, users were forced to drop to raw HTTP nodes, losing schema validation, test mocks and credential management.

Changes 🏗️

Added new category to ``BlockCategory`` in ``block.py`` for ``MARKETING = "Block that helps with marketing"`` Area | Change | Notes -- | -- | -- ai_shortform_video_block.py | Refactored out a shared _RevidMixin (webhook + polling helpers). | Keeps DRY across new blocks.   | Added AudioTrack.DONT_STOP_ME_ABSTRACT_FUTURE_BASS and Voice.EVA enum members. | Required by Revid sample payloads.   | AIAdMakerVideoCreatorBlock | Implements ai-ad-generator flow; supports optional input_media_urls, target_duration, use_only_provided_media.   | AIPromptToVideoCreatorBlock | Implements prompt-to-video flow with prompt_target_duration.   | AIScreenshotToVideoAdBlock | Implements screenshot-to-video-ad flow (avatar narration, BG removal).   | Added full pydantic schemas, test stubs & mock hooks for each new block. | Ensures unit tests pass and blocks appear in UI.

No existing functionality was removed; current AIShortformVideoCreatorBlock is untouched apart from enum imports.

### Checklist 📋 #### For code changes: - [x] I have clearly listed my changes in the PR description - [x] I have made a test plan - [x] I have tested my changes according to the test plan: - [x] use the ``AI ShortForm Video Creator`` block to generate a video and it will work - [x] same with `` ai ad maker video creator`` block test it and it should work - [x] and test ``ai screenshot to video ad`` block it should work --------- Co-authored-by: Bently --- .../blocks/ai_shortform_video_block.py | 457 ++++++++++++++++-- .../backend/backend/data/block.py | 1 + 2 files changed, 415 insertions(+), 43 deletions(-) diff --git a/autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py b/autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py index c3c4e36472..fe13687970 100644 --- a/autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py +++ b/autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py @@ -53,6 +53,7 @@ class AudioTrack(str, Enum): REFRESHER = ("Refresher",) TOURIST = ("Tourist",) TWIN_TYCHES = ("Twin Tyches",) + DONT_STOP_ME_ABSTRACT_FUTURE_BASS = ("Dont Stop Me Abstract Future Bass",) @property def audio_url(self): @@ -78,6 +79,7 @@ class AudioTrack(str, Enum): AudioTrack.REFRESHER: "https://cdn.tfrv.xyz/audio/refresher.mp3", AudioTrack.TOURIST: "https://cdn.tfrv.xyz/audio/tourist.mp3", AudioTrack.TWIN_TYCHES: "https://cdn.tfrv.xyz/audio/twin-tynches.mp3", + AudioTrack.DONT_STOP_ME_ABSTRACT_FUTURE_BASS: "https://cdn.revid.ai/audio/_dont-stop-me-abstract-future-bass.mp3", } return audio_urls[self] @@ -105,6 +107,7 @@ class GenerationPreset(str, Enum): MOVIE = ("Movie",) STYLIZED_ILLUSTRATION = ("Stylized Illustration",) MANGA = ("Manga",) + DEFAULT = ("DEFAULT",) class Voice(str, Enum): @@ -114,6 +117,7 @@ class Voice(str, Enum): JESSICA = "Jessica" CHARLOTTE = "Charlotte" CALLUM = "Callum" + EVA = "Eva" @property def voice_id(self): @@ -124,6 +128,7 @@ class Voice(str, Enum): Voice.JESSICA: "cgSgspJ2msm6clMCkdW9", Voice.CHARLOTTE: "XB0fDUnXU5powFXDhCwa", Voice.CALLUM: "N2lVS1w4EtoT3dr4eOWO", + Voice.EVA: "FGY2WhTYpPnrIDTdsKH5", } return voice_id_map[self] @@ -141,6 +146,8 @@ logger = logging.getLogger(__name__) class AIShortformVideoCreatorBlock(Block): + """Creates a short‑form text‑to‑video clip using stock or AI imagery.""" + class Input(BlockSchema): credentials: CredentialsMetaInput[ Literal[ProviderName.REVID], Literal["api_key"] @@ -184,40 +191,8 @@ class AIShortformVideoCreatorBlock(Block): video_url: str = SchemaField(description="The URL of the created video") error: str = SchemaField(description="Error message if the request failed") - def __init__(self): - super().__init__( - id="361697fb-0c4f-4feb-aed3-8320c88c771b", - description="Creates a shortform video using revid.ai", - categories={BlockCategory.SOCIAL, BlockCategory.AI}, - input_schema=AIShortformVideoCreatorBlock.Input, - output_schema=AIShortformVideoCreatorBlock.Output, - test_input={ - "credentials": TEST_CREDENTIALS_INPUT, - "script": "[close-up of a cat] Meow!", - "ratio": "9 / 16", - "resolution": "720p", - "frame_rate": 60, - "generation_preset": GenerationPreset.LEONARDO, - "background_music": AudioTrack.HIGHWAY_NOCTURNE, - "voice": Voice.LILY, - "video_style": VisualMediaType.STOCK_VIDEOS, - }, - test_output=( - "video_url", - "https://example.com/video.mp4", - ), - test_mock={ - "create_webhook": lambda: ( - "test_uuid", - "https://webhook.site/test_uuid", - ), - "create_video": lambda api_key, payload: {"pid": "test_pid"}, - "wait_for_video": lambda api_key, pid, webhook_token, max_wait_time=1000: "https://example.com/video.mp4", - }, - test_credentials=TEST_CREDENTIALS, - ) - - async def create_webhook(self): + async def create_webhook(self) -> tuple[str, str]: + """Create a new webhook URL for receiving notifications.""" url = "https://webhook.site/token" headers = {"Accept": "application/json", "Content-Type": "application/json"} response = await Requests().post(url, headers=headers) @@ -225,6 +200,7 @@ class AIShortformVideoCreatorBlock(Block): return webhook_data["uuid"], f"https://webhook.site/{webhook_data['uuid']}" async def create_video(self, api_key: SecretStr, payload: dict) -> dict: + """Create a video using the Revid API.""" url = "https://www.revid.ai/api/public/v2/render" headers = {"key": api_key.get_secret_value()} response = await Requests().post(url, json=payload, headers=headers) @@ -234,6 +210,7 @@ class AIShortformVideoCreatorBlock(Block): return response.json() async def check_video_status(self, api_key: SecretStr, pid: str) -> dict: + """Check the status of a video creation job.""" url = f"https://www.revid.ai/api/public/v2/status?pid={pid}" headers = {"key": api_key.get_secret_value()} response = await Requests().get(url, headers=headers) @@ -243,9 +220,9 @@ class AIShortformVideoCreatorBlock(Block): self, api_key: SecretStr, pid: str, - webhook_token: str, max_wait_time: int = 1000, ) -> str: + """Wait for video creation to complete and return the video URL.""" start_time = time.time() while time.time() - start_time < max_wait_time: status = await self.check_video_status(api_key, pid) @@ -266,6 +243,40 @@ class AIShortformVideoCreatorBlock(Block): logger.error("Video creation timed out") raise TimeoutError("Video creation timed out") + def __init__(self): + super().__init__( + id="361697fb-0c4f-4feb-aed3-8320c88c771b", + description="Creates a shortform video using revid.ai", + categories={BlockCategory.SOCIAL, BlockCategory.AI}, + input_schema=AIShortformVideoCreatorBlock.Input, + output_schema=AIShortformVideoCreatorBlock.Output, + test_input={ + "credentials": TEST_CREDENTIALS_INPUT, + "script": "[close-up of a cat] Meow!", + "ratio": "9 / 16", + "resolution": "720p", + "frame_rate": 60, + "generation_preset": GenerationPreset.LEONARDO, + "background_music": AudioTrack.HIGHWAY_NOCTURNE, + "voice": Voice.LILY, + "video_style": VisualMediaType.STOCK_VIDEOS, + }, + test_output=("video_url", "https://example.com/video.mp4"), + test_mock={ + "create_webhook": lambda *args, **kwargs: ( + "test_uuid", + "https://webhook.site/test_uuid", + ), + "create_video": lambda *args, **kwargs: {"pid": "test_pid"}, + "check_video_status": lambda *args, **kwargs: { + "status": "ready", + "videoUrl": "https://example.com/video.mp4", + }, + "wait_for_video": lambda *args, **kwargs: "https://example.com/video.mp4", + }, + test_credentials=TEST_CREDENTIALS, + ) + async def run( self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs ) -> BlockOutput: @@ -273,20 +284,18 @@ class AIShortformVideoCreatorBlock(Block): webhook_token, webhook_url = await self.create_webhook() logger.debug(f"Webhook URL: {webhook_url}") - audio_url = input_data.background_music.audio_url - payload = { "frameRate": input_data.frame_rate, "resolution": input_data.resolution, "frameDurationMultiplier": 18, - "webhook": webhook_url, + "webhook": None, "creationParams": { "mediaType": input_data.video_style, "captionPresetName": "Wrap 1", "selectedVoice": input_data.voice.voice_id, "hasEnhancedGeneration": True, "generationPreset": input_data.generation_preset.name, - "selectedAudio": input_data.background_music, + "selectedAudio": input_data.background_music.value, "origin": "/create", "inputText": input_data.script, "flowType": "text-to-video", @@ -302,7 +311,7 @@ class AIShortformVideoCreatorBlock(Block): "selectedStoryStyle": {"value": "custom", "label": "Custom"}, "hasToGenerateVideos": input_data.video_style != VisualMediaType.STOCK_VIDEOS, - "audioUrl": audio_url, + "audioUrl": input_data.background_music.audio_url, }, } @@ -319,8 +328,370 @@ class AIShortformVideoCreatorBlock(Block): logger.debug( f"Video created with project ID: {pid}. Waiting for completion..." ) - video_url = await self.wait_for_video( - credentials.api_key, pid, webhook_token - ) + video_url = await self.wait_for_video(credentials.api_key, pid) logger.debug(f"Video ready: {video_url}") yield "video_url", video_url + + +class AIAdMakerVideoCreatorBlock(Block): + """Generates a 30‑second vertical AI advert using optional user‑supplied imagery.""" + + class Input(BlockSchema): + credentials: CredentialsMetaInput[ + Literal[ProviderName.REVID], Literal["api_key"] + ] = CredentialsField( + description="Credentials for Revid.ai API access.", + ) + script: str = SchemaField( + description="Short advertising copy. Line breaks create new scenes.", + placeholder="Introducing Foobar – [show product photo] the gadget that does it all.", + ) + ratio: str = SchemaField(description="Aspect ratio", default="9 / 16") + target_duration: int = SchemaField( + description="Desired length of the ad in seconds.", default=30 + ) + voice: Voice = SchemaField( + description="Narration voice", default=Voice.EVA, placeholder=Voice.EVA + ) + background_music: AudioTrack = SchemaField( + description="Background track", + default=AudioTrack.DONT_STOP_ME_ABSTRACT_FUTURE_BASS, + ) + input_media_urls: list[str] = SchemaField( + description="List of image URLs to feature in the advert.", default=[] + ) + use_only_provided_media: bool = SchemaField( + description="Restrict visuals to supplied images only.", default=True + ) + + class Output(BlockSchema): + video_url: str = SchemaField(description="URL of the finished advert") + error: str = SchemaField(description="Error message on failure") + + async def create_webhook(self) -> tuple[str, str]: + """Create a new webhook URL for receiving notifications.""" + url = "https://webhook.site/token" + headers = {"Accept": "application/json", "Content-Type": "application/json"} + response = await Requests().post(url, headers=headers) + webhook_data = response.json() + return webhook_data["uuid"], f"https://webhook.site/{webhook_data['uuid']}" + + async def create_video(self, api_key: SecretStr, payload: dict) -> dict: + """Create a video using the Revid API.""" + url = "https://www.revid.ai/api/public/v2/render" + headers = {"key": api_key.get_secret_value()} + response = await Requests().post(url, json=payload, headers=headers) + logger.debug( + f"API Response Status Code: {response.status}, Content: {response.text}" + ) + return response.json() + + async def check_video_status(self, api_key: SecretStr, pid: str) -> dict: + """Check the status of a video creation job.""" + url = f"https://www.revid.ai/api/public/v2/status?pid={pid}" + headers = {"key": api_key.get_secret_value()} + response = await Requests().get(url, headers=headers) + return response.json() + + async def wait_for_video( + self, + api_key: SecretStr, + pid: str, + max_wait_time: int = 1000, + ) -> str: + """Wait for video creation to complete and return the video URL.""" + start_time = time.time() + while time.time() - start_time < max_wait_time: + status = await self.check_video_status(api_key, pid) + logger.debug(f"Video status: {status}") + + if status.get("status") == "ready" and "videoUrl" in status: + return status["videoUrl"] + elif status.get("status") == "error": + error_message = status.get("error", "Unknown error occurred") + logger.error(f"Video creation failed: {error_message}") + raise ValueError(f"Video creation failed: {error_message}") + elif status.get("status") in ["FAILED", "CANCELED"]: + logger.error(f"Video creation failed: {status.get('message')}") + raise ValueError(f"Video creation failed: {status.get('message')}") + + await asyncio.sleep(10) + + logger.error("Video creation timed out") + raise TimeoutError("Video creation timed out") + + def __init__(self): + super().__init__( + id="58bd2a19-115d-4fd1-8ca4-13b9e37fa6a0", + description="Creates an AI‑generated 30‑second advert (text + images)", + categories={BlockCategory.MARKETING, BlockCategory.AI}, + input_schema=AIAdMakerVideoCreatorBlock.Input, + output_schema=AIAdMakerVideoCreatorBlock.Output, + test_input={ + "credentials": TEST_CREDENTIALS_INPUT, + "script": "Test product launch!", + "input_media_urls": [ + "https://cdn.revid.ai/uploads/1747076315114-image.png", + ], + }, + test_output=("video_url", "https://example.com/ad.mp4"), + test_mock={ + "create_webhook": lambda *args, **kwargs: ( + "test_uuid", + "https://webhook.site/test_uuid", + ), + "create_video": lambda *args, **kwargs: {"pid": "test_pid"}, + "check_video_status": lambda *args, **kwargs: { + "status": "ready", + "videoUrl": "https://example.com/ad.mp4", + }, + "wait_for_video": lambda *args, **kwargs: "https://example.com/ad.mp4", + }, + test_credentials=TEST_CREDENTIALS, + ) + + async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs): + webhook_token, webhook_url = await self.create_webhook() + + payload = { + "webhook": webhook_url, + "creationParams": { + "targetDuration": input_data.target_duration, + "ratio": input_data.ratio, + "mediaType": "aiVideo", + "inputText": input_data.script, + "flowType": "text-to-video", + "slug": "ai-ad-generator", + "slugNew": "", + "isCopiedFrom": False, + "hasToGenerateVoice": True, + "hasToTranscript": False, + "hasToSearchMedia": True, + "hasAvatar": False, + "hasWebsiteRecorder": False, + "hasTextSmallAtBottom": False, + "selectedAudio": input_data.background_music.value, + "selectedVoice": input_data.voice.voice_id, + "selectedAvatar": "https://cdn.revid.ai/avatars/young-woman.mp4", + "selectedAvatarType": "video/mp4", + "websiteToRecord": "", + "hasToGenerateCover": True, + "nbGenerations": 1, + "disableCaptions": False, + "mediaMultiplier": "medium", + "characters": [], + "captionPresetName": "Revid", + "sourceType": "contentScraping", + "selectedStoryStyle": {"value": "custom", "label": "General"}, + "generationPreset": "DEFAULT", + "hasToGenerateMusic": False, + "isOptimizedForChinese": False, + "generationUserPrompt": "", + "enableNsfwFilter": False, + "addStickers": False, + "typeMovingImageAnim": "dynamic", + "hasToGenerateSoundEffects": False, + "forceModelType": "gpt-image-1", + "selectedCharacters": [], + "lang": "", + "voiceSpeed": 1, + "disableAudio": False, + "disableVoice": False, + "useOnlyProvidedMedia": input_data.use_only_provided_media, + "imageGenerationModel": "ultra", + "videoGenerationModel": "pro", + "hasEnhancedGeneration": True, + "hasEnhancedGenerationPro": True, + "inputMedias": [ + {"url": url, "title": "", "type": "image"} + for url in input_data.input_media_urls + ], + "hasToGenerateVideos": True, + "audioUrl": input_data.background_music.audio_url, + "watermark": None, + }, + } + + response = await self.create_video(credentials.api_key, payload) + pid = response.get("pid") + if not pid: + raise RuntimeError("Failed to create video: No project ID returned") + + video_url = await self.wait_for_video(credentials.api_key, pid) + yield "video_url", video_url + + +class AIScreenshotToVideoAdBlock(Block): + """Creates an advert where the supplied screenshot is narrated by an AI avatar.""" + + class Input(BlockSchema): + credentials: CredentialsMetaInput[ + Literal[ProviderName.REVID], Literal["api_key"] + ] = CredentialsField(description="Revid.ai API key") + script: str = SchemaField( + description="Narration that will accompany the screenshot.", + placeholder="Check out these amazing stats!", + ) + screenshot_url: str = SchemaField( + description="Screenshot or image URL to showcase." + ) + ratio: str = SchemaField(default="9 / 16") + target_duration: int = SchemaField(default=30) + voice: Voice = SchemaField(default=Voice.EVA) + background_music: AudioTrack = SchemaField( + default=AudioTrack.DONT_STOP_ME_ABSTRACT_FUTURE_BASS + ) + + class Output(BlockSchema): + video_url: str = SchemaField(description="Rendered video URL") + error: str = SchemaField(description="Error, if encountered") + + async def create_webhook(self) -> tuple[str, str]: + """Create a new webhook URL for receiving notifications.""" + url = "https://webhook.site/token" + headers = {"Accept": "application/json", "Content-Type": "application/json"} + response = await Requests().post(url, headers=headers) + webhook_data = response.json() + return webhook_data["uuid"], f"https://webhook.site/{webhook_data['uuid']}" + + async def create_video(self, api_key: SecretStr, payload: dict) -> dict: + """Create a video using the Revid API.""" + url = "https://www.revid.ai/api/public/v2/render" + headers = {"key": api_key.get_secret_value()} + response = await Requests().post(url, json=payload, headers=headers) + logger.debug( + f"API Response Status Code: {response.status}, Content: {response.text}" + ) + return response.json() + + async def check_video_status(self, api_key: SecretStr, pid: str) -> dict: + """Check the status of a video creation job.""" + url = f"https://www.revid.ai/api/public/v2/status?pid={pid}" + headers = {"key": api_key.get_secret_value()} + response = await Requests().get(url, headers=headers) + return response.json() + + async def wait_for_video( + self, + api_key: SecretStr, + pid: str, + max_wait_time: int = 1000, + ) -> str: + """Wait for video creation to complete and return the video URL.""" + start_time = time.time() + while time.time() - start_time < max_wait_time: + status = await self.check_video_status(api_key, pid) + logger.debug(f"Video status: {status}") + + if status.get("status") == "ready" and "videoUrl" in status: + return status["videoUrl"] + elif status.get("status") == "error": + error_message = status.get("error", "Unknown error occurred") + logger.error(f"Video creation failed: {error_message}") + raise ValueError(f"Video creation failed: {error_message}") + elif status.get("status") in ["FAILED", "CANCELED"]: + logger.error(f"Video creation failed: {status.get('message')}") + raise ValueError(f"Video creation failed: {status.get('message')}") + + await asyncio.sleep(10) + + logger.error("Video creation timed out") + raise TimeoutError("Video creation timed out") + + def __init__(self): + super().__init__( + id="0f3e4635-e810-43d9-9e81-49e6f4e83b7c", + description="Turns a screenshot into an engaging, avatar‑narrated video advert.", + categories={BlockCategory.AI, BlockCategory.MARKETING}, + input_schema=AIScreenshotToVideoAdBlock.Input, + output_schema=AIScreenshotToVideoAdBlock.Output, + test_input={ + "credentials": TEST_CREDENTIALS_INPUT, + "script": "Amazing numbers!", + "screenshot_url": "https://cdn.revid.ai/uploads/1747080376028-image.png", + }, + test_output=("video_url", "https://example.com/screenshot.mp4"), + test_mock={ + "create_webhook": lambda *args, **kwargs: ( + "test_uuid", + "https://webhook.site/test_uuid", + ), + "create_video": lambda *args, **kwargs: {"pid": "test_pid"}, + "check_video_status": lambda *args, **kwargs: { + "status": "ready", + "videoUrl": "https://example.com/screenshot.mp4", + }, + "wait_for_video": lambda *args, **kwargs: "https://example.com/screenshot.mp4", + }, + test_credentials=TEST_CREDENTIALS, + ) + + async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs): + webhook_token, webhook_url = await self.create_webhook() + + payload = { + "webhook": webhook_url, + "creationParams": { + "targetDuration": input_data.target_duration, + "ratio": input_data.ratio, + "mediaType": "aiVideo", + "hasAvatar": True, + "removeAvatarBackground": True, + "inputText": input_data.script, + "flowType": "text-to-video", + "slug": "ai-ad-generator", + "slugNew": "screenshot-to-video-ad", + "isCopiedFrom": "ai-ad-generator", + "hasToGenerateVoice": True, + "hasToTranscript": False, + "hasToSearchMedia": True, + "hasWebsiteRecorder": False, + "hasTextSmallAtBottom": False, + "selectedAudio": input_data.background_music.value, + "selectedVoice": input_data.voice.voice_id, + "selectedAvatar": "https://cdn.revid.ai/avatars/young-woman.mp4", + "selectedAvatarType": "video/mp4", + "websiteToRecord": "", + "hasToGenerateCover": True, + "nbGenerations": 1, + "disableCaptions": False, + "mediaMultiplier": "medium", + "characters": [], + "captionPresetName": "Revid", + "sourceType": "contentScraping", + "selectedStoryStyle": {"value": "custom", "label": "General"}, + "generationPreset": "DEFAULT", + "hasToGenerateMusic": False, + "isOptimizedForChinese": False, + "generationUserPrompt": "", + "enableNsfwFilter": False, + "addStickers": False, + "typeMovingImageAnim": "dynamic", + "hasToGenerateSoundEffects": False, + "forceModelType": "gpt-image-1", + "selectedCharacters": [], + "lang": "", + "voiceSpeed": 1, + "disableAudio": False, + "disableVoice": False, + "useOnlyProvidedMedia": True, + "imageGenerationModel": "ultra", + "videoGenerationModel": "ultra", + "hasEnhancedGeneration": True, + "hasEnhancedGenerationPro": True, + "inputMedias": [ + {"url": input_data.screenshot_url, "title": "", "type": "image"} + ], + "hasToGenerateVideos": True, + "audioUrl": input_data.background_music.audio_url, + "watermark": None, + }, + } + + response = await self.create_video(credentials.api_key, payload) + pid = response.get("pid") + if not pid: + raise RuntimeError("Failed to create video: No project ID returned") + + video_url = await self.wait_for_video(credentials.api_key, pid) + yield "video_url", video_url diff --git a/autogpt_platform/backend/backend/data/block.py b/autogpt_platform/backend/backend/data/block.py index f26788e991..f5e9e8b1cd 100644 --- a/autogpt_platform/backend/backend/data/block.py +++ b/autogpt_platform/backend/backend/data/block.py @@ -78,6 +78,7 @@ class BlockCategory(Enum): PRODUCTIVITY = "Block that helps with productivity" ISSUE_TRACKING = "Block that helps with issue tracking" MULTIMEDIA = "Block that interacts with multimedia content" + MARKETING = "Block that helps with marketing" def dict(self) -> dict[str, str]: return {"category": self.name, "description": self.value}