youtube-cli/app.py

#!/usr/bin/env python3
"""
REST API for YouTube CLI application
"""

from flask import Flask, request, jsonify
from youtube_cli.main import YouTubeCLI
import json
import os
import subprocess
import sys

app = Flask(__name__)

# Initialize YouTube CLI
cli = YouTubeCLI()

def search_youtube_api(query, page=1):
    """Search YouTube and return structured results for API"""
    try:
        # Use yt-dlp directly to get search results
        cmd = [
            "yt-dlp",
            "--flat-playlist",  # Get video info without downloading
            "--dump-single-json",  # Output as JSON single item
            f"--playlist-start={15 * (page - 1) + 1}",
            f"--playlist-end={15 * page}",
            "--no-warnings",
            "--no-progress",
            "--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            f"ytsearch{15 * page}:{query}",
        ]

        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        
        if result.returncode != 0:
            return {'error': f'Error searching videos: {result.stderr}'}

        # Parse JSON output
        import json
        
        try:
            data = json.loads(result.stdout.strip())
        except json.JSONDecodeError as e:
            return {'error': f'Error parsing search results: {e}'}

        # Process videos into our format
        videos = []
        
        if isinstance(data, list):
            videos_data = data
        elif "entries" in data:
            videos_data = data["entries"]
        else:
            videos_data = [data]

        for entry in videos_data:
            if not entry:
                continue

            title = entry.get("title", "Unknown Title")
            author = entry.get("uploader", "Unknown Author")
            duration = entry.get("duration", 0)
            url = entry.get("url", "") or entry.get("webpage_url", "")
            view_count = entry.get("view_count", None)
            playlist_title = entry.get("playlist_title", "")

            # Format duration
            length = format_duration(duration)

            # Check if this is a short video
            is_short = "/shorts/" in url or "/shorts" in url

            # Check if this is a playlist (look for playlist-specific attributes)
            is_playlist = "playlist" in url.lower() or "list=" in url

            # Validate URL before adding to videos list
            if not url or url.strip() == "":
                continue  # Skip videos with invalid/missing URLs

            # Create video object
            videos.append(
                {
                    "title": title,
                    "author": author,
                    "length": length,
                    "url": url,
                    "is_short": is_short,
                    "is_playlist": is_playlist,
                    "id": entry.get("id", ""),
                    "thumbnail": entry.get("thumbnail", ""),
                    "view_count": view_count,
                }
            )

        return {
            'query': query,
            'page': page,
            'videos': videos,
            'total': len(videos)
        }
        
    except Exception as e:
        return {'error': str(e)}

def format_duration(seconds):
    """Convert seconds to MM:SS or HH:MM:SS format."""
    if not seconds:
        return "0:00"

    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = int(seconds % 60)

    if hours > 0:
        return f"{hours}:{minutes:02d}:{secs:02d}"
    else:
        return f"{minutes}:{secs:02d}"

@app.route('/search', methods=['GET'])
def search_videos():
    """Search YouTube videos"""
    query = request.args.get('q', '')
    page = int(request.args.get('page', 1))
    
    if not query:
        return jsonify({'error': 'Query parameter "q" is required'}), 400
    
    try:
        result = search_youtube_api(query, page)
        if 'error' in result:
            return jsonify(result), 500
        return jsonify(result)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/download', methods=['POST'])
def download_video():
    """Download a video by URL"""
    data = request.get_json()
    url = data.get('url', '')
    
    if not url:
        return jsonify({'error': 'URL is required'}), 400
    
    try:
        # For now, return a placeholder response indicating download would start
        # In a real implementation, this would call the actual download functionality
        return jsonify({
            'status': 'download_started',
            'url': url,
            'message': 'Download process initiated (not implemented in this demo)'
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({'status': 'healthy', 'service': 'youtube-cli-api'})

@app.route('/version', methods=['GET'])
def get_version():
    """Get API version"""
    return jsonify({'version': '1.0.0'})

@app.route('/capabilities', methods=['GET'])
def get_capabilities():
    """MCP capabilities endpoint"""
    return jsonify({
        "name": "YouTube CLI API",
        "version": "1.0.0",
        "description": "YouTube CLI API for searching and downloading videos",
        "endpoints": [
            {
                "path": "/search",
                "method": "GET",
                "description": "Search YouTube videos"
            },
            {
                "path": "/download",
                "method": "POST",
                "description": "Download a video by URL"
            },
            {
                "path": "/health",
                "method": "GET",
                "description": "Health check endpoint"
            },
            {
                "path": "/version",
                "method": "GET",
                "description": "Get API version"
            },
            {
                "path": "/capabilities",
                "method": "GET",
                "description": "MCP capabilities endpoint"
            },
            {
                "path": "/openapi.json",
                "method": "GET",
                "description": "OpenAPI specification"
            }
        ],
        "features": [
            "Video search",
            "Video download",
            "Health monitoring",
            "Version information",
            "MCP compliance"
        ]
    })

@app.route('/openapi.json', methods=['GET'])
def get_openapi():
    """Serve the OpenAPI specification file"""
    try:
        # Read the openapi.json file from the filesystem
        with open('openapi.json', 'r') as f:
            import json
            spec = json.load(f)
        return jsonify(spec)
    except Exception as e:
        return jsonify({"error": "OpenAPI specification not found"}), 404

if __name__ == '__main__':
    # Fix the port issue by using a different port
    app.run(host='0.0.0.0', port=4096, debug=True)