Rewrite the entire codebase to work with yt-dlp

Author: DevHanza

README.md

@@ -51,21 +51,22 @@ But then you won't be able to **upload files up to 2000 MB** and get these [feat
 
 Read the instructions on [eternnoir/pyTelegramBotAPI](https://github.com/eternnoir/pyTelegramBotAPI/#using-local-bot-api-sever) and [tdlib/telegram-bot-api](https://github.com/tdlib/telegram-bot-api) for more information.
 
-
 #### Run your bot
 
 Open a new "[screen](https://www.geeksforgeeks.org/screen-command-in-linux-with-examples/)" or tab on your terminal. Then run:
+
 ```bash
 python bot.py
 ```
 
 **both the script & api server should run at the same time order to work.**
 
+## Dependencies
+
+- [pytelegramBotAPI](https://github.com/eternnoir/pyTelegramBotAPI/)
+- [python-dotenv](https://pypi.org/project/python-dotenv/)
+- [yt-dlp](https://github.com/yt-dlp/yt-dlp)
 
 ## Disclaimer
 
 This repository is intended for educational and personal use only. The use of this repository for any commercial or illegal purposes is strictly prohibited. The repository owner does not endorse or encourage the downloading or sharing of copyrighted material without permission. The repository owner is not responsible for any misuse of the software or any legal consequences that may arise from such misuse
-
-- **APIs : [y2mate-api](https://github.com/Simatwa/y2mate-api/) , [pytelegramBotAPI](https://github.com/eternnoir/pyTelegramBotAPI/)**
-- **Contact for issues : [@dev00111](https://t.me/dev00111)**
-

bot.py

@@ -1,69 +1,69 @@
-import os
-import telebot
+import os, re
 import threading
+import telebot
 
-from modules import checker, myqueues 
+from dotenv import load_dotenv
 
-from dotenv import load_dotenv 
 load_dotenv()
 
+from modules.checker import youtube_regex
+from modules.downloader import download_video
+
 TOKEN = os.getenv("BOT_API_KEY")
 
 bot = telebot.TeleBot(TOKEN, parse_mode="HTML")
-                      
-@bot.message_handler(commands=['start'])
+
+
+# '/start' command reply
+@bot.message_handler(commands=["start"])
 def send_welcome(message):
     bot.reply_to(
-        message, "Hello, I'm a <b>Simple Youtube Downloader!👋</b>\n\nTo get started, just type the /help command.")
+        message,
+        "Hello, I'm a <b>Simple Youtube Downloader!👋</b>\n\nTo get started, just type the /help command.",
+    )
+
 
-@bot.message_handler(commands=['help'])
+# '/help' command reply
+@bot.message_handler(commands=["help"])
 def send_help(message):
     bot.reply_to(
         message,
         """
-        <b>Just send your youtube link and select the video quality.</b> 😉
-  <i>
-  Developer: @dev00111
-  Source: <a href="https://github.com/hansanaD/TelegramYTDLBot">TelegramYTDLBot</a></i>
-        """, disable_web_page_preview=True,)
-        
-    
+        <b>Just send me your video link.</b> ▶️
+
+<i>Source: <a href="https://github.com/hansanaD/TelegramYTDLBot">TelegramYTDLBot</a> by <a href="https://github.com/DevHanza/">DevHanza</a></i>
+        """,
+        disable_web_page_preview=True,
+    )
 
+
+# Youtube Link Listener
 @bot.message_handler(func=lambda m: True)
-def link_check(message):
-    checker.linkCheck(bot=bot, message=message)
-    # print(checker.videoURL)
-
-# Callback handler for # getVidInfo() 
-@bot.callback_query_handler(func=lambda call: [call.data == item for item in checker.showList])
-def callback_query(call):
-
-    data = call.data.split("#")
-    receivedData = data[0]
-    videoURL = data[1]
-    # print(receivedData)
-    
-    bot.answer_callback_query(call.id, f"Selected {receivedData} to download.")
-    bot.delete_message(chat_id=call.message.chat.id, message_id=call.message.message_id)
-
-    myqueues.download_queue.put((call.message, videoURL, receivedData))
-    queue_position = myqueues.download_queue.qsize()
-
-    
-    if queue_position == 0 & 1:
-        bot.send_message(call.message.chat.id, f"Download has been added to the queue.")
-    else:
-        bot.send_message(call.message.chat.id, f"Download has been added to the queue at #{queue_position}.")
-
-
-    # downloader.download(bot=bot, message=call.message, userInput=receivedData, videoURL=checker.videoURL)
-    # bot.send_message(call.message.chat.id, f"{videoURL} \n{receivedData} : Download Triggered!")
-            
-# message, videoURL, receivedData
-    
-download_thread = threading.Thread(target=myqueues.download_worker, args=(bot, myqueues.download_queue))
-download_thread.daemon = True
-download_thread.start()
+def on_yt_link(message):
+    # run download in background so bot stays responsive
+    threading.Thread(target=yt_link_handler, args=(message,)).start()
+
+
+def yt_link_handler(message):
+
+    matches = re.findall(youtube_regex, message.text)
+
+    if matches:
+
+        url = matches[0]
+        status_msg = bot.reply_to(message, f"Starting to download..")
+
+        try:
+            file_path = download_video(url, bot, message.chat.id, status_msg.message_id)
+
+            with open(file_path, "rb") as file:
+                bot.send_video(message.chat.id, file)
+
+            os.remove(file_path)
+
+        except Exception as e:
+            bot.reply_to(message, f"❌ Error: {e}")
+
 
 print("TelegramYTDLBot is running..\n")
 bot.infinity_polling()

modules/checker.py

@@ -1,91 +1,2 @@
-from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
-import re
-from y2mate_api import Handler
-
-
-def linkCheck(bot, message):
-
-    linkFilter = re.compile(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+')
-    userLinks = re.findall(linkFilter, message.text)
-
-    yt_link = []
-    for link in userLinks:
-        if 'youtube.com' in link or 'youtu.be' in link:
-            yt_link.append(link)
-
-    if yt_link:
-        # bot.reply_to(message, "YouTube links found.")
-
-        # global videoURL
-        # global ytApi
-
-        videoURL = yt_link[0]
-        
-
-        qualityChecker(bot=bot, message=message, videoURL=videoURL)
-
-    else:
-        bot.reply_to(message, "No YouTube links found!")
-
-
-def qualityChecker(bot, message, videoURL):
-
-    qualityCheckerMsg = bot.reply_to(message, "Looking for Available Qualities..🔎")
-
-    ytApi = Handler(videoURL)
-
-    q_list = ['4k', '1080p', '720p', '480p', '360p', '240p']
-    # q_list.reverse()
-
-    urlList = []
-
-    def getVidInfo(r):
-        for video_metadata in ytApi.run(quality=r):
-        
-            q = video_metadata.get("q")
-            dlink = video_metadata.get("dlink")
-            size = video_metadata.get("size")
-            
-            if dlink == None:
-                pass
-            else:
-                urlList.append([q, size, dlink])
-                # print(r, " fetched")
-                
-    # Iterate over q_list to check if res quality exist on that video
-    for r in q_list:
-        getVidInfo(r)
-
-    # print(urlList)
-
-    # Create a new list to show
-    global showList
-    showList = {}
-    for count, item in enumerate(urlList, 1):
-        del item[2] # Remove dlink from list
-        q = item[0]
-        # print(i)
-        size = item[1] 
-        showList.update( { count: { "q":q, "size": size }} )
-    
-    # print(showList)
-
-
-    # Add Inline Buttons to get user input
-
-    def gen_markup():
-        markup = InlineKeyboardMarkup() 
-        for value in showList.values(): 
-            callbackData = f"{ value["q"] }#{ videoURL }"
-            button = InlineKeyboardButton(text=f"{value['q']} ({value['size']})", callback_data=callbackData)
-            markup.add(button)
-        return markup
-    
-
-    bot.delete_message(qualityCheckerMsg.chat.id, qualityCheckerMsg.message_id)
-
-    bot.reply_to(message=message, text="Choose a stream:", reply_markup=gen_markup())
-
-
-
-
+# Simple YouTube URL regex
+youtube_regex = r"(https?://(?:www\.)?(?:youtube\.com/watch\?v=|youtu\.be/)[\w\-]+)"

modules/downloader.py

@@ -0,0 +1,23 @@
+from yt_dlp import YoutubeDL
+
+from modules.progress import create_progress_bar
+
+
+def download_video(url: str, bot, chat_id: int, message_id: int) -> str:
+    options = {
+        "outtmpl": "downloads/%(title)s.%(ext)s",
+        "format": "bestvideo+bestaudio/best",
+        "noplaylist": True,
+        "quiet": False,
+        "merge_output_format": "mp4",
+        "progress_hooks": [create_progress_bar(bot, chat_id, message_id)],
+    }
+
+    try:
+        with YoutubeDL(options) as ydl:
+            info = ydl.extract_info(url, download=True)
+            filename = ydl.prepare_filename(info)
+        return filename
+
+    except Exception as e:
+        print(f"Error downloading video: {e} ❌")

modules/myqueues.py

@@ -0,0 +1,45 @@
+import time
+
+
+def progress_bar(percent, length=20):
+    filled = int(length * percent // 100)
+    return "█" * filled + "░" * (length - filled)
+
+
+def create_progress_bar(bot, chat_id, message_id):
+    """
+    Returns a yt-dlp progress hook function
+    that updates a Telegram message.
+    """
+    last_update_time = 0
+
+    def hook(d):
+        nonlocal last_update_time
+
+        if d["status"] == "downloading":
+            total = d.get("total_bytes") or d.get("total_bytes_estimate")
+            downloaded = d.get("downloaded_bytes", 0)
+
+            if total:
+                percent = downloaded / total * 100
+
+                # throttle edits to avoid Telegram limits
+                if time.time() - last_update_time > 2:
+                    bar = progress_bar(percent)
+
+                    text = f"Downloading...\n" f"[{bar}] {percent:.1f}%"
+
+                    try:
+                        bot.edit_message_text(text, chat_id, message_id)
+                    except Exception:
+                        pass
+
+                    last_update_time = time.time()
+
+        elif d["status"] == "finished":
+            try:
+                bot.edit_message_text("Uploading...", chat_id, message_id)
+            except Exception:
+                pass
+
+    return hook