bytelyst-devops-tools/supabase monitor/demo.py

"""
Demo script for the YouTube Processing Workflow.
"""
import json
from workflow import YouTubeProcessingWorkflow

def demo_workflow():
    """Demonstrate the complete workflow with example data."""
    
    print("🎬 YouTube Processing Workflow Demo")
    print("=" * 50)
    
    # Example data
    demo_configs = [
        {
            "youtube_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
            "target_language": "Spanish", 
            "summarization_prompt": "Summarize in 5 bullet points for students to revise quickly",
            "description": "Rick Roll - Student Learning Summary"
        },
        {
            "youtube_url": "https://www.youtube.com/watch?v=jNQXAC9IVRw", 
            "target_language": "French",
            "summarization_prompt": "Create a 3-point executive summary highlighting key business insights",
            "description": "Me at the zoo - Business Insights"
        }
    ]
    
    # Initialize workflow
    try:
        workflow = YouTubeProcessingWorkflow()
        print("✅ Workflow initialized successfully")
    except Exception as e:
        print(f"❌ Failed to initialize workflow: {str(e)}")
        return
    
    # Process each example
    for i, config in enumerate(demo_configs, 1):
        print(f"\n🎯 Demo {i}: {config['description']}")
        print("-" * 40)
        
        try:
            results = workflow.process_youtube_video(
                youtube_url=config["youtube_url"],
                target_language=config["target_language"],
                summarization_prompt=config["summarization_prompt"],
                workflow_metadata={
                    "demo_run": True,
                    "demo_id": i
                }
            )
            
            # Print detailed results
            workflow.print_workflow_summary(results)
            
            # Save results to file
            filename = f"demo_results_{i}.json"
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(results, f, indent=2, ensure_ascii=False)
            print(f"📁 Results saved to: {filename}")
            
        except Exception as e:
            print(f"❌ Error processing demo {i}: {str(e)}")
            continue

def demo_individual_operations():
    """Demonstrate individual agent operations."""
    
    print("\n🔧 Individual Agent Operations Demo")
    print("=" * 50)
    
    # Sample data
    sample_transcript = """
    Welcome to this educational video about machine learning. 
    Today we'll cover the basics of supervised learning, 
    including algorithms like linear regression and decision trees.
    These concepts are fundamental to understanding AI.
    """
    
    sample_translated = """
    Bienvenidos a este video educativo sobre aprendizaje automático.
    Hoy cubriremos los conceptos básicos de aprendizaje supervisado,
    incluyendo algoritmos como regresión lineal y árboles de decisión.
    Estos conceptos son fundamentales para entender la IA.
    """
    
    try:
        workflow = YouTubeProcessingWorkflow()
        
        # Test translation
        print("🌍 Testing Translation:")
        translated = workflow.translator.translate(sample_transcript, "Spanish")
        print(f"Original: {sample_transcript[:100]}...")
        print(f"Translated: {translated[:100]}...")
        
        # Test summarization
        print("\n📝 Testing Summarization:")
        summary = workflow.summarizer.summarize(
            sample_translated, 
            "Summarize in 3 bullet points about machine learning concepts"
        )
        print(f"Summary: {summary}")
        
    except Exception as e:
        print(f"❌ Error in individual operations demo: {str(e)}")

def demo_api_calls():
    """Demonstrate API usage examples."""
    
    print("\n🌐 API Usage Examples")
    print("=" * 50)
    
    api_examples = {
        "Complete Workflow": """
curl -X POST http://localhost:5000/process \\
  -H "Content-Type: application/json" \\
  -d '{
    "youtube_url": "https://www.youtube.com/watch?v=example",
    "target_language": "Spanish",
    "summarization_prompt": "Summarize in 5 bullet points",
    "metadata": {"user_id": "demo_user"}
  }'
        """,
        
        "Transcribe Only": """
curl -X POST http://localhost:5000/transcribe \\
  -H "Content-Type: application/json" \\
  -d '{"youtube_url": "https://www.youtube.com/watch?v=example"}'
        """,
        
        "Translate Text": """
curl -X POST http://localhost:5000/translate \\
  -H "Content-Type: application/json" \\
  -d '{
    "text": "Your text here",
    "target_language": "French"
  }'
        """,
        
        "Summarize Text": """
curl -X POST http://localhost:5000/summarize \\
  -H "Content-Type: application/json" \\
  -d '{
    "text": "Your text here", 
    "summarization_prompt": "Summarize in 3 bullet points"
  }'
        """
    }
    
    for operation, example in api_examples.items():
        print(f"\n📡 {operation}:")
        print(example)
    
    print(f"\n💡 To test these API calls:")
    print(f"1. Start the API server: python api.py")
    print(f"2. Run the curl.commands above in another terminal")
    print(f"3. Check the responses")

def main():
    """Main demo function."""
    
    print("🚀 Multi-Agent YouTube Processing Workflow")
    print("Demo Script - Comprehensive Testing")
    print("=" * 60)
    
    # Check if API keys are configured
    import os
    from dotenv import load_dotenv
    load_dotenv()
    
    missing_keys = []
    if not os.getenv("PERPLEXITY_API_KEY"):
        missing_keys.append("PERPLEXITY_API_KEY")
    if not os.getenv("OPENAI_API_KEY"):
        missing_keys.append("OPENAI_API_KEY")
    
    if missing_keys:
        print(f"⚠️  Missing API keys: {', '.join(missing_keys)}")
        print("Please configure your API keys in the .env file")
        print("\nDemo will show individual operations only...\n")
        
        # Show just the API examples
        demo_api_calls()
        return
    
    # Run all demos
    try:
        demo_individual_operations()
        demo_api_calls()
        
        # Ask user if they want to run the full workflow
        response = input("\nRun full workflow demo with YouTube videos? (y/n): ")
        if response.lower() == 'y':
            demo_workflow()
        else:
            print("\nDemo completed! Check the examples above.")
            
    except KeyboardInterrupt:
        print("\n\n👋 Demo interrupted by user")
    except Exception as e:
        print(f"\n❌ Demo error: {str(e)}")

if __name__ == "__main__":
    main()