app.py

import subprocess
from functools import wraps
from bs4 import BeautifulSoup
from tqdm import tqdm
import json
import requests

def start_tor_and_wait(tor_path="tor"):
    process = subprocess.Popen(
        [tor_path],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1
    )
    for line in process.stdout:
        print(line)
        if "Bootstrapped 100% (done)" in line:
            print("✅ Tor fully started")
            return process
    return None

class SocksProxy:
    def __init__(self, proxy_url):
        self.proxies = {"http": proxy_url, "https": proxy_url}
        self._original_request = None

    def __enter__(self):
        self._original_request = requests.sessions.Session.request

        def patched_request(session, method, url, **kwargs):
            kwargs.setdefault("proxies", self.proxies)
            return self._original_request(session, method, url, **kwargs)

        requests.sessions.Session.request = patched_request

    def __exit__(self, exc_type, exc_val, exc_tb):
        requests.sessions.Session.request = self._original_request

def use_proxy(proxy_url="socks5h://127.0.0.1:9050"):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            with SocksProxy(proxy_url):
                return func(*args, **kwargs)
        return wrapper
    return decorator

class YTDownloader:
    @use_proxy()
    def __init__(self):
        self.api = self.get_api()

    def get_between(self,src,left_str,right_str):
        return src.split(left_str)[1].split(right_str)[0]

    @use_proxy()
    def __call__(self, id, download_file="output.opus"):
        data = [v for i,v in self.get_yt_audio_data(id).items() if "OPUS-audio" in i][-1]
        return self.download_file(self.get_yt_link(self.get_task_id(data['resource_content'])),download_file)

    def download_file(self, url, output_file):
        response = requests.get(url, stream=True)
        total_size = int(response.headers.get('content-length', 0))
        chunk_size = 1024  # 1 KB

        with open(output_file, 'wb') as file, tqdm(
            desc=output_file,
            total=total_size,
            unit='B',
            unit_scale=True,
            unit_divisor=1024,
        ) as bar:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if chunk:
                    file.write(chunk)
                    bar.update(len(chunk))
        return output_file

    def get_yt_audio_data(self,id):
        return {"-".join([i['quality'],i['format'],i['type']]):i for i in requests.post('https://api.vidssave.com/api/contentsite_api/media/parse', headers={
            'accept': '*/*',
            'accept-language': 'en-GB,en;q=0.9,gu;q=0.8,en-US;q=0.7,hi;q=0.6',
            'content-type': 'application/x-www-form-urlencoded',
            'dnt': '1',
            'origin': 'https://vidssave.com',
            'priority': 'u=1, i',
            'referer': 'https://vidssave.com/',
            'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
            'sec-ch-ua-mobile': '?0',
            'sec-ch-ua-platform': '"Windows"',
            'sec-fetch-dest': 'empty',
            'sec-fetch-mode': 'cors',
            'sec-fetch-site': 'same-site',
            'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36',
        }, data={
            'auth': self.api,
            'domain': 'api-ak.vidssave.com',
            'origin': 'source',
            'link': f'https://youtu.be/{id}',
        }).json()['data']['resources']}

    def get_api(self):
        for i in BeautifulSoup(requests.get('https://vidssave.com/yt', headers={
        'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
        'accept-language': 'en-GB,en;q=0.9,gu;q=0.8,en-US;q=0.7,hi;q=0.6',
        'cache-control': 'max-age=0',
        'dnt': '1',
        'if-modified-since': 'Thu, 05 Feb 2026 09:20:05 GMT',
        'priority': 'u=0, i',
        'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'document',
        'sec-fetch-mode': 'navigate',
        'sec-fetch-site': 'same-origin',
        'sec-fetch-user': '?1',
        'upgrade-insecure-requests': '1',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36',
    }).text,"html.parser").find_all("script"):
            try:
                if i.has_attr("async"):
                    i = i['src'].split("/")[4]
                    if ".js" in i:
                        return self.get_between(requests.get("https://vidssave.com/_next/static/chunks/"+i, headers={'sec-ch-ua-platform': '"Windows"', 'Referer': 'https://vidssave.com/yt', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36', 'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"', 'DNT': '1', 'sec-ch-ua-mobile': '?0'}).text,'auth:"','",')
            except:
                pass

    def get_task_id(self,resource_id):
        return requests.post('https://api.vidssave.com/api/contentsite_api/media/download', headers={
    'accept': '*/*',
    'accept-language': 'en-GB,en;q=0.9,gu;q=0.8,en-US;q=0.7,hi;q=0.6',
    'content-type': 'application/x-www-form-urlencoded',
    'dnt': '1',
    'origin': 'https://vidssave.com',
    'priority': 'u=1, i',
    'referer': 'https://vidssave.com/',
    'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
    'sec-fetch-dest': 'empty',
    'sec-fetch-mode': 'cors',
    'sec-fetch-site': 'same-site',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36',
}, data={
    'auth': self.api,
    'domain': 'api-ak.vidssave.com',
    'request': resource_id,
    'no_encrypt': '1',
}).json()['data']['task_id']

    def get_yt_link(self,task_id):
        return json.loads(requests.get(
    f'https://api.vidssave.com/sse/contentsite_api/media/download_query?auth={self.api}&domain=api-ak.vidssave.com&task_id={task_id}&download_domain=vidssave.com&origin=content_site',
    headers={
    'accept': 'text/event-stream',
    'accept-language': 'en-GB,en;q=0.9,gu;q=0.8,en-US;q=0.7,hi;q=0.6',
    'cache-control': 'no-cache',
    'dnt': '1',
    'origin': 'https://vidssave.com',
    'priority': 'u=1, i',
    'referer': 'https://vidssave.com/',
    'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
    'sec-fetch-dest': 'empty',
    'sec-fetch-mode': 'cors',
    'sec-fetch-site': 'same-site',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36',
},
).text.strip().split("\n")[-1].lstrip("data: "))['download_link']

start_tor_and_wait()
client = YTDownloader()
import gradio as gr
def f(x,prog=gr.Progress(True)):
    return client(x)
gr.Interface(client,gr.Textbox(),gr.Audio()).launch()