import os import hashlib import subprocess import time import concurrent.futures from yt_dlp import YoutubeDL from http.server import SimpleHTTPRequestHandler, HTTPServer import urllib import json pool = concurrent.futures.ThreadPoolExecutor(max_workers=2) def extract_youtube_id(url): """Extract the YouTube video ID from the URL.""" if "v=" in url: return url.split("v=")[1].split("&")[0] return url def get_video_info(youtube_url, cache_dir="cache"): """Fetch video information using yt-dlp and cache it for a day.""" video_id = extract_youtube_id(youtube_url) info_cache_path = os.path.join(cache_dir, f"{video_id}_info.json") # Check cache if os.path.exists(info_cache_path): with open(info_cache_path, 'r') as cache_file: cached_data = json.load(cache_file) if time.time() < cached_data.get("expires", 0): print(f"Using cached video info for {video_id}.") return cached_data["info"] # Fetch info if not cached or expired os.makedirs(cache_dir, exist_ok=True) ydl_opts = { 'cachedir': cache_dir, # Enable caching 'quiet': False, # Suppress unnecessary output } with YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(youtube_url, download=False) print(info) video_info = { "title": info.get("title", "Unknown Title"), "uploader": info.get("uploader", "Unknown Uploader"), "duration": info.get("duration", 0) # Duration in seconds } # Cache the info with open(info_cache_path, 'w') as cache_file: json.dump({ "info": video_info, "expires": time.time() + 86400 # Cache expires in 1 day }, cache_file) return video_info def download_audio_with_yt_dlp(youtube_url, cache_dir="cache"): """Download audio from YouTube using yt-dlp and cache it.""" video_id = extract_youtube_id(youtube_url) output_path = os.path.join(cache_dir, f"{video_id}") if os.path.exists(output_path): print(f"Audio already cached: {output_path}") return output_path + ".wav" os.makedirs(cache_dir, exist_ok=True) ydl_opts = { 'format': 'bestaudio/best', 'outtmpl': output_path, 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'wav', }], } with YoutubeDL(ydl_opts) as ydl: ydl.download([youtube_url]) return output_path + ".wav" def split_file(file_path, chunk_size=1024 * 1024 * 8, cache_dir="cache"): """Split a WAV file into chunks and cache them.""" video_id = os.path.splitext(os.path.basename(file_path))[0] chunk_dir = os.path.join(cache_dir, video_id) if os.path.exists(chunk_dir): print(f"Chunks already cached in: {chunk_dir}") return chunk_dir os.makedirs(chunk_dir, exist_ok=True) with open(file_path, 'rb') as file: chunk_number = 0 while chunk := file.read(chunk_size): chunk_name = os.path.join(chunk_dir, f'chunk_{chunk_number}') with open(chunk_name, 'wb') as chunk_file: chunk_file.write(chunk) chunk_number += 1 print(f"Split into {chunk_number} chunks and cached in: {chunk_dir}") return chunk_dir def get_playlist_videos(playlist_url, cache_dir="cache"): """Fetch all video IDs, playlist name, and creator from a playlist.""" ydl_opts = { 'quiet': True, 'extract_flat': True, # Only extract metadata } with YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(playlist_url, download=False) if 'entries' in info: playlist_name = info.get('title', 'Unknown Playlist') creator = info.get('uploader', 'Unknown Creator') video_ids = [entry['id'] for entry in info['entries']] return video_ids, playlist_name, creator return [], 'Unknown Playlist', 'Unknown Creator' def process_playlist(playlist_url, cache_dir="cache"): """Process all videos in the playlist.""" video_ids = get_playlist_videos(playlist_url, cache_dir) for video_id in video_ids: youtube_url = f"https://www.youtube.com/watch?v={video_id}" video_info = get_video_info(youtube_url, cache_dir) wav_file = download_audio_with_yt_dlp(youtube_url, cache_dir) split_file(wav_file, cache_dir) class CustomHTTPRequestHandler(SimpleHTTPRequestHandler): def do_GET(self): """Handle GET requests.""" parsed_path = urllib.parse.urlparse(self.path) query_params = urllib.parse.parse_qs(parsed_path.query) if parsed_path.path == "/download": youtube_id = query_params.get("id", [None])[0] if youtube_id: youtube_url = f"https://www.youtube.com/watch?v={youtube_id}" cache_directory = "cache" video_info = get_video_info(youtube_url) wav_file = download_audio_with_yt_dlp(youtube_url, cache_dir=cache_directory) split_file(wav_file, cache_dir=cache_directory) self.send_response(200) self.send_header("Content-type", "application/json") self.end_headers() response = { "status": "success", "video_id": youtube_id, "title": video_info["title"], "uploader": video_info["uploader"], "duration": video_info["duration"], "message": "Processing completed and chunks are available in cache." } self.wfile.write(json.dumps(response).encode("utf-8")) else: self.send_response(400) self.end_headers() self.wfile.write(b"Missing YouTube ID") elif parsed_path.path == "/playlist": youtube_id = query_params.get("id", [None])[0] if youtube_id: youtube_url = f"https://www.youtube.com/playlist?list={youtube_id}" playlist, title, creator = get_playlist_videos(youtube_url) self.send_response(200) self.send_header("Content-type", "application/json") self.end_headers() response = { "status": "success", "playlist": playlist, "title": title, "creator": creator, "message": "Playlist info" } self.wfile.write(json.dumps(response).encode("utf-8")) pool.submit(process_playlist, youtube_url) else: self.send_response(400) self.end_headers() self.wfile.write(b"Missing YouTube ID") else: super().do_GET() if __name__ == "__main__": server_address = ('', 8000) httpd = HTTPServer(server_address, CustomHTTPRequestHandler) print("Starting server on port 8000...") httpd.serve_forever()