Sora video creator agent

from opperator import OpperatorAgent
import openai
import os
import time
from pathlib import Path

class SoraAgent(OpperatorAgent):
    """Agent for generating videos using OpenAI's Sora API"""

    def __init__(self):
        super().__init__()
        self.current_generation = None

    async def on_start(self):
        """Initialize the agent when it starts"""
        self.log("Sora video generation agent started")

        # Get API key from secret manager
        api_key = self.get_secret("openai_api_key")
        if not api_key:
            self.log("⚠️  No OpenAI API key found. Set it with: op secret create openai_api_key")
        else:
            self.client = openai.OpenAI(api_key=api_key)
            self.log("✓ OpenAI API key configured")

    async def generate_video(
        self,
        prompt: str,
        duration: int = 5,
        aspect_ratio: str = "16:9"
    ) -> str:
        """
        Generate a video using Sora API

        Args:
            prompt: Description of the video to generate
            duration: Video duration in seconds (default: 5)
            aspect_ratio: Video aspect ratio (default: 16:9)

        Returns:
            Path to the generated video file
        """
        try:
            # Check if API key is configured
            if not hasattr(self, 'client'):
                return "❌ OpenAI API key not configured. Set it with: op secret create openai_api_key"

            self.log(f"🎬 Generating video: {prompt[:50]}...")

            # Update status section
            self.set_status_section("generation", {
                "title": "Current Generation",
                "items": [
                    {"label": "Status", "value": "Generating..."},
                    {"label": "Prompt", "value": prompt[:50] + "..."},
                    {"label": "Duration", "value": f"{duration}s"}
                ]
            })

            # Call Sora API
            # Note: This is a high-level example. Actual Sora API may differ.
            response = self.client.videos.generate(
                model="sora-1.0",
                prompt=prompt,
                duration=duration,
                aspect_ratio=aspect_ratio
            )

            # Wait for generation to complete
            video_id = response.id
            self.log(f"⏳ Video generation started (ID: {video_id})")

            while True:
                status = self.client.videos.retrieve(video_id)

                if status.status == "completed":
                    break
                elif status.status == "failed":
                    error_msg = f"❌ Video generation failed: {status.error}"
                    self.log(error_msg)
                    self.clear_status_section("generation")
                    return error_msg

                time.sleep(2)

            # Download the video
            video_url = status.url
            output_dir = Path.home() / "Videos" / "sora_generated"
            output_dir.mkdir(parents=True, exist_ok=True)

            timestamp = int(time.time())
            output_path = output_dir / f"sora_{timestamp}.mp4"

            # Download video content
            video_data = self.client.videos.download(video_id)
            with open(output_path, 'wb') as f:
                f.write(video_data)

            self.log(f"✅ Video saved to: {output_path}")

            # Update status to show completion
            self.set_status_section("generation", {
                "title": "Last Generation",
                "items": [
                    {"label": "Status", "value": "✅ Complete"},
                    {"label": "Prompt", "value": prompt[:50] + "..."},
                    {"label": "File", "value": str(output_path)}
                ]
            })

            return f"✅ Video generated successfully!\n\nSaved to: {output_path}\n\nPrompt: {prompt}"

        except Exception as e:
            error_msg = f"❌ Error generating video: {str(e)}"
            self.log(error_msg)
            self.clear_status_section("generation")
            return error_msg

    async def list_generations(self) -> str:
        """
        List all generated videos

        Returns:
            List of generated video files
        """
        output_dir = Path.home() / "Videos" / "sora_generated"

        if not output_dir.exists():
            return "No videos generated yet."

        videos = list(output_dir.glob("sora_*.mp4"))

        if not videos:
            return "No videos found in the output directory."

        result = f"📹 Found {len(videos)} generated video(s):\n\n"
        for video in sorted(videos, reverse=True):
            size_mb = video.stat().st_size / (1024 * 1024)
            result += f"- {video.name} ({size_mb:.1f} MB)\n"

        return result

if __name__ == "__main__":
    agent = SoraAgent()
    agent.run()