import yt_dlp import os import asyncio # Buat folder penampungan sementara jika belum ada DOWNLOAD_DIR = "downloads" os.makedirs(DOWNLOAD_DIR, exist_ok=True) def generate_netscape_cookies(): """ Mengubah data tabel cookie (copy-paste dari tab Application browser) menjadi format baku Netscape HTTP Cookie agar bisa dibaca yt-dlp. """ if not os.path.exists("raw_cookie.txt"): print("⚠️ raw_cookie.txt tidak ditemukan. Lanjut tanpa cookies.") return False try: with open("raw_cookie.txt", "r", encoding="utf-8") as f: lines = f.readlines() with open("cookies.txt", "w", encoding="utf-8") as f: f.write("# Netscape HTTP Cookie File\n") f.write("# This is a generated file! Do not edit.\n\n") for line in lines: # Pisahkan kolom berdasarkan tombol Tab (\t) bawaan copy-paste tabel parts = line.split('\t') # Pastikan baris tersebut valid (minimal punya Nama dan Value) if len(parts) >= 2: name = parts[0].strip() value = parts[1].strip() # Abaikan jika yang ter-copy adalah Header tabelnya ("Name", "Value") if name.lower() == "name" or not name: continue # Format baku Netscape: domain, subdomains, path, secure, expiration, name, value # Kita paksa domainnya .youtube.com agar yt-dlp bisa membacanya f.write(f".youtube.com\tTRUE\t/\tTRUE\t2147483647\t{name}\t{value}\n") print("✅ Cookies berhasil di-generate dari raw_cookie.txt") return True except Exception as e: print(f"❌ Error saat memproses cookies: {e}") return False def search_and_download(query_or_url: str): """ Fungsi utama untuk mencari dan mengunduh audio dari YouTube. Berjalan secara sinkronus. """ # 1. Konversi cookie setiap kali mau download has_cookies = generate_netscape_cookies() # 2. Pengaturan yt-dlp ydl_opts = { 'format': 'bestaudio/best/m4a/mp3', 'outtmpl': f'{DOWNLOAD_DIR}/%(id)s.%(ext)s', 'max_filesize': 45000000, # Batas aman Telegram 45MB 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '128', }], 'quiet': True, 'noplaylist': True, 'default_search': 'ytsearch1', 'js_runtimes': {'node': {}}, # 'extractor_args': {'youtube': ['player_client=android,ios']}, } # 3. Sisipkan cookie jika berhasil digenerate if has_cookies and os.path.exists("cookies.txt"): ydl_opts['cookiefile'] = 'cookies.txt' try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: # Mulai proses ekstraksi & download info = ydl.extract_info(query_or_url, download=True) # Jika hasil berupa playlist/pencarian, ambil item pertama if 'entries' in info: info = info['entries'][0] video_id = info.get('id') title = info.get('title') duration = info.get('duration') file_path = f"{DOWNLOAD_DIR}/{video_id}.mp3" return { "status": "success", "video_id": video_id, "title": title, "duration": duration, "file_path": file_path if os.path.exists(file_path) else None } except Exception as e: return {"status": "error", "message": str(e)} async def process_youtube_request(query: str): """ Bungkus (Wrapper) Async agar bot Telegram tidak nge-hang saat menunggu proses download selesai. """ return await asyncio.to_thread(search_and_download, query) # ========================================== # FITUR TAMBAHAN: YOUTUBE MIX RECOMMENDATION # ========================================== def fetch_recommendations_sync(video_id: str, limit: int = 3): """ Mengambil rekomendasi lagu dari YouTube Mix algoritma (Vibes/Genre serupa) """ # Gunakan settingan ringan karena kita cuma mau ambil teks judul, BUKAN download ydl_opts_recs = { 'quiet': True, 'extract_flat': True, 'playlistend': limit + 1, # +1 karena urutan pertama biasanya lagu aslinya 'extractor_args': {'youtube': ['player_client=android,ios']}, } # Sisipkan cookie jika ada if os.path.exists("cookies.txt"): ydl_opts_recs['cookiefile'] = 'cookies.txt' try: with yt_dlp.YoutubeDL(ydl_opts_recs) as ydl: # RD = Playlist YouTube Mix khusus untuk video tersebut mix_url = f"https://www.youtube.com/watch?v={video_id}&list=RD{video_id}" info = ydl.extract_info(mix_url, download=False) recs = [] if 'entries' in info: # Loop dan ambil lagunya (Lewati index 0 karena itu lagu yg sedang diputar) for entry in info['entries'][1:]: if entry and entry.get('id') and entry.get('title'): recs.append({ 'id': entry['id'], 'title': entry['title'] }) return recs except Exception as e: print(f"Gagal mengambil rekomendasi: {e}") return [] async def get_recommendations(video_id: str): """Bungkus Async agar bot tidak hang""" return await asyncio.to_thread(fetch_recommendations_sync, video_id)