import asyncio import os import time import logging import re from aiogram import Bot, Dispatcher, types, BaseMiddleware, F from aiogram.filters import Command from aiogram.types import FSInputFile, InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery from aiogram.client.default import DefaultBotProperties from aiogram.client.session.aiohttp import AiohttpSession from dotenv import load_dotenv from apscheduler.schedulers.asyncio import AsyncIOScheduler from yt_engine import process_youtube_request, get_recommendations from db_manager import db from s3_manager import upload_audio, download_audio, delete_audio logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) load_dotenv() BOT_TOKEN = os.getenv("BOT_TOKEN") AUTHORIZED_USER_ID = int(os.getenv("AUTHORIZED_USER_ID", 0)) session = AiohttpSession(timeout=300) bot = Bot(token=BOT_TOKEN, session=session, default=DefaultBotProperties(parse_mode=None)) dp = Dispatcher() music_queue = asyncio.Queue() async def queue_worker(): while True: query, chat_id, msg_id = await music_queue.get() try: await process_music_request(query, chat_id, msg_id) except Exception as e: logger.error(f"Worker Error: {e}") finally: music_queue.task_done() class SecurityMiddleware(BaseMiddleware): async def __call__(self, handler, event: types.Message | CallbackQuery, data: dict): if event.from_user.id != AUTHORIZED_USER_ID: return return await handler(event, data) dp.message.middleware(SecurityMiddleware()) dp.callback_query.middleware(SecurityMiddleware()) # Ekstraktor ID YouTube def extract_video_id(query): match = re.search(r"(?:v=|\/)([0-9A-Za-z_-]{11}).*", query) return match.group(1) if match else None @dp.message(Command("start")) async def cmd_start(message: types.Message): await message.answer("🎧 **Music Bot Ready (M4A Turbo)!**\nKirim format: `/play `\nSetelah lagu terkirim, klik rekomendasi di bawahnya.", parse_mode="Markdown") @dp.message(Command("play")) async def cmd_play(message: types.Message): query = message.text.replace("/play", "").strip() if not query: return await message.answer("⚠️ Masukkan judul lagu.") status_msg = await message.answer("⏳ Memproses pesananmu...") await music_queue.put((query, message.chat.id, status_msg.message_id)) @dp.callback_query(F.data.startswith("play_")) async def handle_suggestion_click(callback: CallbackQuery): video_id = callback.data.split("_", 1)[1] await callback.message.edit_reply_markup(reply_markup=None) status_msg = await bot.send_message(callback.message.chat.id, "⏳ Memproses lagu pilihanmu...") query_url = f"https://www.youtube.com/watch?v={video_id}" await music_queue.put((query_url, callback.message.chat.id, status_msg.message_id)) await callback.answer("Siapp! Diproses...") # --- CORE LOGIC (DENGAN SMART CACHE) --- async def process_music_request(query: str, chat_id: int, status_msg_id: int): local_file = None total_start_time = time.time() try: logger.info(f"▶️ MEMULAI PROSES: {query}") # 1. SMART CACHE CHECK (CEK DB DULUAN) video_id = extract_video_id(query) cached_data = None if video_id: t_s3_check = time.time() cached_data = await db.get_cache(video_id) if cached_data: logger.info(f"⚡ CACHE HIT! File sudah ada di MinIO. (Cek butuh {time.time() - t_s3_check:.2f}s)") title = cached_data['title'] s3_object_name = cached_data['s3_object_key'] # Bisa saja file lama berekstensi mp3, atau baru berekstensi m4a ext = s3_object_name.split('.')[-1] local_file = f"downloads/cache_{video_id}.{ext}" await bot.edit_message_text("⚡ Mengambil langsung dari Cache (MinIO)...", chat_id, status_msg_id) await download_audio(s3_object_name, local_file) # 2. JIKA TIDAK ADA DI CACHE, BARU PANGGIL YOUTUBE if not cached_data: await bot.edit_message_text("🔍 Mencari dan mengunduh dari YouTube...", chat_id, status_msg_id) t_yt_start = time.time() yt_result = await process_youtube_request(query) logger.info(f"⏱️ yt-dlp (Tanpa FFmpeg) memakan waktu: {time.time() - t_yt_start:.2f} detik") if yt_result["status"] == "error": return await bot.edit_message_text(f"❌ Gagal: {yt_result['message']}", chat_id, status_msg_id) video_id = yt_result["video_id"] title = yt_result["title"] local_file = yt_result["file_path"] # Simpan sesuai ekstensi aslinya (sekarang biasanya m4a) s3_object_name = f"{video_id}.{yt_result['ext']}" await bot.edit_message_text("☁️ Mengunggah ke MinIO Server...", chat_id, status_msg_id) if local_file and os.path.exists(local_file): t_s3_up = time.time() await upload_audio(local_file, s3_object_name) await db.save_cache(video_id, title, s3_object_name) logger.info(f"⏱️ Upload ke MinIO memakan waktu: {time.time() - t_s3_up:.2f} detik") # 3. KIRIM KE TELEGRAM await bot.edit_message_text("📤 Mengirim file ke Telegram...", chat_id, status_msg_id) if local_file and os.path.exists(local_file): t_tele_start = time.time() audio = FSInputFile(local_file) await bot.send_audio(chat_id=chat_id, audio=audio, title=title, performer="Music Bot", request_timeout=300) logger.info(f"⏱️ Upload ke Telegram memakan waktu: {time.time() - t_tele_start:.2f} detik") if status_msg_id: await bot.delete_message(chat_id, status_msg_id) # 4. SUGGEST 5 LAGU (Judul lebih panjang) t_recs = time.time() recs = await get_recommendations(video_id) logger.info(f"⏱️ Fetch Rekomendasi memakan waktu: {time.time() - t_recs:.2f} detik") if recs: kb = [] for rec in recs[:5]: short_title = rec['title'][:60] + "..." if len(rec['title']) > 60 else rec['title'] kb.append([InlineKeyboardButton(text=f"▶️ {short_title}", callback_data=f"play_{rec['id']}")]) reply_markup = InlineKeyboardMarkup(inline_keyboard=kb) await bot.send_message(chat_id, text=f"💡 **Rekomendasi dari:**\n_{title}_", reply_markup=reply_markup, parse_mode="Markdown") logger.info(f"✅ TOTAL WAKTU E2E: {time.time() - total_start_time:.2f} detik\n{'-'*40}") except Exception as e: logger.error(f"Error: {e}") finally: if local_file and os.path.exists(local_file): os.remove(local_file) async def cleanup_expired_cache(): expired_items = await db.get_expired_cache(days=7) for item in expired_items: await delete_audio(item['s3_object_key']) await db.delete_cache(item['youtube_id']) async def main(): await db.connect() asyncio.create_task(queue_worker()) scheduler = AsyncIOScheduler() scheduler.add_job(cleanup_expired_cache, 'cron', hour=3, minute=0) scheduler.start() logger.info("🚀 Bot Music Turbo Berjalan!") await bot.delete_webhook(drop_pending_updates=True) await dp.start_polling(bot) if __name__ == "__main__": asyncio.run(main())