Update DragMusic/platforms/Youtube.py

DragMusic/platforms/Youtube.py

@@ -1,16 +1,76 @@
 import asyncio
+import glob
+import json
 import os
+import random
 import re
-import json
-import httpx  # Make sure to import httpx
+from concurrent.futures import ThreadPoolExecutor
 from typing import Union
+import string
+import requests
 import yt_dlp
 from pyrogram.enums import MessageEntityType
 from pyrogram.types import Message
-from youtubesearchpython.__future__ import VideosSearch
-from DragMusic.utils.database import is_on_off
-from DragMusic.utils.formatters import time_to_seconds
-import tempfile
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+from youtubesearchpython.__future__ import VideosSearch, CustomSearch
+import base64
+from AnonXMusic import LOGGER
+from AnonXMusic.utils.database import is_on_off
+from AnonXMusic.utils.formatters import time_to_seconds
+from config import YT_API_KEY, YTPROXY_URL as YTPROXY
+
+logger = LOGGER(__name__)
+
+def cookie_txt_file():
+    try:
+        folder_path = f"{os.getcwd()}/cookies"
+        filename = f"{os.getcwd()}/cookies/logs.csv"
+        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)
+        with open(filename, 'a') as file:
+            file.write(f'Choosen File : {cookie_txt_file}\n')
+        return f"""cookies/{str(cookie_txt_file).split("/")[-1]}"""
+    except:
+        return None
+
+
+async def check_file_size(link):
+    async def get_format_info(link):
+        proc = await asyncio.create_subprocess_exec(
+            "yt-dlp",
+            "--cookies", cookie_txt_file(),
+            "-J",
+            link,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            print(f'Error:\n{stderr.decode()}')
+            return None
+        return json.loads(stdout.decode())
+
+    def parse_size(formats):
+        total_size = 0
+        for format in formats:
+            if 'filesize' in format:
+                total_size += format['filesize']
+        return total_size
+
+    info = await get_format_info(link)
+    if info is None:
+        return None
+    
+    formats = info.get('formats', [])
+    if not formats:
+        print("No formats found.")
+        return None
+    
+    total_size = parse_size(formats)
+    return total_size
 
 async def shell_cmd(cmd):
     proc = await asyncio.create_subprocess_shell(
@@ -26,6 +86,7 @@ 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="
@@ -33,6 +94,49 @@ class YouTubeAPI:
         self.status = "https://www.youtube.com/oembed?url="
         self.listbase = "https://youtube.com/playlist?list="
         self.reg = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
+        self.dl_stats = {
+            "total_requests": 0,
+            "okflix_downloads": 0,
+            "cookie_downloads": 0,
+            "existing_files": 0
+        }
+
+    async def _get_video_details(self, link: str, limit: int = 20) -> Union[dict, None]:
+        """Helper function to get video details with duration limit and error handling"""
+        try:
+            results = VideosSearch(link, limit=limit)
+            search_results = (await results.next()).get("result", [])
+
+            for result in search_results:
+                duration_str = result.get("duration", "0:00")
+
+                # Convert duration to seconds
+                try:
+                    parts = duration_str.split(":")
+                    duration_secs = 0
+                    if len(parts) == 3:  # HH:MM:SS
+                        duration_secs = int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2])
+                    elif len(parts) == 2:  # MM:SS
+                        duration_secs = int(parts[0]) * 60 + int(parts[1])
+
+                    # Skip videos longer than 1 hour
+                    if duration_secs > 3600:
+                        continue
+
+                    return result
+
+                except (ValueError, IndexError):
+                    continue
+            
+            search = CustomSearch(query=link, searchPreferences="EgIYAw==" ,limit=1)
+            for res in (await search.next()).get("result", []):
+                return res
+
+            return None
+
+        except Exception as e:
+            LOGGER(__name__).error(f"Error in _get_video_details: {str(e)}")
+            return None
 
     async def exists(self, link: str, videoid: Union[bool, str] = None):
         if videoid:
@@ -71,16 +175,26 @@ class YouTubeAPI:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        results = VideosSearch(link, limit=1)
-        for result in (await results.next())["result"]:
-            title = result["title"]
-            duration_min = result["duration"]
-            thumbnail = result["thumbnails"][0]["url"].split("?")[0]
-            vidid = result["id"]
-            if str(duration_min) == "None":
-                duration_sec = 0
-            else:
-                duration_sec = int(time_to_seconds(duration_min))
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+
+
+        result = await self._get_video_details(link)
+        if not result:
+            raise ValueError("No suitable video found (duration > 1 hour or video unavailable)")
+
+        title = result["title"]
+        duration_min = result["duration"]
+        thumbnail = result["thumbnails"][0]["url"].split("?")[0]
+        vidid = result["id"]
+
+        if str(duration_min) == "None":
+            duration_sec = 0
+        else:
+            duration_sec = int(time_to_seconds(duration_min))
+
         return title, duration_min, duration_sec, thumbnail, vidid
 
     async def title(self, link: str, videoid: Union[bool, str] = None):
@@ -88,38 +202,59 @@ class YouTubeAPI:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        results = VideosSearch(link, limit=1)
-        for result in (await results.next())["result"]:
-            title = result["title"]
-        return title
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+            
+        result = await self._get_video_details(link)
+        if not result:
+            raise ValueError("No suitable video found (duration > 1 hour or video unavailable)")
+        return result["title"]
 
     async def duration(self, link: str, videoid: Union[bool, str] = None):
         if videoid:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        results = VideosSearch(link, limit=1)
-        for result in (await results.next())["result"]:
-            duration = result["duration"]
-        return duration
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+
+        result = await self._get_video_details(link)
+        if not result:
+            raise ValueError("No suitable video found (duration > 1 hour or video unavailable)")
+        return result["duration"]
 
     async def thumbnail(self, link: str, videoid: Union[bool, str] = None):
         if videoid:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        results = VideosSearch(link, limit=1)
-        for result in (await results.next())["result"]:
-            thumbnail = result["thumbnails"][0]["url"].split("?")[0]
-        return thumbnail
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+
+        result = await self._get_video_details(link)
+        if not result:
+            raise ValueError("No suitable video found (duration > 1 hour or video unavailable)")
+        return result["thumbnails"][0]["url"].split("?")[0]
 
     async def video(self, link: str, videoid: Union[bool, str] = None):
         if videoid:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+
         proc = await asyncio.create_subprocess_exec(
             "yt-dlp",
+            "--cookies",cookie_txt_file(),
             "-g",
             "-f",
             "best[height<=?720][width<=?1280]",
@@ -138,8 +273,12 @@ class YouTubeAPI:
             link = self.listbase + link
         if "&" in link:
             link = link.split("&")[0]
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
         playlist = await shell_cmd(
-            f"yt-dlp -i --get-id --flat-playlist --playlist-end {limit} --skip-download {link}"
+            f"yt-dlp -i --get-id --flat-playlist --cookies {cookie_txt_file()} --playlist-end {limit} --skip-download {link}"
         )
         try:
             result = playlist.split("\n")
@@ -155,28 +294,34 @@ class YouTubeAPI:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        results = VideosSearch(link, limit=1)
-        for result in (await results.next())["result"]:
-            title = result["title"]
-            duration_min = result["duration"]
-            vidid = result["id"]
-            yturl = result["link"]
-            thumbnail = result["thumbnails"][0]["url"].split("?")[0]
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+
+        result = await self._get_video_details(link)
+        if not result:
+            raise ValueError("No suitable video found (duration > 1 hour or video unavailable)")
+
         track_details = {
-            "title": title,
-            "link": yturl,
-            "vidid": vidid,
-            "duration_min": duration_min,
-            "thumb": thumbnail,
+            "title": result["title"],
+            "link": result["link"],
+            "vidid": result["id"],
+            "duration_min": result["duration"],
+            "thumb": result["thumbnails"][0]["url"].split("?")[0],
         }
-        return track_details, vidid
+        return track_details, result["id"]
 
     async def formats(self, link: str, videoid: Union[bool, str] = None):
         if videoid:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        ytdl_opts = {"quiet": True}
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+        ytdl_opts = {"quiet": True, "cookiefile" : cookie_txt_file()}
         ydl = yt_dlp.YoutubeDL(ytdl_opts)
         with ydl:
             formats_available = []
@@ -207,39 +352,51 @@ class YouTubeAPI:
                     )
         return formats_available, link
 
-    async def slider(
-        self,
-        link: str,
-        query_type: int,
-        videoid: Union[bool, str] = None,
-    ):
+    async def slider(self, link: str, query_type: int, videoid: Union[bool, str] = None):
         if videoid:
             link = self.base + link
         if "&" in link:
             link = link.split("&")[0]
-        a = VideosSearch(link, limit=10)
-        result = (await a.next()).get("result")
-        title = result[query_type]["title"]
-        duration_min = result[query_type]["duration"]
-        vidid = result[query_type]["id"]
-        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."}
+        if "?si=" in link:
+            link = link.split("?si=")[0]
+        elif "&si=" in link:
+            link = link.split("&si=")[0]
+
+        try:
+            results = []
+            search = VideosSearch(link, limit=10)
+            search_results = (await search.next()).get("result", [])
+
+            # Filter videos longer than 1 hour
+            for result in search_results:
+                duration_str = result.get("duration", "0:00")
+                try:
+                    parts = duration_str.split(":")
+                    duration_secs = 0
+                    if len(parts) == 3:
+                        duration_secs = int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2])
+                    elif len(parts) == 2:
+                        duration_secs = int(parts[0]) * 60 + int(parts[1])
+
+                    if duration_secs <= 3600:
+                        results.append(result)
+                except (ValueError, IndexError):
+                    continue
+
+            if not results or query_type >= len(results):
+                raise ValueError("No suitable videos found within duration limit")
+
+            selected = results[query_type]
+            return (
+                selected["title"],
+                selected["duration"],
+                selected["thumbnails"][0]["url"].split("?")[0],
+                selected["id"]
+            )
+
+        except Exception as e:
+            LOGGER(__name__).error(f"Error in slider: {str(e)}")
+            raise ValueError("Failed to fetch video details")
 
     async def download(
         self,
@@ -253,65 +410,164 @@ class YouTubeAPI:
         title: Union[bool, str] = None,
     ) -> str:
         if videoid:
+            vid_id = link
             link = self.base + link
-        if "&" in link:
-            link = link.split("&")[0]
         loop = asyncio.get_running_loop()
-        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": xyz,
-                "geo_bypass": True,
-                "nocheckcertificate": True,
+
+        def create_session():
+            session = requests.Session()
+            retries = Retry(total=3, backoff_factor=0.1)
+            session.mount('http://', HTTPAdapter(max_retries=retries))
+            session.mount('https://', HTTPAdapter(max_retries=retries))
+            return session
+
+        def get_ydl_opts(output_path):
+            return {
+                "outtmpl": output_path,
                 "quiet": True,
-                "no_warnings": True,
-            }
-            x = yt_dlp.YoutubeDL(ydl_optssx)
-            if os.path.exists(xyz):
-                return xyz
-            x.download([url])
-            return xyz
-        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": xyz,
-                "geo_bypass": True,
+                "xff": "IN",
                 "nocheckcertificate": True,
-                "quiet": True,
-                "no_warnings": True,
+                "compat-options": "allow-unsafe-ext",
+                "concurrent-fragments": 99,  # Increased from 10
+                "retries": 3,
+                # "http-chunk-size": 10485760,  # 10MB chunks
+                # "buffersize": 32768,
             }
-            x = yt_dlp.YoutubeDL(ydl_optssx)
-            if os.path.exists(xyz):
-                return xyz
-            x.download([url])
-            return xyz
+        ## Method added by sparrow...
+        # def audio_dl(vid_id):
+        #     try:
+        #         session = create_session()
+        #         random_prefix = ''.join(random.choices(string.ascii_letters + string.digits, k=12))
+        #         ruseragent = f"{random_prefix} Mozilla/5.9 ({random.randint(1000, 9999)})"
+        #         bsid = base64.b64encode(vid_id.encode()).decode() ## For next update 
+        #         headers = {
+        #             "x-api-key": f"{YT_API_KEY}",
+        #             "User-Agent": ruseragent
+        #         }
+        #         res = session.get(f"{YTPROXY}/api/info?video_id={vid_id}", headers=headers, timeout=300)
+        #         response = res.json()
+
+        #         if res.status_code == 200 and response['status'] == 'success':
+        #             xyz = os.path.join("downloads", f"{vid_id}.m4a")
+        #             if os.path.exists(xyz):
+        #                 return xyz
+
+        #             with session.get(response['download_url'], headers=headers, stream=True, timeout=300) as r:
+        #                 r.raise_for_status()
+        #                 total_size = int(r.headers.get('content-length', 0))
+        #                 block_size = 1024 * 1024  # 1MB chunks
+                        
+        #                 with open(xyz, 'wb') as f:
+        #                     for chunk in r.iter_content(chunk_size=block_size):
+        #                         if chunk:  # filter out keep-alive chunks
+        #                             f.write(chunk)
+                    
+        #             return xyz
+        #         else:
+        #             print(f"Proxy returned error status: {response}")
+
+        #     except requests.exceptions.RequestException as e:
+        #         print(f"Network error while downloading: {str(e)}")
+        #     except json.JSONDecodeError as e:
+        #         print(f"Invalid response from proxy: {str(e)}")
+        #     except Exception as e:
+        #         print(f"Error in downloading song: {str(e)}")
+        #     return None
+
+
+        #  New updated Method...
+        def audio_dl(vid_id):
+            try:
+                if not YT_API_KEY:
+                    print("API KEY not set in config, Set API Key you got from @tgmusic_apibot")
+                    return None
+                if not YTPROXY:
+                    print("API Endpoint not set in config\nPlease set a valid endpoint for YTPROXY_URL in config.")
+                    return None
+                headers = {
+                    "x-api-key": f"{YT_API_KEY}",
+                    "User-Agent": "Mozilla/5"
+                }
+                xyz = os.path.join("downloads", f"{vid_id}.mp3")
+                if os.path.exists(xyz):
+                    return xyz
+                getAudio = requests.get(f"{YTPROXY}/audio/{vid_id}", headers=headers, timeout=60)
+                try:
+                    songData = getAudio.json()
+                except Exception as e:
+                    print(f"Invalid response from API: {str(e)}")
+                    return None
+                status = songData.get('status')
+                if status == 'success':
+                    songlink = songData['audio_url']
+                    audio_url = base64.b64decode(songlink).decode()
+                    ydl_opts = get_ydl_opts(xyz)
+                    with ThreadPoolExecutor(max_workers=4) as executor:
+                        future = executor.submit(lambda: yt_dlp.YoutubeDL(ydl_opts).download(audio_url))
+                        future.result()
+                    return xyz
+                elif status == 'error':
+                    print(f"Error: {songData.get('message', 'Unknown error from API.')}")
+                    return None
+                else:
+                    print("Could not fetch Backend \nPlease contact API provider.")
+                    return None
+            except requests.exceptions.RequestException as e:
+                print(f"Network error while downloading: {str(e)}")
+            except json.JSONDecodeError as e:
+                print(f"Invalid response from proxy: {str(e)}")
+            except Exception as e:
+                print(f"Error in downloading song: {str(e)}")
+            return None
+        
+        def video_dl(vid_id):
+            try:
+                if not YT_API_KEY:
+                    print("API KEY not set in config, Set API Key you got from @tgmusic_apibot")
+                    return None
+                if not YTPROXY:
+                    print("API Endpoint not set in config\nPlease set a valid endpoint for YTPROXY_URL in config.")
+                    return None
+                headers = {
+                    "x-api-key": f"{YT_API_KEY}",
+                    "User-Agent": "Mozilla/5"
+                }
+                xyz = os.path.join("downloads", f"{vid_id}.mp4")
+                if os.path.exists(xyz):
+                    return xyz
+                getVideo = requests.get(f"{YTPROXY}/beta/{vid_id}", headers=headers, timeout=60)
+                try:
+                    videoData = getVideo.json()
+                except Exception as e:
+                    print(f"Invalid response from API: {str(e)}")
+                    return None
+                status = videoData.get('status')
+                if status == 'success':
+                    videolink = videoData['video_sd']
+                    video_url = base64.b64decode(videolink).decode()
+                    logger.debug(f"Got video url {video_url}")
+                    ydl_opts = get_ydl_opts(f"downloads/{vid_id}.mp4")
+                    with ThreadPoolExecutor(max_workers=4) as executor:
+                        future = executor.submit(lambda: yt_dlp.YoutubeDL(ydl_opts).download(video_url))
+                        future.result()  
+                    return xyz
+                elif status == 'error':
+                    print(f"Error: {videoData.get('message', 'Unknown error from API.')}")
+                    return None
+                else:
+                    print("Could not fetch Backend \nPlease contact API provider.")
+                    return None
+            except requests.exceptions.RequestException as e:
+                print(f"Network error while downloading: {str(e)}")
+            except json.JSONDecodeError as e:
+                print(f"Invalid response from proxy: {str(e)}")
+            except Exception as e:
+                print(f"Error in downloading song: {str(e)}")
+            return None
+        
         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,
@@ -319,14 +575,15 @@ class YouTubeAPI:
                 "nocheckcertificate": True,
                 "quiet": True,
                 "no_warnings": True,
+                "cookiefile" : cookie_txt_file(),
                 "prefer_ffmpeg": True,
                 "merge_output_format": "mp4",
             }
             x = yt_dlp.YoutubeDL(ydl_optssx)
             x.download([link])
+
         def song_audio_dl():
-            temp_dir = tempfile.gettempdir()
-            fpath = os.path.join(temp_dir, f"{title}.%(ext)s")
+            fpath = f"downloads/{title}.%(ext)s"
             ydl_optssx = {
                 "format": format_id,
                 "outtmpl": fpath,
@@ -334,6 +591,7 @@ class YouTubeAPI:
                 "nocheckcertificate": True,
                 "quiet": True,
                 "no_warnings": True,
+                "cookiefile" : cookie_txt_file(),
                 "prefer_ffmpeg": True,
                 "postprocessors": [
                     {
@@ -345,20 +603,20 @@ class YouTubeAPI:
             }
             x = yt_dlp.YoutubeDL(ydl_optssx)
             x.download([link])
+
         if songvideo:
             await loop.run_in_executor(None, song_video_dl)
-            temp_dir = tempfile.gettempdir()
-            fpath = os.path.join(temp_dir, f"{title}.mp4")
+            fpath = f"downloads/{title}.mp4"
             return fpath
         elif songaudio:
             await loop.run_in_executor(None, song_audio_dl)
-            temp_dir = tempfile.gettempdir()
-            fpath = os.path.join(temp_dir, f"{title}.mp3")
+            fpath = f"downloads/{title}.mp3"
             return fpath
         elif video:
             direct = True
-            downloaded_file = await loop.run_in_executor(None, video_dl, bitflow_info)
+            downloaded_file = await loop.run_in_executor(None, lambda:video_dl(vid_id))
         else:
             direct = True
-            downloaded_file = await loop.run_in_executor(None, audio_dl, bitflow_info)
+            downloaded_file = await loop.run_in_executor(None, lambda:audio_dl(vid_id))
+        
         return downloaded_file, direct