Update Devine/platforms/Youtube.py

Devine/platforms/Youtube.py

@@ -1,51 +1,16 @@
 import asyncio
-import glob
 import os
-import random
 import re
+import json
+import httpx  # Make sure to import httpx
 from typing import Union
-
+import yt_dlp
 from pyrogram.enums import MessageEntityType
 from pyrogram.types import Message
 from youtubesearchpython.__future__ import VideosSearch
-from yt_dlp import YoutubeDL
-
-import config
 from Devine.utils.database import is_on_off
 from Devine.utils.formatters import time_to_seconds
-
-def cookies():
-    folder_path = f"{os.getcwd()}/cookies"
-    txt_files = glob.glob(os.path.join(folder_path, "*.txt"))
-    if not txt_files:
-        raise FileNotFoundError("No .txt files found in the specified folder.")
-    cookie_txt_file = random.choice(txt_files)
-    return f"""cookies/{str(cookie_txt_file).split("/")[-1]}"""
-
-
-def get_ytdl_options(ytdl_opts: Union[str, dict, list], commandline: bool = True) -> Union[str, dict, list]:
-    token_data = os.getenv("TOKEN_DATA")
-    
-    if isinstance(ytdl_opts, list):
-        if token_data:
-            ytdl_opts += ["--username" if commandline else "username", "oauth2", "--password" if commandline else "password", "''"]
-        else:
-            ytdl_opts += ["--cookies" if commandline else "cookiefile", cookies()]
-    
-    elif isinstance(ytdl_opts, str):
-        if token_data:
-            ytdl_opts += "--username oauth2 --password '' " if commandline else "username oauth2 password '' "
-        else:
-            ytdl_opts += f"--cookies {cookies()}" if commandline else f"cookiefile {cookies()}"
-    
-    elif isinstance(ytdl_opts, dict):
-        if token_data:
-            ytdl_opts.update({"username": "oauth2", "password": ""})
-        else:
-            ytdl_opts["cookiefile"] = cookies()
-    
-    return ytdl_opts
-
+import tempfile
 
 async def shell_cmd(cmd):
     proc = await asyncio.create_subprocess_shell(
@@ -61,7 +26,6 @@ async def shell_cmd(cmd):
             return errorz.decode("utf-8")
     return out.decode("utf-8")
 
-
 class YouTubeAPI:
     def __init__(self):
         self.base = "https://www.youtube.com/watch?v="
@@ -154,16 +118,12 @@ class YouTubeAPI:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        cmd = [
+        proc = await asyncio.create_subprocess_exec(
             "yt-dlp",
             "-g",
             "-f",
             "best[height<=?720][width<=?1280]",
             f"{link}",
-        ]
-        cmd = get_ytdl_options(cmd)
-        proc = await asyncio.create_subprocess_exec(
-            *cmd,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
         )
@@ -178,11 +138,9 @@ class YouTubeAPI:
             link = self.listbase + link
         if "&" in link:
             link = link.split("&")[0]
-
-        cmd = get_ytdl_options(
+        playlist = await shell_cmd(
             f"yt-dlp -i --get-id --flat-playlist --playlist-end {limit} --skip-download {link}"
         )
-        playlist = await shell_cmd(cmd)
         try:
             result = playlist.split("\n")
             for key in result:
@@ -218,29 +176,24 @@ class YouTubeAPI:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-
-        ytdl_opts = {
-            "quiet": True,
-        }
-        ytdl_opts = get_ytdl_options(ytdl_opts, False)
-
-        ydl = YoutubeDL(ytdl_opts)
+        ytdl_opts = {"quiet": True}
+        ydl = yt_dlp.YoutubeDL(ytdl_opts)
         with ydl:
             formats_available = []
             r = ydl.extract_info(link, download=False)
             for format in r["formats"]:
                 try:
                     str(format["format"])
-                except Exception:
+                except:
                     continue
-                if "dash" not in str(format["format"]).lower():
+                if not "dash" in str(format["format"]).lower():
                     try:
                         format["format"]
                         format["filesize"]
                         format["format_id"]
                         format["ext"]
                         format["format_note"]
-                    except KeyError:
+                    except:
                         continue
                     formats_available.append(
                         {
@@ -272,6 +225,22 @@ class YouTubeAPI:
         thumbnail = result[query_type]["thumbnails"][0]["url"].split("?")[0]
         return title, duration_min, thumbnail, vidid
 
+    async def get_video_info_from_bitflow(self, url: str, video: bool):
+        api_url = "https://bitflow.in/api/youtube"
+        params = {
+            "query": url,
+            "format": "video" if video else "audio",
+            "download": True,
+            "api_key": "1spiderkey2"
+        }
+        
+        async with httpx.AsyncClient() as client:
+            response = await client.get(api_url, params=params, timeout=150)
+            if response.status_code == 200:
+                return response.json()
+            else:
+                return {"status": "error", "message": "Failed to fetch data from Bitflow API."}
+
     async def download(
         self,
         link: str,
@@ -285,49 +254,64 @@ class YouTubeAPI:
     ) -> str:
         if videoid:
             link = self.base + link
+        if "&" in link:
+            link = link.split("&")[0]
         loop = asyncio.get_running_loop()
-
-        def audio_dl():
+        bitflow_info = await self.get_video_info_from_bitflow(link, video)
+        def audio_dl(bitflow_info):
+            temp_dir = tempfile.gettempdir()
+            xyz = os.path.join(temp_dir, f"{bitflow_info['videoid']}.{bitflow_info['ext']}")
+            url = bitflow_info['url']
+            # If the url is not a YouTube link, download directly
+            if not (url.startswith('http') and ('youtube.com' in url or 'youtu.be' in url)):
+                import httpx
+                with httpx.Client() as client:
+                    r = client.get(url)
+                    with open(xyz, "wb") as f:
+                        f.write(r.content)
+                return xyz
             ydl_optssx = {
                 "format": "bestaudio/best",
-                "outtmpl": "downloads/%(id)s.%(ext)s",
+                "outtmpl": xyz,
                 "geo_bypass": True,
                 "nocheckcertificate": True,
                 "quiet": True,
                 "no_warnings": True,
             }
-            ydl_optssx = get_ytdl_options(ydl_optssx, False)
-
-            x = YoutubeDL(ydl_optssx)
-            info = x.extract_info(link, False)
-            xyz = os.path.join("downloads", f"{info['id']}.{info['ext']}")
+            x = yt_dlp.YoutubeDL(ydl_optssx)
             if os.path.exists(xyz):
                 return xyz
-            x.download([link])
+            x.download([url])
             return xyz
-
-        def video_dl():
+        def video_dl(bitflow_info):
+            temp_dir = tempfile.gettempdir()
+            xyz = os.path.join(temp_dir, f"{bitflow_info['videoid']}.{bitflow_info['ext']}")
+            url = bitflow_info['url']
+            # If the url is not a YouTube link, download directly
+            if not (url.startswith('http') and ('youtube.com' in url or 'youtu.be' in url)):
+                import httpx
+                with httpx.Client() as client:
+                    r = client.get(url)
+                    with open(xyz, "wb") as f:
+                        f.write(r.content)
+                return xyz
             ydl_optssx = {
                 "format": "(bestvideo[height<=?720][width<=?1280][ext=mp4])+(bestaudio[ext=m4a])",
-                "outtmpl": "downloads/%(id)s.%(ext)s",
+                "outtmpl": xyz,
                 "geo_bypass": True,
                 "nocheckcertificate": True,
                 "quiet": True,
                 "no_warnings": True,
             }
-            ydl_optssx = get_ytdl_options(ydl_optssx, False)
-
-            x = YoutubeDL(ydl_optssx)
-            info = x.extract_info(link, False)
-            xyz = os.path.join("downloads", f"{info['id']}.{info['ext']}")
+            x = yt_dlp.YoutubeDL(ydl_optssx)
             if os.path.exists(xyz):
                 return xyz
-            x.download([link])
+            x.download([url])
             return xyz
-
         def song_video_dl():
+            temp_dir = tempfile.gettempdir()
+            fpath = os.path.join(temp_dir, f"{title}")
             formats = f"{format_id}+140"
-            fpath = f"downloads/{title}"
             ydl_optssx = {
                 "format": formats,
                 "outtmpl": fpath,
@@ -338,13 +322,11 @@ class YouTubeAPI:
                 "prefer_ffmpeg": True,
                 "merge_output_format": "mp4",
             }
-            ydl_optssx = get_ytdl_options(ydl_optssx, False)
-
-            x = YoutubeDL(ydl_optssx)
+            x = yt_dlp.YoutubeDL(ydl_optssx)
             x.download([link])
-
         def song_audio_dl():
-            fpath = f"downloads/{title}.%(ext)s"
+            temp_dir = tempfile.gettempdir()
+            fpath = os.path.join(temp_dir, f"{title}.%(ext)s")
             ydl_optssx = {
                 "format": format_id,
                 "outtmpl": fpath,
@@ -361,47 +343,22 @@ class YouTubeAPI:
                     }
                 ],
             }
-            ydl_optssx = get_ytdl_options(ydl_optssx, False)
-
-            x = YoutubeDL(ydl_optssx)
+            x = yt_dlp.YoutubeDL(ydl_optssx)
             x.download([link])
-
         if songvideo:
             await loop.run_in_executor(None, song_video_dl)
-            fpath = f"downloads/{title}.mp4"
+            temp_dir = tempfile.gettempdir()
+            fpath = os.path.join(temp_dir, f"{title}.mp4")
             return fpath
         elif songaudio:
             await loop.run_in_executor(None, song_audio_dl)
-            fpath = f"downloads/{title}.mp3"
+            temp_dir = tempfile.gettempdir()
+            fpath = os.path.join(temp_dir, f"{title}.mp3")
             return fpath
         elif video:
-            if await is_on_off(config.YTDOWNLOADER):
-                direct = True
-                downloaded_file = await loop.run_in_executor(None, video_dl)
-            else:
-                command = [
-                    "yt-dlp",
-                    "-g",
-                    "-f",
-                    "best[height<=?720][width<=?1280]",
-                    link,
-                ]
-                command = get_ytdl_options(command)
-                
-                proc = await asyncio.create_subprocess_exec(
-                    *command,
-                    stdout=asyncio.subprocess.PIPE,
-                    stderr=asyncio.subprocess.PIPE,
-                )
-                stdout, stderr = await proc.communicate()
-
-                if stdout:
-                    downloaded_file = stdout.decode().split("\n")[0]
-                    direct = None
-                else:
-                    return
+            direct = True
+            downloaded_file = await loop.run_in_executor(None, video_dl, bitflow_info)
         else:
             direct = True
-            downloaded_file = await loop.run_in_executor(None, audio_dl)
-
-        return downloaded_file, direct
+            downloaded_file = await loop.run_in_executor(None, audio_dl, bitflow_info)
+        return downloaded_file, direct