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

"""
Simple API interface for the YouTube processing workflow.
"""
from flask import Flask, request, jsonify
from typing import Dict, Any
import traceback
import logging

from workflow import YouTubeProcessingWorkflow

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize Flask app
app = Flask(__name__)

# Initialize workflow
workflow = None

def init_workflow():
    """Initialize the workflow instance."""
    global workflow
    try:
        workflow = YouTubeProcessingWorkflow()
        logger.info("Workflow initialized successfully")
        return True
    except Exception as e:
        logger.error(f"Failed to initialize workflow: {str(e)}")
        return False

@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint."""
    return jsonify({
        "status": "healthy",
        "workflow_initialized": workflow is not None
    })

@app.route('/process', methods=['POST'])
def process_video():
    """Process a YouTube video through the complete workflow."""
    try:
        # Validate request
        if not workflow:
            return jsonify({
                "success": False,
                "error": "Workflow not initialized"
            }), 500
        
        # Get request data
        data = request.get_json()
        if not data:
            return jsonify({
                "success": False,
                "error": "No JSON data provided"
            }), 400
        
        # Validate required fields
        required_fields = ['youtube_url', 'target_language', 'summarization_prompt']
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            return jsonify({
                "success": False,
                "error": f"Missing required fields: {', '.join(missing_fields)}"
            }), 400
        
        youtube_url = data['youtube_url']
        target_language = data['target_language']
        summarization_prompt = data['summarization_prompt']
        metadata = data.get('metadata', {})
        
        # Add request metadata
        metadata.update({
            "api_source": "flask_api",
            "request_timestamp": str(request.environ.get('REQUEST_TIME', '')),
        })
        
        logger.info(f"Processing video: {youtube_url}")
        
        # Process the video
        results = workflow.process_youtube_video(
            youtube_url=youtube_url,
            target_language=target_language,
            summarization_prompt=summarization_prompt,
            workflow_metadata=metadata
        )
        
        # Return results
        status_code = 200 if results['success'] else 500
        return jsonify(results), status_code
        
    except Exception as e:
        logger.error(f"Error processing request: {str(e)}")
        logger.error(f"Traceback: {traceback.format_exc()}")
        return jsonify({
            "success": False,
            "error": f"Internal server error: {str(e)}"
        }), 500

@app.route('/transcribe', methods=['POST'])
def transcribe_only():
    """Transcribe a YouTube video without translation or summarization."""
    try:
        if not workflow:
            return jsonify({
                "success": False,
                "error": "Workflow not initialized"
            }), 500
        
        data = request.get_json()
        if not data or 'youtube_url' not in data:
            return jsonify({
                "success": False,
                "error": "youtube_url is required"
            }), 400
        
        youtube_url = data['youtube_url']
        logger.info(f"Transcribing video: {youtube_url}")
        
        transcript = workflow.transcriber.transcribe(youtube_url)
        
        return jsonify({
            "success": not transcript.startswith("Error"),
            "transcript": transcript,
            "error": transcript if transcript.startswith("Error") else None
        })
        
    except Exception as e:
        logger.error(f"Error in transcription: {str(e)}")
        return jsonify({
            "success": False,
            "error": f"Internal server error: {str(e)}"
        }), 500

@app.route('/translate', methods=['POST'])
def translate_text():
    """Translate text to a target language."""
    try:
        if not workflow:
            return jsonify({
                "success": False,
                "error": "Workflow not initialized"
            }), 500
        
        data = request.get_json()
        if not data:
            return jsonify({
                "success": False,
                "error": "No JSON data provided"
            }), 400
        
        required_fields = ['text', 'target_language']
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            return jsonify({
                "success": False,
                "error": f"Missing required fields: {', '.join(missing_fields)}"
            }), 400
        
        text = data['text']
        target_language = data['target_language']
        
        logger.info(f"Translating text to {target_language}")
        
        translated_text = workflow.translator.translate(text, target_language)
        
        return jsonify({
            "success": not translated_text.startswith("Error"),
            "translated_text": translated_text,
            "original_text": text,
            "target_language": target_language,
            "error": translated_text if translated_text.startswith("Error") else None
        })
        
    except Exception as e:
        logger.error(f"Error in translation: {str(e)}")
        return jsonify({
            "success": False,
            "error": f"Internal server error: {str(e)}"
        }), 500

@app.route('/summarize', methods=['POST'])
def summarize_text():
    """Summarize text based on a custom prompt."""
    try:
        if not workflow:
            return jsonify({
                "success": False,
                "error": "Workflow not initialized"
            }), 500
        
        data = request.get_json()
        if not data:
            return jsonify({
                "success": False,
                "error": "No JSON data provided"
            }), 400
        
        required_fields = ['text', 'summarization_prompt']
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            return jsonify({
                "success": False,
                "error": f"Missing required fields: {', '.join(missing_fields)}"
            }), 400
        
        text = data['text']
        summarization_prompt = data['summarization_prompt']
        
        logger.info("Summarizing text")
        
        summary = workflow.summarizer.summarize(text, summarization_prompt)
        
        return jsonify({
            "success": not summary.startswith("Error"),
            "summary": summary,
            "original_text": text,
            "summarization_prompt": summarization_prompt,
            "error": summary if summary.startswith("Error") else None
        })
        
    except Exception as e:
        logger.error(f"Error in summarization: {str(e)}")
        return jsonify({
            "success": False,
            "error": f"Internal server error: {str(e)}"
        }), 500

if __name__ == '__main__':
    # Initialize workflow
    if init_workflow():
        app.run(host='0.0.0.0', port=5000, debug=True)
    else:
        logger.error("Failed to initialize workflow. Exiting.")