""" Video Service High-level service for video generation. Abstracts all API complexity from the UI layer. """ from typing import Callable, Optional, List from dataclasses import dataclass from ..api.client import StackNetClient @dataclass class GeneratedVideo: """Generated video result.""" video_url: str video_path: Optional[str] = None thumbnail_url: Optional[str] = None duration: Optional[float] = None prompt: Optional[str] = None class VideoService: """ Service for video generation. Provides clean interfaces for: - Text-to-video generation - Image-to-video animation """ def __init__(self, client: Optional[StackNetClient] = None): self.client = client or StackNetClient() async def generate_video( self, prompt: str, duration: int = 10, style: Optional[str] = None, on_progress: Optional[Callable[[float, str], None]] = None ) -> List[GeneratedVideo]: """ Generate video from a text prompt using generate_video_2 tool. Args: prompt: Description of desired video duration: Target duration in seconds style: Style preset (Cinematic, Animation, etc.) on_progress: Callback for progress updates Returns: List of generated videos """ full_prompt = prompt if style and style != "Cinematic": full_prompt = f"{prompt}, {style.lower()} style" result = await self.client.submit_tool_task( tool_name="generate_video_2", parameters={ "prompt": full_prompt, "duration": duration }, on_progress=on_progress ) if not result.success: raise Exception(result.error or "Video generation failed") return self._parse_video_result(result.data, prompt) async def animate_image( self, image_url: str, motion_prompt: str, duration: int = 5, on_progress: Optional[Callable[[float, str], None]] = None ) -> List[GeneratedVideo]: """ Animate a static image into video using generate_image_to_video_2 tool. Args: image_url: URL to source image motion_prompt: Description of desired motion duration: Target duration in seconds on_progress: Progress callback Returns: List of animated videos """ result = await self.client.submit_tool_task( tool_name="generate_image_to_video_2", parameters={ "prompt": motion_prompt, "image_url": image_url, "duration": duration }, on_progress=on_progress ) if not result.success: raise Exception(result.error or "Image animation failed") return self._parse_video_result(result.data, motion_prompt) def _parse_video_result(self, data: dict, prompt: str) -> List[GeneratedVideo]: """Parse API response into GeneratedVideo objects.""" videos = [] # Handle various response formats raw_videos = data.get("videos", []) if not raw_videos: # Check for single video URL video_url = ( data.get("video_url") or data.get("videoUrl") or data.get("url") or data.get("content") ) if video_url: raw_videos = [{"url": video_url}] for vid_data in raw_videos: if isinstance(vid_data, str): video_url = vid_data else: video_url = ( vid_data.get("url") or vid_data.get("video_url") or vid_data.get("videoUrl") ) if video_url: videos.append(GeneratedVideo( video_url=video_url, thumbnail_url=vid_data.get("thumbnail") if isinstance(vid_data, dict) else None, duration=vid_data.get("duration") if isinstance(vid_data, dict) else None, prompt=prompt )) return videos async def download_video(self, video: GeneratedVideo) -> str: """Download a video to local file.""" if video.video_path: return video.video_path # Determine extension from URL url = video.video_url if ".webm" in url: ext = ".webm" elif ".mov" in url: ext = ".mov" else: ext = ".mp4" filename = f"video_{hash(url) % 10000}{ext}" video.video_path = await self.client.download_file(url, filename) return video.video_path def cleanup(self): """Clean up temporary files.""" self.client.cleanup()