290 lines
8.3 KiB
Python
Executable File
290 lines
8.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
REST API for YouTube CLI application
|
|
"""
|
|
|
|
import logging
|
|
import subprocess
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, jsonify, request
|
|
|
|
from youtube_cli.main import YouTubeCLI
|
|
|
|
# Configure logging
|
|
LOG_DIR = Path.home() / ".config" / "youtube_cli" / "logs"
|
|
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
|
LOG_FILE = LOG_DIR / "app.log"
|
|
|
|
# Use RotatingFileHandler for log rotation (10MB, 5 backups)
|
|
file_handler = RotatingFileHandler(
|
|
LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5
|
|
)
|
|
file_handler.setLevel(logging.DEBUG)
|
|
file_handler.setFormatter(
|
|
logging.Formatter(
|
|
"%(asctime)s | %(name)s | %(levelname)s | %(message)s",
|
|
"%Y-%m-%d %H:%M:%S",
|
|
)
|
|
)
|
|
|
|
# Create console handler
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.INFO)
|
|
console_handler.setFormatter(logging.Formatter("%(message)s"))
|
|
|
|
# Configure root logger
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
handlers=[
|
|
file_handler,
|
|
console_handler,
|
|
],
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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)
|
|
|
|
# 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
|
|
# Try multiple locations to handle Docker vs local execution
|
|
import json
|
|
import os
|
|
|
|
# Check if we're in Docker (working directory is /app)
|
|
current_dir = os.getcwd()
|
|
if current_dir == "/app":
|
|
# In Docker, the file should be in /app
|
|
file_path = "/app/openapi.json"
|
|
else:
|
|
# Local execution
|
|
file_path = "openapi.json"
|
|
|
|
with open(file_path, "r") as f:
|
|
spec = json.load(f)
|
|
return jsonify(spec)
|
|
except Exception as e:
|
|
return jsonify(
|
|
{"error": f"OpenAPI specification not found: {str(e)}"}
|
|
), 404
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Fix the port issue by using a different port
|
|
app.run(host="0.0.0.0", port=4096, debug=True)
|