Text To Video

Text To Video

Identity

Reference System Usage

You must ground your responses in the provided reference files, treating them as the source of truth for this domain:

• For Creation: Always consult `references/patterns.md`. This file dictates how things should be built. Ignore generic approaches if a specific pattern exists here.
• For Diagnosis: Always consult `references/sharp_edges.md`. This file lists the critical failures and "why" they happen. Use it to explain risks to the user.
• For Review: Always consult `references/validations.md`. This contains the strict rules and constraints. Use it to validate user inputs objectively.

Note: If a user's request conflicts with the guidance in these files, politely correct them using the information provided in the references.

Text to Video

Patterns

---

Id

replicate-text-to-video

Name

Text-to-Video with Replicate

Description

Generate videos from text prompts using Replicate models.

Model recommendations:

• Wan 2.2: Open-source, fast, good quality
• Kling 2.1: High quality, character consistency
• minimax/hailuo: Physics realism

Videos typically take 30-150 seconds to generate.

Code Example

import replicate import asyncio from typing import Optional

Initialize client

client = replicate.Client(api_token=os.environ["REPLICATE_API_TOKEN"])

Text-to-Video with Wan 2.2 (fast, open-source)

def generate_video_wan( prompt: str, resolution: str = "480p", # 480p, 720p duration: int = 5, # seconds ) -> str: """Generate video from text with Wan 2.2."""

model = f"wavespeedai/wan-2.1-t2v-{resolution}"

output = client.run( model, input={ "prompt": prompt, "num_frames": duration * 16, # ~16 FPS "guidance_scale": 6.0, "num_inference_steps": 30, } )

return output # Video URL

Text-to-Video with Kling (high quality)

def generate_video_kling( prompt: str, duration: int = 5, # 5 or 10 seconds resolution: str = "720p", ) -> str: """Generate video with Kling v2.1."""

output = client.run( "kwaivgi/kling-v2.1", input={ "prompt": prompt, "duration": duration, "aspect_ratio": "16:9", "negative_prompt": "blurry, distorted, low quality", } )

return output

Async generation with polling

async def generate_video_async( prompt: str, model: str = "wavespeedai/wan-2.1-t2v-480p", timeout: int = 300, ) -> str: """Generate video asynchronously with status polling."""

prediction = client.predictions.create( model=model, input={"prompt": prompt} )

start = asyncio.get_event_loop().time()

while True: prediction = client.predictions.get(prediction.id)

if prediction.status == "succeeded": return prediction.output

if prediction.status == "failed": raise Exception(f"Generation failed: {prediction.error}")

if asyncio.get_event_loop().time() - start > timeout: client.predictions.cancel(prediction.id) raise TimeoutError("Video generation timed out")

await asyncio.sleep(5) # Poll every 5 seconds

Batch generation

async def generate_batch(prompts: list[str]) -> list[str]: """Generate multiple videos concurrently."""

tasks = [ generate_video_async(prompt) for prompt in prompts ]

return await asyncio.gather(*tasks)

Anti Patterns

---

Pattern

Blocking synchronous generation

Why

Videos take 30-150 seconds to generate

Fix

Use async with polling or webhooks

---

Pattern

No timeout on generation

Why

Failed generations can hang forever

Fix

Add timeout and cancellation logic

References

• https://replicate.com/collections/text-to-video
• https://replicate.com/blog/wan-21-generate-videos-with-an-api

---

Id

image-to-video

Name

Image-to-Video Animation

Description

Animate still images into videos. Preserve image content while adding motion.

Best for:

• Product animations
• Character motion
• Scene transitions
• Social media content

Code Example

import replicate import fal_client

Image-to-Video with Wan (Replicate)

def animate_image_wan( image_url: str, prompt: str, duration: int = 5, ) -> str: """Animate image with Wan 2.2."""

client = replicate.Client()

output = client.run( "wavespeedai/wan-2.1-i2v-480p", input={ "image": image_url, "prompt": prompt, "num_frames": duration * 16, "guidance_scale": 6.0, } )

return output

Image-to-Video with Kling (high quality)

def animate_image_kling( image_url: str, prompt: str, duration: int = 5, # 5 or 10 seconds motion_amount: float = 0.5, # 0-1 ) -> str: """Animate image with Kling v2.1."""

client = replicate.Client()

output = client.run( "kwaivgi/kling-v2.1", input={ "image": image_url, "prompt": prompt, "duration": duration, "cfg_scale": 0.5, "negative_prompt": "static, frozen, no motion", } )

return output

Image-to-Video with Fal.ai

def animate_image_fal( image_url: str, prompt: str, ) -> dict: """Animate image with Fal.ai."""

result = fal_client.submit( "fal-ai/wan-i2v", arguments={ "image_url": image_url, "prompt": prompt, "num_inference_steps": 30, } )

return result.get()

Luma Dream Machine style animation

def animate_with_camera_motion( image_url: str, camera_motion: str = "zoom_in", # zoom_in, zoom_out, pan_left, pan_right, orbit prompt: str = "", ) -> str: """Animate with specific camera motion."""

Map camera motion to prompt additions

MOTION_PROMPTS = { "zoom_in": "smooth zoom in, camera pushing forward", "zoom_out": "smooth zoom out, camera pulling back", "pan_left": "smooth pan left, camera sliding left", "pan_right": "smooth pan right, camera sliding right", "orbit": "camera orbiting around subject, 3D parallax", "tilt_up": "camera tilting upward, looking up", "tilt_down": "camera tilting downward, looking down", }

motion_prompt = MOTION_PROMPTS.get(camera_motion, "") full_prompt = f"{prompt}, {motion_prompt}".strip(", ")

client = replicate.Client() output = client.run( "wavespeedai/wan-2.1-i2v-720p", input={ "image": image_url, "prompt": full_prompt, "guidance_scale": 7.0, } )

return output

Product turntable animation

def create_product_turntable( product_image_url: str, rotation_degrees: int = 360, duration: int = 5, ) -> str: """Create 360-degree product rotation video."""

prompt = f"product rotating {rotation_degrees} degrees on turntable, smooth continuous rotation, studio lighting, white background"

return animate_image_kling( image_url=product_image_url, prompt=prompt, duration=duration, motion_amount=0.7, )

Anti Patterns

---

Pattern

Vague motion prompts

Why

Results in random or no motion

Fix

Be specific about motion type and direction

---

Pattern

Expecting perfect character consistency

Why

Current models drift across frames

Fix

Use shorter clips, or models with element/character features

References

• https://replicate.com/collections/image-to-video
• https://lumalabs.ai/dream-machine

---

Id

video-prompting

Name

Video Prompting Best Practices

Description

Write effective prompts for video generation. Key elements: subject, action, camera, lighting, style.

Video prompts differ from image prompts:

• Emphasize motion and action
• Describe camera movement
• Specify timing and pacing

Code Example

Video prompt structure

class VideoPrompt: """Build structured video prompts."""

def __init__(self): self.subject = "" self.action = "" self.camera = "" self.lighting = "" self.style = "" self.negative = ""

def with_subject(self, subject: str) -> "VideoPrompt": self.subject = subject return self

def with_action(self, action: str) -> "VideoPrompt": self.action = action return self

def with_camera(self, camera: str) -> "VideoPrompt": self.camera = camera return self

def with_lighting(self, lighting: str) -> "VideoPrompt": self.lighting = lighting return self

def with_style(self, style: str) -> "VideoPrompt": self.style = style return self

def build(self) -> str: parts = [ self.subject, self.action, self.camera, self.lighting, self.style, ] return ", ".join(p for p in parts if p)

Usage examples

prompt = ( VideoPrompt() .with_subject("a young woman in a red dress") .with_action("walking confidently through a busy city street") .with_camera("tracking shot, following from the side") .with_lighting("golden hour sunlight, dramatic shadows") .with_style("cinematic, film grain, shallow depth of field") .build() )

Camera motion vocabulary

CAMERA_MOTIONS = { "static": "locked off shot, stationary camera", "pan": "horizontal pan, smooth lateral movement", "tilt": "vertical tilt, camera looking up/down", "zoom": "slow zoom in, pushing forward", "dolly": "dolly shot, camera moving forward", "tracking": "tracking shot, following subject", "orbit": "orbital shot, camera circling subject", "crane": "crane shot, camera rising upward", "handheld": "handheld camera, slight shake", "steadicam": "steadicam shot, smooth floating movement", }

Action speed modifiers

SPEED_MODIFIERS = { "slow": "slow motion, graceful movement, 0.5x speed", "normal": "natural pace, realistic timing", "fast": "quick movement, energetic, rapid motion", "timelapse": "time lapse, accelerated, fast forward", }

Prompt templates by genre

GENRE_TEMPLATES = { "commercial": "{subject}, {action}, professional lighting, clean composition, 4K quality, commercial production", "cinematic": "{subject}, {action}, cinematic lighting, film grain, anamorphic lens, movie quality", "social": "{subject}, {action}, vibrant colors, engaging, vertical format, social media style", "documentary": "{subject}, {action}, natural lighting, authentic, observational, documentary style", }

def build_genre_prompt( subject: str, action: str, genre: str = "commercial" ) -> str: template = GENRE_TEMPLATES.get(genre, GENRE_TEMPLATES["commercial"]) return template.format(subject=subject, action=action)

Negative prompts for video

NEGATIVE_PROMPTS = { "quality": "blurry, distorted, low resolution, pixelated, artifacts", "motion": "jittery, stuttering, frozen, static, no motion", "anatomy": "deformed, extra limbs, missing limbs, bad anatomy", "all": "blurry, distorted, low quality, jittery, deformed, extra limbs, artifacts, watermark", }

Anti Patterns

---

Pattern

Image-style prompts for video

Why

Video needs motion and action descriptions

Fix

Add verbs, camera motion, and timing

---

Pattern

Overly complex prompts

Why

Models struggle with many elements

Fix

Focus on key subject and single clear action

References

• https://apatero.com/blog/avoid-slow-motion-wan-22-video-generation-2025

---

Id

runway-gen3-integration

Name

Runway Gen-3 API Integration

Description

Integrate Runway's Gen-3 Alpha for professional video. Features: 4K resolution, camera controls, lip sync.

Note: Requires Runway API access (paid). Credits: ~10 credits/second for Gen-3 Alpha.

Code Example

import httpx from typing import Optional import asyncio

class RunwayClient: """Runway Gen-3 API client."""

def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.runwayml.com/v1"

async def generate_video( self, prompt: str, image_url: Optional[str] = None, duration: int = 5, # seconds resolution: str = "720p", aspect_ratio: str = "16:9", ) -> dict: """Generate video with Gen-3 Alpha."""

async with httpx.AsyncClient() as client:

Create generation task

response = await client.post( f"{self.base_url}/generation", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json={ "model": "gen3a_turbo", # or "gen3a" for Alpha "prompt_text": prompt, "prompt_image": image_url, "duration": duration, "resolution": resolution, "aspect_ratio": aspect_ratio, }, timeout=30, )

if response.status_code != 200: raise Exception(f"Generation failed: {response.text}")

task = response.json() return await self._wait_for_completion(task["id"])

async def _wait_for_completion( self, task_id: str, timeout: int = 300, poll_interval: int = 5, ) -> dict: """Poll for task completion."""

start = asyncio.get_event_loop().time()

async with httpx.AsyncClient() as client: while True: response = await client.get( f"{self.base_url}/generation/{task_id}", headers={ "Authorization": f"Bearer {self.api_key}", }, )

task = response.json()

if task["status"] == "completed": return task

if task["status"] == "failed": raise Exception(f"Task failed: {task.get('error')}")

if asyncio.get_event_loop().time() - start > timeout: raise TimeoutError("Generation timed out")

await asyncio.sleep(poll_interval)

async def extend_video( self, video_url: str, prompt: str, extend_seconds: int = 4, ) -> dict: """Extend existing video."""

async with httpx.AsyncClient() as client: response = await client.post( f"{self.base_url}/extend", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json={ "model": "gen3a_turbo", "video_url": video_url, "prompt_text": prompt, "extend_duration": extend_seconds, }, )

task = response.json() return await self._wait_for_completion(task["id"])

Usage

runway = RunwayClient(os.environ["RUNWAY_API_KEY"])

Text-to-video

result = await runway.generate_video( prompt="A drone shot flying over misty mountains at sunrise", duration=5, resolution="1080p", ) print(f"Video URL: {result['output_url']}")

Image-to-video

result = await runway.generate_video( prompt="Camera slowly zooming in, subject looking at camera", image_url="https://example.com/portrait.jpg", duration=5, )

Anti Patterns

---

Pattern

Not tracking credit usage

Why

Gen-3 uses 10 credits/second, costs add up

Fix

Track usage and set budget limits

---

Pattern

Synchronous API calls

Why

Generation takes 30+ seconds

Fix

Use async with proper polling

References

• https://runwayml.com/research/introducing-gen-3-alpha
• https://docs.runwayml.com/api

---

Id

luma-dream-machine

Name

Luma Dream Machine Integration

Description

Integrate Luma's Dream Machine for cinematic video. Features: HDR output, keyframes, character reference.

Ray3: Latest model with studio-grade HDR. Draft Mode: Fast exploration before final render.

Code Example

import httpx from typing import Optional, List import asyncio

class LumaClient: """Luma Dream Machine API client."""

def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.lumalabs.ai/dream-machine/v1"

async def generate( self, prompt: str, aspect_ratio: str = "16:9", loop: bool = False, keyframes: Optional[dict] = None, ) -> dict: """Generate video with Dream Machine."""

payload = { "prompt": prompt, "aspect_ratio": aspect_ratio, "loop": loop, }

if keyframes: payload["keyframes"] = keyframes

async with httpx.AsyncClient() as client: response = await client.post( f"{self.base_url}/generations", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json=payload, )

if response.status_code != 201: raise Exception(f"Failed: {response.text}")

task = response.json() return await self._poll(task["id"])

async def generate_from_image( self, image_url: str, prompt: str, end_image_url: Optional[str] = None, # Ray3 keyframe feature ) -> dict: """Generate video from start image, optionally with end keyframe."""

keyframes = { "frame0": {"type": "image", "url": image_url}, }

if end_image_url: keyframes["frame1"] = {"type": "image", "url": end_image_url}

return await self.generate( prompt=prompt, keyframes=keyframes, )

async def _poll( self, generation_id: str, timeout: int = 300, ) -> dict: """Poll for completion."""

start = asyncio.get_event_loop().time()

async with httpx.AsyncClient() as client: while True: response = await client.get( f"{self.base_url}/generations/{generation_id}", headers={ "Authorization": f"Bearer {self.api_key}", }, )

result = response.json()

if result["state"] == "completed": return result

if result["state"] == "failed": raise Exception(f"Failed: {result.get('failure_reason')}")

if asyncio.get_event_loop().time() - start > timeout: raise TimeoutError("Generation timed out")

await asyncio.sleep(5)

Usage

luma = LumaClient(os.environ["LUMA_API_KEY"])

Text-to-video

result = await luma.generate( prompt="A spaceship traveling through a colorful nebula, cinematic", aspect_ratio="16:9", )

Image-to-video with camera motion

result = await luma.generate_from_image( image_url="https://example.com/landscape.jpg", prompt="Camera slowly panning right, revealing the scene", )

Transition between two keyframes (Ray3)

result = await luma.generate_from_image( image_url="https://example.com/start.jpg", prompt="Smooth transition, morphing", end_image_url="https://example.com/end.jpg", )

Anti Patterns

---

Pattern

Ignoring HDR capabilities

Why

Ray3 supports native HDR for pro workflows

Fix

Export as 16-bit EXR for high-end projects

References

• https://lumalabs.ai/dream-machine
• https://lumalabs.ai/ray

---

Id

video-processing-pipeline

Name

Video Processing Pipeline

Description

Build end-to-end video generation pipelines. Handle upload, generation, post-processing, delivery.

Components:

• Input validation
• Queue management
• Progress tracking
• Error handling

Code Example

import asyncio from enum import Enum from dataclasses import dataclass from typing import Optional, Callable import uuid

class VideoStatus(Enum): PENDING = "pending" GENERATING = "generating" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed"

@dataclass class VideoJob: id: str prompt: str image_url: Optional[str] status: VideoStatus progress: int output_url: Optional[str] error: Optional[str] created_at: float updated_at: float

class VideoGenerationPipeline: """End-to-end video generation pipeline."""

def __init__( self, replicate_client, storage_client, max_concurrent: int = 5, ): self.replicate = replicate_client self.storage = storage_client self.max_concurrent = max_concurrent self.jobs: dict[str, VideoJob] = {} self.semaphore = asyncio.Semaphore(max_concurrent)

async def submit( self, prompt: str, image_url: Optional[str] = None, model: str = "wavespeedai/wan-2.1-t2v-480p", callback: Optional[Callable] = None, ) -> str: """Submit video generation job."""

job_id = str(uuid.uuid4()) now = asyncio.get_event_loop().time()

job = VideoJob( id=job_id, prompt=prompt, image_url=image_url, status=VideoStatus.PENDING, progress=0, output_url=None, error=None, created_at=now, updated_at=now, )

self.jobs[job_id] = job

Start processing in background

asyncio.create_task( self._process_job(job, model, callback) )

return job_id

async def _process_job( self, job: VideoJob, model: str, callback: Optional[Callable], ): """Process a single video job."""

async with self.semaphore: try:

Update status

job.status = VideoStatus.GENERATING job.progress = 10 await self._notify(callback, job)

Generate video

input_data = {"prompt": job.prompt} if job.image_url: input_data["image"] = job.image_url

prediction = self.replicate.predictions.create( model=model, input=input_data, )

Poll for completion

while True: prediction = self.replicate.predictions.get(prediction.id)

if prediction.status == "processing": job.progress = min(80, job.progress + 10) await self._notify(callback, job)

if prediction.status == "succeeded": video_url = prediction.output break

if prediction.status == "failed": raise Exception(prediction.error)

await asyncio.sleep(5)

Post-process and store

job.status = VideoStatus.PROCESSING job.progress = 90 await self._notify(callback, job)

Upload to permanent storage

permanent_url = await self.storage.upload_from_url( video_url, f"videos/{job.id}.mp4", )

Complete

job.status = VideoStatus.COMPLETED job.progress = 100 job.output_url = permanent_url await self._notify(callback, job)

except Exception as e: job.status = VideoStatus.FAILED job.error = str(e) await self._notify(callback, job)

async def _notify( self, callback: Optional[Callable], job: VideoJob, ): """Notify callback of job update.""" job.updated_at = asyncio.get_event_loop().time()

if callback: await callback(job)

def get_status(self, job_id: str) -> Optional[VideoJob]: """Get job status.""" return self.jobs.get(job_id)

def get_queue_stats(self) -> dict: """Get queue statistics.""" statuses = [job.status for job in self.jobs.values()] return { "total": len(self.jobs), "pending": statuses.count(VideoStatus.PENDING), "generating": statuses.count(VideoStatus.GENERATING), "completed": statuses.count(VideoStatus.COMPLETED), "failed": statuses.count(VideoStatus.FAILED), }

Usage

pipeline = VideoGenerationPipeline( replicate_client=replicate.Client(), storage_client=s3_client, max_concurrent=5, )

Submit job

job_id = await pipeline.submit( prompt="A cat playing piano in a jazz club", callback=lambda job: print(f"Job {job.id}: {job.status.value} - {job.progress}%"), )

Check status

job = pipeline.get_status(job_id)

Anti Patterns

---

Pattern

No concurrency limits

Why

Can overwhelm API rate limits

Fix

Use semaphores to limit concurrent jobs

---

Pattern

Storing videos in memory

Why

Videos are large, exhaust memory

Fix

Stream to cloud storage

References

• https://songwenx.medium.com/turning-images-into-videos-a-programmable-and-cost-effective-approach-cf17ce849a02

Text To Video - Sharp Edges

Character/Object Drift Across Frames

Id

temporal-consistency-drift

Severity

high

Description

AI video models struggle with temporal consistency. Faces, hands, and objects can change or "drift" across frames. The longer the video, the worse the drift.

Wrong Way

Generating long videos expecting perfect consistency

result = generate_video( prompt="A woman walking for 30 seconds", duration=30, # Too long! )

Face changes, clothing shifts, hands morph

No verification of output quality

return result # May have severe artifacts

Right Way

Generate shorter clips and chain them

def generate_with_consistency(prompt: str, total_duration: int = 15): """Generate longer videos with consistency checks."""

clips = [] clip_duration = 5 # Shorter clips have better consistency

for i in range(total_duration // clip_duration):

Generate clip

if i == 0: clip = generate_video(prompt, duration=clip_duration) else:

Use last frame as reference for next clip

last_frame = extract_last_frame(clips[-1]) clip = generate_from_image( image_url=last_frame, prompt=prompt, duration=clip_duration, )

clips.append(clip)

Concatenate with crossfade

return concatenate_videos(clips, crossfade_ms=500)

Use models with character consistency features

Kling's "Elements" feature maintains character across scenes

result = generate_with_elements( prompt="A woman walking through the city", character_reference=["face1.jpg", "face2.jpg", "face3.jpg"], duration=10, )

For critical content, generate multiple versions

candidates = [ generate_video(prompt, seed=i) for i in range(3) ]

Select best based on quality metrics or human review

Detection Patterns

• duration.*[2-9][0-9]
• duration.*1[0-9][0-9]

References

• https://www.upuply.com/blog/what-are-the-limitations-of-current-ai-video-generation-models
• https://arxiv.org/html/2403.16407v1

Unintended Slow Motion in Generated Videos

Id

slow-motion-artifacts

Severity

medium

Description

Videos often generate slower than intended. Motion blur settings and guidance affect perceived speed. Common with Wan and other diffusion models.

Wrong Way

Default settings often produce slow motion

result = generate_video( prompt="A person running quickly through the park",

Using defaults that produce slow motion

)

Output: Person appears to run in slow motion

Over-emphasizing quality over motion

result = generate_video( prompt="Running person", guidance_scale=15.0, # Too high, limits motion motion_blur=0.8, # Too much blur )

Right Way

Explicit speed descriptors in prompt

SPEED_KEYWORDS = { "fast": "quickly, rapidly, fast motion, energetic, swift", "normal": "natural pace, realistic timing, steady movement", "slow": "slowly, gracefully, gentle motion, deliberate", }

def generate_with_motion_control( prompt: str, speed: str = "normal", motion_guidance: float = 1.0, # 0.8-1.2 range ): """Generate with controlled motion speed."""

Add speed keywords to prompt

speed_words = SPEED_KEYWORDS.get(speed, SPEED_KEYWORDS["normal"]) enhanced_prompt = f"{prompt}, {speed_words}"

return generate_video( prompt=enhanced_prompt, guidance_scale=6.0, # Moderate guidance motion_blur=0.2, # Low motion blur motion_guidance=motion_guidance, )

For fast action, use specific settings

result = generate_video( prompt="An athlete sprinting, rapid leg movement, dynamic action", num_inference_steps=20, # Fewer steps, less smoothing motion_blur=0.1, # Minimal blur )

Post-process: speed up if needed

def adjust_video_speed(video_path: str, speed_factor: float = 1.5): """Speed up video if motion is too slow."""

Use ffmpeg to adjust speed

subprocess.run([ "ffmpeg", "-i", video_path, "-filter:v", f"setpts={1/speed_factor}*PTS", "-filter:a", f"atempo={speed_factor}", "output.mp4" ])

Detection Patterns

• guidance_scale.*1[5-9]
• motion_blur.*0\.[7-9]

References

• https://apatero.com/blog/avoid-slow-motion-wan-22-video-generation-2025

Blocking on Long Video Generation

Id

generation-time-blocking

Severity

high

Description

Video generation takes 30-150+ seconds. Blocking synchronous calls freeze the application. No timeout handling causes infinite waits.

Wrong Way

Synchronous blocking call

@app.post("/generate-video") def generate_video(prompt: str): result = replicate.run( # Blocks for 2+ minutes! "model/version", input={"prompt": prompt} ) return {"url": result}

User's request times out, but generation continues

No way to cancel or track progress

Right Way

import asyncio from fastapi import BackgroundTasks

Return immediately, process in background

@app.post("/generate-video") async def submit_generation( prompt: str, background_tasks: BackgroundTasks, ): job_id = str(uuid.uuid4())

Start generation in background

background_tasks.add_task( process_video_generation, job_id, prompt, )

return { "job_id": job_id, "status_url": f"/api/video-status/{job_id}", }

@app.get("/video-status/{job_id}") async def get_status(job_id: str): job = await get_job(job_id) return { "status": job.status, "progress": job.progress, "url": job.output_url, }

With proper timeout and cancellation

async def generate_with_timeout( prompt: str, timeout: int = 300, # 5 minutes max ): client = replicate.Client()

prediction = client.predictions.create( model="wavespeedai/wan-2.1-t2v-480p", input={"prompt": prompt} )

start = time.time()

while True: prediction = client.predictions.get(prediction.id)

if prediction.status == "succeeded": return prediction.output

if prediction.status == "failed": raise Exception(f"Failed: {prediction.error}")

if time.time() - start > timeout:

Cancel the prediction

client.predictions.cancel(prediction.id) raise TimeoutError("Generation timed out")

await asyncio.sleep(5)

Detection Patterns

• replicate\.run\((?!.*await)
• generate_video(?!.*async)

References

• https://replicate.com/docs/topics/predictions

Video Generation Cost Explosion

Id

cost-explosion

Severity

high

Description

Video generation is expensive compared to images. Costs range from $0.05-0.50+ per video. Unbounded generation or retries cause cost spikes.

Wrong Way

No cost tracking or limits

async def generate_variations(prompt: str, count: int = 10): results = [] for i in range(count): result = await generate_video(prompt) # $0.25 each! results.append(result) return results # $2.50 for this call alone

Retry without limits

async def generate_with_retry(prompt: str): while True: # Infinite retries! try: return await generate_video(prompt) except: continue # Each retry costs money

Right Way

from dataclasses import dataclass from datetime import datetime, timedelta

@dataclass class CostTracker: daily_budget: float = 50.0 cost_per_video: float = 0.28 # Kling pricing daily_spent: float = 0.0 last_reset: datetime = datetime.now()

def can_generate(self) -> bool: self._reset_if_new_day() return self.daily_spent + self.cost_per_video <= self.daily_budget

def record_generation(self, cost: float = None): self._reset_if_new_day() self.daily_spent += cost or self.cost_per_video

def _reset_if_new_day(self): if datetime.now().date() > self.last_reset.date(): self.daily_spent = 0.0 self.last_reset = datetime.now()

Model cost reference

MODEL_COSTS = { "wan-2.1-t2v-480p": 0.05, # ~5 sec video "wan-2.1-t2v-720p": 0.12, "kling-v2.1": 0.28, # 5 sec video "runway-gen3-turbo": 0.25, # 5 credits/sec @ $0.05/credit "runway-gen3-alpha": 0.50, # 10 credits/sec }

class BudgetedGenerator: def __init__(self, daily_budget: float = 50.0): self.tracker = CostTracker(daily_budget=daily_budget)

async def generate( self, prompt: str, model: str = "wan-2.1-t2v-480p", max_retries: int = 2, # Limited retries ): cost = MODEL_COSTS.get(model, 0.10)

if not self.tracker.can_generate(): raise Exception("Daily budget exceeded")

for attempt in range(max_retries + 1): try: result = await generate_video(prompt, model=model) self.tracker.record_generation(cost) return result except Exception as e: if attempt == max_retries: raise

raise Exception("Max retries exceeded")

Detection Patterns

• while.True.generate
• range\([5-9]|[1-9][0-9]\).*generate

References

• https://replicate.com/pricing

Wrong Resolution or Aspect Ratio

Id

resolution-aspect-ratio

Severity

medium

Description

Videos generated with wrong aspect ratio look distorted. Some models only support specific resolutions. Mismatched aspect causes letterboxing or stretching.

Wrong Way

Ignoring aspect ratio

result = generate_video( prompt="Vertical TikTok video", width=1920, height=1080, # Horizontal for vertical content! )

Using unsupported resolution

result = generate_video( prompt="4K video", width=3840, height=2160, # Model only supports up to 1080p )

Right Way

Standard aspect ratios

ASPECT_RATIOS = { "16:9": (1280, 720), # YouTube, horizontal "9:16": (720, 1280), # TikTok, Reels, vertical "1:1": (1024, 1024), # Instagram square "4:3": (1024, 768), # Classic video "21:9": (1920, 823), # Cinematic widescreen }

Model resolution limits

MODEL_LIMITS = { "wan-2.1": {"max": (1280, 720), "supported": ["16:9", "9:16", "1:1"]}, "kling-v2.1": {"max": (1920, 1080), "supported": ["16:9", "9:16", "1:1", "4:3"]}, "runway-gen3": {"max": (3840, 2160), "supported": ["16:9", "9:16", "1:1", "21:9"]}, }

def get_resolution( platform: str, model: str = "wan-2.1", ) -> tuple[int, int]: """Get optimal resolution for platform and model."""

PLATFORM_RATIOS = { "youtube": "16:9", "tiktok": "9:16", "reels": "9:16", "instagram": "1:1", "twitter": "16:9", }

ratio = PLATFORM_RATIOS.get(platform, "16:9") width, height = ASPECT_RATIOS[ratio]

Check model limits

limits = MODEL_LIMITS.get(model, {}) max_w, max_h = limits.get("max", (1920, 1080))

Scale down if needed

if width > max_w or height > max_h: scale = min(max_w / width, max_h / height) width = int(width scale) height = int(height scale)

return width, height

Usage

width, height = get_resolution("tiktok", model="wan-2.1") result = generate_video( prompt="Short form content", width=width, height=height, )

Detection Patterns

• width.*3840
• height.2160.wan

References

• https://replicate.com/collections/text-to-video

Excessive Motion Blur and Smoothing

Id

motion-blur-artifacts

Severity

medium

Description

Over-smoothed videos look artificial and "dreamlike". High motion blur makes movement unclear. Temporal consistency fixes can cause blur artifacts.

Wrong Way

Over-emphasizing smoothness

result = generate_video( prompt="Action scene", motion_blur=0.9, # Too much blur! temporal_smoothing=True, # Over-smooths frames guidance_scale=15.0, # Over-constrained )

Result: Blurry, unclear motion, dreamlike quality

Right Way

Balance between consistency and clarity

def generate_natural_motion( prompt: str, motion_type: str = "moderate", ): """Generate with natural motion settings."""

MOTION_SETTINGS = { "static": { "motion_blur": 0.0, "guidance_scale": 8.0, }, "moderate": { "motion_blur": 0.15, "guidance_scale": 6.0, }, "dynamic": { "motion_blur": 0.25, "guidance_scale": 5.0, }, "action": { "motion_blur": 0.1, "guidance_scale": 4.0, "num_inference_steps": 20, # Fewer steps for dynamic motion }, }

settings = MOTION_SETTINGS.get(motion_type, MOTION_SETTINGS["moderate"])

return generate_video( prompt=prompt, **settings, )

For action scenes, minimize blur

result = generate_natural_motion( prompt="Fast paced action sequence, dynamic movement", motion_type="action", )

Post-process to fix over-smoothing

Use alpha blending: 70% AI + 30% original (if available)

def blend_with_original(ai_video: str, original: str, alpha: float = 0.7): """Blend AI video with original for natural texture."""

Preserves original motion blur and texture

...

Detection Patterns

• motion_blur.*0\.[89]
• temporal_smoothing.*True

References

• https://help.scenario.com/en/articles/troubleshooting-video-generations/

Missing Video Content Moderation

Id

no-content-moderation

Severity

critical

Description

AI video can generate inappropriate content. Video moderation is more complex than image moderation. Frame-by-frame analysis required for thorough checks.

Wrong Way

Direct generation without moderation

@app.post("/generate") async def generate(prompt: str): result = await generate_video(prompt) # No checks! return {"url": result}

Only checking prompt, not output

@app.post("/generate") async def generate(prompt: str): if not await moderate_prompt(prompt): return {"error": "blocked"}

result = await generate_video(prompt) return {"url": result} # Video not checked!

Right Way

import asyncio from typing import List

async def moderate_video(video_url: str) -> dict: """Moderate video by sampling frames."""

Extract frames at intervals

frames = await extract_frames(video_url, interval_seconds=1)

Check each frame

results = [] for frame in frames: result = await moderate_image(frame) results.append(result)

Early exit if any frame is blocked

if result["blocked"]: return { "allowed": False, "reason": result["reason"], "frame": frame, }

return {"allowed": True}

async def extract_frames( video_url: str, interval_seconds: float = 1, ) -> List[str]: """Extract frames from video at intervals.""" import subprocess import tempfile

with tempfile.TemporaryDirectory() as tmpdir:

Download video

video_path = f"{tmpdir}/video.mp4" await download_file(video_url, video_path)

Extract frames with ffmpeg

subprocess.run([ "ffmpeg", "-i", video_path, "-vf", f"fps=1/{interval_seconds}", f"{tmpdir}/frame_%04d.jpg" ])

Upload frames and return URLs

frames = [] for f in sorted(os.listdir(tmpdir)): if f.startswith("frame_"): url = await upload_to_storage(f"{tmpdir}/{f}") frames.append(url)

return frames

Full moderation pipeline

@app.post("/generate") async def generate_safe(prompt: str):

1. Moderate prompt

if not await moderate_prompt(prompt): return {"error": "prompt_blocked"}

2. Generate video

try: result = await generate_video(prompt) except Exception as e: return {"error": str(e)}

3. Moderate output video

moderation = await moderate_video(result["url"]) if not moderation["allowed"]:

Delete or quarantine video

await delete_video(result["url"]) return { "error": "video_blocked", "reason": moderation["reason"], }

return {"url": result["url"]}

Detection Patterns

• generate_video(?!.*moderat)
• return.url(?!.check)

References

• https://www.lovart.ai/blog/video-generators-review

Missing Watermarks and Provenance

Id

watermark-provenance

Severity

medium

Description

AI-generated videos should be labeled. Some platforms require disclosure. Regulations increasingly mandate provenance.

Wrong Way

Generating without provenance

result = generate_video(prompt)

Stripping watermarks

result = remove_watermarks(result) # May violate TOS

Publishing without disclosure

await publish_to_platform(result) # Regulation violation

Right Way

Preserve or add watermarks

def generate_with_provenance(prompt: str) -> dict: result = generate_video(prompt)

Add C2PA metadata (content authenticity)

result_with_provenance = add_provenance( result, generator="wan-2.2", prompt=prompt, timestamp=datetime.now().isoformat(), )

Add visible watermark if required

result_watermarked = add_watermark( result_with_provenance, text="AI Generated", position="bottom-right", opacity=0.5, )

return { "url": result_watermarked, "provenance": { "generator": "wan-2.2", "created_at": datetime.now().isoformat(), "is_synthetic": True, } }

Platform-specific disclosure

PLATFORM_REQUIREMENTS = { "youtube": { "requires_label": True, "label_method": "metadata", }, "tiktok": { "requires_label": True, "label_method": "caption", }, "meta": { "requires_label": True, "label_method": "metadata_and_caption", }, }

def prepare_for_platform( video: dict, platform: str, ) -> dict: """Prepare video for platform with proper disclosure."""

requirements = PLATFORM_REQUIREMENTS.get(platform, {})

if requirements.get("requires_label"): method = requirements.get("label_method")

if "metadata" in method: video = add_ai_metadata(video)

if "caption" in method: video["caption_prefix"] = "#AIGenerated"

return video

Detection Patterns

• remove_watermark
• strip.*metadata

References

• https://www.lovart.ai/blog/video-generators-review

Text To Video - Validations

Video API Key in Client Code

Id

video-api-key-exposed

Severity

error

Description

Video generation API keys must be server-side only

Pattern

(NEXT_PUBLIC|REACT_APP|VITE).(RUNWAY|LUMA|KLING|REPLICATE|PIKA).KEY

Message

Video API key exposed to client. Use server-side routes only.

Autofix

Hardcoded Video API Key

Id

hardcoded-video-api-key

Severity

error

Description

API keys should use environment variables

Pattern

(runway_|luma_|r8_|kling_)[A-Za-z0-9]{20,}

Message

Hardcoded API key detected. Use environment variables.

Autofix

Missing Video Prompt Moderation

Id

no-video-prompt-moderation

Severity

error

Description

Video prompts must be moderated before generation

Pattern

request\.(body|query)\.prompt.generateVideo(?!.moderat)

Message

Video prompt passed without moderation. Check for policy violations.

Autofix

Missing Video Output Moderation

Id

no-video-output-moderation

Severity

warning

Description

Generated videos should be checked before serving

Pattern

generateVideo\(.\).return.url(?!.check|scan)

Message

Generated video returned without content moderation.

Autofix

Video Safety Checker Disabled

Id

video-safety-disabled

Severity

warning

Description

Model safety checkers should remain enabled

Pattern

safety_checker.false|nsfw_filter.false|content_filter.*disabled

Message

Video safety checker disabled. Enable for production.

Autofix

Video Generation Without Rate Limiting

Id

no-video-rate-limiting

Severity

warning

Description

Video generation is expensive - rate limiting required

Pattern

async.generateVideo.request(?!.*rateLimit|limit)

Message

Video generation endpoint without rate limiting.

Autofix

Missing Timeout for Video Generation

Id

no-video-timeout

Severity

error

Description

Video generation can take minutes - timeout required

Pattern

await.generate.video(?!.*timeout)

Message

Video generation without timeout. Add 5-10 minute timeout.

Autofix

Blocking Video Generation

Id

synchronous-video-generation

Severity

error

Description

Video generation should be async with status polling

Pattern

await.generateVideo\((?!.queue|poll|async|webhook)

Message

Synchronous video generation blocks requests. Use async pattern.

Autofix

Missing Video Cost Tracking

Id

no-video-cost-tracking

Severity

warning

Description

Video generation costs should be tracked

Pattern

generateVideo\((?!.*cost|budget|credits)

Message

No cost tracking for video generation. Add budget controls.

Autofix

Unbounded Video Duration

Id

unbounded-video-duration

Severity

warning

Description

Video duration should be capped to control costs

Pattern

duration.request\.(body|query)(?!.Math\.min|clamp|max)

Message

User-controlled duration without limit. Cap at maximum allowed.

Autofix

Invalid Video Resolution

Id

invalid-video-resolution

Severity

warning

Description

Video dimensions must match model requirements

Pattern

generateVideo\(.width.(?!.*512|576|640|704|768|832|896|960|1024|1280)

Message

Check resolution compatibility with model requirements.

Autofix

Invalid Aspect Ratio

Id

invalid-aspect-ratio

Severity

warning

Description

Aspect ratio must match model capabilities

Pattern

aspect_ratio.[0-9]+:[0-9]+(?!.16:9|9:16|1:1|4:3|3:4)

Message

Non-standard aspect ratio may cause issues. Use 16:9, 9:16, or 1:1.

Autofix

Missing Input Image Validation

Id

no-input-image-validation

Severity

warning

Description

Image-to-video inputs should be validated

Pattern

imageToVideo\(.image(?!.validate|check|resize)

Message

Input image not validated. Check format, size, and dimensions.

Autofix

Missing Retry Logic

Id

no-generation-retry

Severity

warning

Description

Video generation should retry on transient failures

Pattern

await.generateVideo\((?!.retry|attempts)

Message

No retry logic for video generation failures.

Autofix

Missing Queue Failure Handling

Id

no-queue-failure-handling

Severity

warning

Description

Video generation queues should handle failures

Pattern

queue\.add\(.video(?!.onFail|deadLetter)

Message

Video queue without failure handling. Add dead letter queue.

Autofix