Update app.py

app.py

@@ -1,11 +1,44 @@
-import gradio as gr
+from fastapi import FastAPI, HTTPException, Query
+from fastapi.responses import FileResponse
+from pydantic import BaseModel
 from pytubefix import YouTube
 from pytubefix.cli import on_progress
 import os
+import csv
+import uuid
+from datetime import datetime
+from typing import List, Optional
 import re
 
-def is_youtube_url(url):
-    """Check if the provided string is a valid YouTube URL"""
+app = FastAPI(title="YouTube Download API", version="1.0.0")
+
+# Database file
+DB_FILE = "download_logs.csv"
+DOWNLOAD_DIR = "downloads"
+
+# Initialize CSV database
+def init_database():
+    """Initialize CSV database if it doesn't exist"""
+    os.makedirs(DOWNLOAD_DIR, exist_ok=True)
+    
+    if not os.path.exists(DB_FILE):
+        with open(DB_FILE, 'w', newline='', encoding='utf-8') as f:
+            writer = csv.writer(f)
+            writer.writerow(['id', 'username', 'youtube_url', 'choice_type', 'quality', 'timestamp', 'status'])
+
+def log_download(username: str, youtube_url: str, choice_type: str, quality: str, status: str):
+    """Log download to CSV database"""
+    download_id = str(uuid.uuid4())
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    
+    with open(DB_FILE, 'a', newline='', encoding='utf-8') as f:
+        writer = csv.writer(f)
+        writer.writerow([download_id, username, youtube_url, choice_type, quality, timestamp, status])
+    
+    return download_id
+
+def is_youtube_url(url: str) -> bool:
+    """Validate YouTube URL"""
     if not url or not isinstance(url, str):
         return False
     
@@ -16,106 +49,267 @@ def is_youtube_url(url):
     )
     return bool(re.match(youtube_regex, url))
 
-def download_youtube_video(url):
-    """Download YouTube video and return the file path"""
+# Pydantic models
+class VideoChoice(BaseModel):
+    id: str
+    type: str  # "video" or "audio"
+    quality: str
+    resolution: Optional[str] = None
+    fps: Optional[int] = None
+    file_size: Optional[str] = None
+    mime_type: str
+
+class VideoInfoResponse(BaseModel):
+    title: str
+    duration: str
+    thumbnail: str
+    author: str
+    views: int
+    choices: List[VideoChoice]
+
+class DownloadRequest(BaseModel):
+    youtube_url: str
+    choice_id: str
+    username: str
+
+class DownloadResponse(BaseModel):
+    id: str
+    status: str
+    message: str
+    download_url: Optional[str] = None
+
+# API Endpoints
+
+@app.get("/")
+def read_root():
+    """Root endpoint with API documentation"""
+    return {
+        "message": "YouTube Download API",
+        "version": "1.0.0",
+        "endpoints": {
+            "/get-choices": "GET - Get available download options for a YouTube video",
+            "/download": "POST - Download video with selected options",
+            "/logs": "GET - View download logs (admin)",
+            "/health": "GET - Health check"
+        }
+    }
+
+@app.get("/get-choices", response_model=VideoInfoResponse)
+def get_video_choices(url: str = Query(..., description="YouTube video URL")):
+    """
+    Get all available download choices for a YouTube video
     
-    if not url or not url.strip():
-        return None, "Please provide a YouTube URL"
+    Returns:
+    - Video information (title, duration, thumbnail, etc.)
+    - Available video qualities (MP4)
+    - Available audio qualities (MP3)
+    """
     
     if not is_youtube_url(url):
-        return None, "Invalid YouTube URL. Please provide a valid YouTube link."
+        raise HTTPException(status_code=400, detail="Invalid YouTube URL")
     
     try:
-        # Create output directory
-        output_dir = "downloads"
-        os.makedirs(output_dir, exist_ok=True)
-        
-        # Create YouTube object
         yt = YouTube(url, on_progress_callback=on_progress)
         
         # Get video info
-        video_title = yt.title
+        title = yt.title
         duration = yt.length
-        duration_str = f"{duration // 60}:{duration % 60:02d}" if duration else "Unknown"
+        duration_str = f"{duration // 60}:{duration % 60:02d}"
+        thumbnail = yt.thumbnail_url
+        author = yt.author
+        views = yt.views
+        
+        choices = []
+        
+        # Get video streams (MP4)
+        video_streams = yt.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc()
+        
+        for idx, stream in enumerate(video_streams):
+            file_size_mb = stream.filesize / (1024 * 1024) if stream.filesize else 0
+            choices.append(VideoChoice(
+                id=f"video_{idx}_{stream.itag}",
+                type="video",
+                quality=f"{stream.resolution} - {stream.fps}fps",
+                resolution=stream.resolution,
+                fps=stream.fps,
+                file_size=f"{file_size_mb:.2f} MB",
+                mime_type=stream.mime_type
+            ))
+        
+        # Get audio-only streams (MP3)
+        audio_streams = yt.streams.filter(only_audio=True).order_by('abr').desc()
+        
+        for idx, stream in enumerate(audio_streams):
+            file_size_mb = stream.filesize / (1024 * 1024) if stream.filesize else 0
+            choices.append(VideoChoice(
+                id=f"audio_{idx}_{stream.itag}",
+                type="audio",
+                quality=stream.abr if stream.abr else "Unknown",
+                file_size=f"{file_size_mb:.2f} MB",
+                mime_type=stream.mime_type
+            ))
+        
+        return VideoInfoResponse(
+            title=title,
+            duration=duration_str,
+            thumbnail=thumbnail,
+            author=author,
+            views=views,
+            choices=choices
+        )
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error fetching video info: {str(e)}")
+
+@app.post("/download", response_model=DownloadResponse)
+def download_video(request: DownloadRequest):
+    """
+    Download video with selected quality/format
+    
+    Body:
+    - youtube_url: YouTube video URL
+    - choice_id: Choice ID from /get-choices endpoint
+    - username: Username for logging
+    """
+    
+    if not is_youtube_url(request.youtube_url):
+        raise HTTPException(status_code=400, detail="Invalid YouTube URL")
+    
+    if not request.username:
+        raise HTTPException(status_code=400, detail="Username is required")
+    
+    try:
+        yt = YouTube(request.youtube_url, on_progress_callback=on_progress)
         
-        # Get highest resolution progressive stream (video + audio combined)
-        stream = yt.streams.get_highest_resolution()
+        # Parse choice_id to get itag
+        choice_parts = request.choice_id.split('_')
+        if len(choice_parts) < 3:
+            raise HTTPException(status_code=400, detail="Invalid choice_id format")
+        
+        choice_type = choice_parts[0]  # "video" or "audio"
+        itag = int(choice_parts[2])
+        
+        # Get the stream by itag
+        stream = yt.streams.get_by_itag(itag)
         
         if not stream:
-            return None, "❌ No suitable stream found for this video"
+            raise HTTPException(status_code=404, detail="Selected stream not found")
+        
+        # Download the file
+        output_path = stream.download(output_path=DOWNLOAD_DIR)
+        
+        # If audio, convert extension to mp3 (rename)
+        if choice_type == "audio":
+            base_name = os.path.splitext(output_path)[0]
+            new_path = f"{base_name}.mp3"
+            os.rename(output_path, new_path)
+            output_path = new_path
+        
+        # Log to database
+        quality = f"{stream.resolution}" if choice_type == "video" else f"{stream.abr}"
+        download_id = log_download(
+            username=request.username,
+            youtube_url=request.youtube_url,
+            choice_type=choice_type,
+            quality=quality,
+            status="success"
+        )
         
-        # Download the video
-        output_path = stream.download(output_path=output_dir)
+        # Get filename for download URL
+        filename = os.path.basename(output_path)
+        
+        return DownloadResponse(
+            id=download_id,
+            status="success",
+            message="Download completed successfully",
+            download_url=f"/download-file/{filename}"
+        )
         
-        if os.path.exists(output_path):
-            success_msg = f"✅ Downloaded Successfully!\n\nTitle: {video_title}\nDuration: {duration_str}\nResolution: {stream.resolution}"
-            return output_path, success_msg
-        else:
-            return None, "❌ Error: Video file not found after download"
-                
     except Exception as e:
-        error_msg = str(e)
-        if "unavailable" in error_msg.lower():
-            return None, "❌ This video is unavailable or private"
-        elif "regex" in error_msg.lower():
-            return None, "❌ YouTube changed their format. Please try updating pytubefix"
-        else:
-            return None, f"❌ Error: {error_msg}"
-
-def chat_youtube(message, history):
-    """Process chat messages and download YouTube videos"""
-    
-    # Handle None or empty messages
-    if not message:
-        return "Please send me a YouTube video URL!"
-    
-    # Convert message to string
-    message_text = str(message).strip()
-    
-    # Check if message contains a YouTube URL
-    if is_youtube_url(message_text):
-        video_path, status_msg = download_youtube_video(message_text)
-        return status_msg
-    else:
-        # Regular chat response
-        return """👋 Hi! I'm your YouTube Video Downloader assistant.
-
-**How to use:**
-1. Send me a YouTube video URL
-2. I'll download the video for you
-3. You can then save it to your device
-
-**Example URLs:**
-- https://www.youtube.com/watch?v=jNQXAC9IVRw
-- https://youtu.be/jNQXAC9IVRw
-
-Just paste any YouTube link to get started!"""
-
-# Create Gradio interface
-with gr.Blocks(title="YouTube Video Downloader") as demo:
-    gr.Markdown("# 🎥 YouTube Video Downloader")
-    gr.Markdown("Send a YouTube video URL and download videos easily!")
-    
-    gr.ChatInterface(
-        fn=chat_youtube,
-        examples=[
-            "How does this work?",
-            "https://www.youtube.com/watch?v=jNQXAC9IVRw"
-        ],
-        type="messages"
+        # Log failed download
+        log_download(
+            username=request.username,
+            youtube_url=request.youtube_url,
+            choice_type="unknown",
+            quality="unknown",
+            status=f"failed: {str(e)}"
+        )
+        
+        raise HTTPException(status_code=500, detail=f"Download error: {str(e)}")
+
+@app.get("/download-file/{filename}")
+def download_file(filename: str):
+    """
+    Download the actual file
+    """
+    file_path = os.path.join(DOWNLOAD_DIR, filename)
+    
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="File not found")
+    
+    return FileResponse(
+        path=file_path,
+        filename=filename,
+        media_type='application/octet-stream'
     )
+
+@app.get("/logs")
+def get_logs(limit: int = Query(100, description="Number of logs to retrieve")):
+    """
+    Get download logs from CSV database
+    """
+    if not os.path.exists(DB_FILE):
+        return {"logs": [], "message": "No logs found"}
     
-    gr.Markdown("""
-    ### 📝 Instructions:
-    - Paste any YouTube video URL (short or long format)
-    - The bot will download the video in highest quality
-    - Videos are saved and can be downloaded
+    logs = []
+    with open(DB_FILE, 'r', newline='', encoding='utf-8') as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            logs.append(row)
     
-    ### ⚠️ Note:
-    - Download time depends on video length
-    - Private/unavailable videos cannot be downloaded
-    - Uses pytubefix library for reliable downloads
-    """)
+    # Return last N logs
+    return {
+        "total": len(logs),
+        "logs": logs[-limit:] if len(logs) > limit else logs
+    }
+
+@app.get("/logs/user/{username}")
+def get_user_logs(username: str):
+    """
+    Get download logs for a specific user
+    """
+    if not os.path.exists(DB_FILE):
+        return {"logs": [], "message": "No logs found"}
+    
+    logs = []
+    with open(DB_FILE, 'r', newline='', encoding='utf-8') as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            if row['username'] == username:
+                logs.append(row)
+    
+    return {
+        "username": username,
+        "total": len(logs),
+        "logs": logs
+    }
+
+@app.get("/health")
+def health_check():
+    """Health check endpoint"""
+    return {
+        "status": "healthy",
+        "database": "connected" if os.path.exists(DB_FILE) else "not initialized",
+        "timestamp": datetime.now().isoformat()
+    }
+
+# Initialize database on startup
+@app.on_event("startup")
+def startup_event():
+    init_database()
+    print("✅ Database initialized")
+    print("✅ YouTube Download API is ready!")
 
 if __name__ == "__main__":
-    demo.launch()
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)