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()