| import os |
| import sys |
| import wget |
| import zipfile |
| from bs4 import BeautifulSoup |
| import requests |
| from urllib.parse import unquote, urlencode, parse_qs, urlparse |
| import re |
| import shutil |
| import six |
|
|
|
|
| def find_folder_parent(search_dir, folder_name): |
| for dirpath, dirnames, _ in os.walk(search_dir): |
| if folder_name in dirnames: |
| return os.path.abspath(dirpath) |
| return None |
|
|
|
|
| now_dir = os.getcwd() |
| sys.path.append(now_dir) |
|
|
| from rvc.lib.utils import format_title |
|
|
| from rvc.lib.tools import gdown |
|
|
| file_path = find_folder_parent(now_dir, "logs") |
|
|
| zips_path = os.getcwd() + "/logs/zips" |
|
|
|
|
| def search_pth_index(folder): |
| pth_paths = [ |
| os.path.join(folder, file) |
| for file in os.listdir(folder) |
| if os.path.isfile(os.path.join(folder, file)) and file.endswith(".pth") |
| ] |
| index_paths = [ |
| os.path.join(folder, file) |
| for file in os.listdir(folder) |
| if os.path.isfile(os.path.join(folder, file)) and file.endswith(".index") |
| ] |
|
|
| return pth_paths, index_paths |
|
|
|
|
| def get_mediafire_download_link(url): |
| response = requests.get(url) |
| response.raise_for_status() |
| soup = BeautifulSoup(response.text, "html.parser") |
| download_button = soup.find( |
| "a", {"class": "input popsok", "aria-label": "Download file"} |
| ) |
| if download_button: |
| download_link = download_button.get("href") |
| return download_link |
| else: |
| return None |
|
|
|
|
| def download_from_url(url): |
| os.makedirs(zips_path, exist_ok=True) |
| if url != "": |
| if "drive.google.com" in url: |
| if "file/d/" in url: |
| file_id = url.split("file/d/")[1].split("/")[0] |
| elif "id=" in url: |
| file_id = url.split("id=")[1].split("&")[0] |
| else: |
| return None |
|
|
| if file_id: |
| os.chdir(zips_path) |
| try: |
| gdown.download( |
| f"https://drive.google.com/uc?id={file_id}", |
| quiet=True, |
| fuzzy=True, |
| ) |
| except Exception as error: |
| error_message = str(error) |
| if ( |
| "Too many users have viewed or downloaded this file recently" |
| in error_message |
| ): |
| os.chdir(now_dir) |
| return "too much use" |
| elif ( |
| "Cannot retrieve the public link of the file." in error_message |
| ): |
| os.chdir(now_dir) |
| return "private link" |
| else: |
| print(error_message) |
| os.chdir(now_dir) |
| return None |
| elif "disk.yandex.ru" in url: |
| base_url = "https://cloud-api.yandex.net/v1/disk/public/resources/download?" |
| public_key = url |
| final_url = base_url + urlencode(dict(public_key=public_key)) |
| response = requests.get(final_url) |
| download_url = response.json()["href"] |
| download_response = requests.get(download_url) |
|
|
| if download_response.status_code == 200: |
| filename = parse_qs(urlparse(unquote(download_url)).query).get( |
| "filename", [""] |
| )[0] |
| if filename: |
| os.chdir(zips_path) |
| with open(filename, "wb") as f: |
| f.write(download_response.content) |
| else: |
| print("Failed to get filename from URL.") |
| return None |
|
|
| elif "pixeldrain.com" in url: |
| try: |
| file_id = url.split("pixeldrain.com/u/")[1] |
| os.chdir(zips_path) |
| print(file_id) |
| response = requests.get(f"https://pixeldrain.com/api/file/{file_id}") |
| if response.status_code == 200: |
| file_name = ( |
| response.headers.get("Content-Disposition") |
| .split("filename=")[-1] |
| .strip('";') |
| ) |
| os.makedirs(zips_path, exist_ok=True) |
| with open(os.path.join(zips_path, file_name), "wb") as newfile: |
| newfile.write(response.content) |
| os.chdir(file_path) |
| return "downloaded" |
| else: |
| os.chdir(file_path) |
| return None |
| except Exception as e: |
| print(e) |
| os.chdir(file_path) |
| return None |
|
|
| elif "cdn.discordapp.com" in url: |
| file = requests.get(url) |
| os.chdir(zips_path) |
| if file.status_code == 200: |
| name = url.split("/") |
| with open(os.path.join(name[-1]), "wb") as newfile: |
| newfile.write(file.content) |
| else: |
| return None |
| elif "/blob/" in url or "/resolve/" in url: |
| os.chdir(zips_path) |
| if "/blob/" in url: |
| url = url.replace("/blob/", "/resolve/") |
|
|
| response = requests.get(url, stream=True) |
| if response.status_code == 200: |
| content_disposition = six.moves.urllib_parse.unquote( |
| response.headers["Content-Disposition"] |
| ) |
| m = re.search(r'filename="([^"]+)"', content_disposition) |
| file_name = m.groups()[0] |
| file_name = file_name.replace(os.path.sep, "_") |
| total_size_in_bytes = int(response.headers.get("content-length", 0)) |
| block_size = 1024 |
| progress_bar_length = 50 |
| progress = 0 |
|
|
| with open(os.path.join(zips_path, file_name), "wb") as file: |
| for data in response.iter_content(block_size): |
| file.write(data) |
| progress += len(data) |
| progress_percent = int((progress / total_size_in_bytes) * 100) |
| num_dots = int( |
| (progress / total_size_in_bytes) * progress_bar_length |
| ) |
| progress_bar = ( |
| "[" |
| + "." * num_dots |
| + " " * (progress_bar_length - num_dots) |
| + "]" |
| ) |
| print( |
| f"{progress_percent}% {progress_bar} {progress}/{total_size_in_bytes} ", |
| end="\r", |
| ) |
| if progress_percent == 100: |
| print("\n") |
|
|
| else: |
| os.chdir(now_dir) |
| return None |
| elif "/tree/main" in url: |
| os.chdir(zips_path) |
| response = requests.get(url) |
| soup = BeautifulSoup(response.content, "html.parser") |
| temp_url = "" |
| for link in soup.find_all("a", href=True): |
| if link["href"].endswith(".zip"): |
| temp_url = link["href"] |
| break |
| if temp_url: |
| url = temp_url |
| url = url.replace("blob", "resolve") |
| if "huggingface.co" not in url: |
| url = "https://huggingface.co" + url |
|
|
| wget.download(url) |
| else: |
| os.chdir(now_dir) |
| return None |
| elif "applio.org" in url: |
| parts = url.split("/") |
| id_with_query = parts[-1] |
| id_parts = id_with_query.split("?") |
| id_number = id_parts[0] |
|
|
| url = "https://cjtfqzjfdimgpvpwhzlv.supabase.co/rest/v1/models" |
| headers = { |
| "apikey": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImNqdGZxempmZGltZ3B2cHdoemx2Iiwicm9sZSI6ImFub24iLCJpYXQiOjE2OTUxNjczODgsImV4cCI6MjAxMDc0MzM4OH0.7z5WMIbjR99c2Ooc0ma7B_FyGq10G8X-alkCYTkKR10" |
| } |
|
|
| params = {"id": f"eq.{id_number}"} |
| response = requests.get(url, headers=headers, params=params) |
| if response.status_code == 200: |
| json_response = response.json() |
| print(json_response) |
| if json_response: |
| link = json_response[0]["link"] |
| verify = download_from_url(link) |
| if verify == "downloaded": |
| return "downloaded" |
| else: |
| return None |
| else: |
| return None |
| else: |
| try: |
| os.chdir(zips_path) |
| wget.download(url) |
| except Exception as error: |
| os.chdir(now_dir) |
| print(error) |
| return None |
|
|
| for currentPath, _, zipFiles in os.walk(zips_path): |
| for Files in zipFiles: |
| filePart = Files.split(".") |
| extensionFile = filePart[len(filePart) - 1] |
| filePart.pop() |
| nameFile = "_".join(filePart) |
| realPath = os.path.join(currentPath, Files) |
| os.rename(realPath, nameFile + "." + extensionFile) |
|
|
| os.chdir(now_dir) |
| return "downloaded" |
|
|
| os.chdir(now_dir) |
| return None |
|
|
|
|
| def extract_and_show_progress(zipfile_path, unzips_path): |
| try: |
| with zipfile.ZipFile(zipfile_path, "r") as zip_ref: |
| for file_info in zip_ref.infolist(): |
| zip_ref.extract(file_info, unzips_path) |
| os.remove(zipfile_path) |
| return True |
| except Exception as error: |
| print(error) |
| return False |
|
|
|
|
| def unzip_file(zip_path, zip_file_name): |
| zip_file_path = os.path.join(zip_path, zip_file_name + ".zip") |
| extract_path = os.path.join(file_path, zip_file_name) |
| with zipfile.ZipFile(zip_file_path, "r") as zip_ref: |
| zip_ref.extractall(extract_path) |
| os.remove(zip_file_path) |
|
|
|
|
| def model_download_pipeline(url): |
| try: |
| verify = download_from_url(url) |
| if verify == "downloaded": |
| extract_folder_path = "" |
| for filename in os.listdir(zips_path): |
| if filename.endswith(".zip"): |
| zipfile_path = os.path.join(zips_path, filename) |
| print("Proceeding with the extraction...") |
| |
| model_zip = os.path.basename(zipfile_path) |
| model_name = format_title(model_zip.split(".zip")[0]) |
| extract_folder_path = os.path.join( |
| "logs", |
| os.path.normpath(model_name), |
| ) |
| |
| success = extract_and_show_progress(zipfile_path, extract_folder_path) |
| |
| subfolders = [ |
| f |
| for f in os.listdir(extract_folder_path) |
| if os.path.isdir(os.path.join(extract_folder_path, f)) |
| ] |
| if len(subfolders) == 1: |
| subfolder_path = os.path.join(extract_folder_path, subfolders[0]) |
| for item in os.listdir(subfolder_path): |
| s = os.path.join(subfolder_path, item) |
| d = os.path.join(extract_folder_path, item) |
| shutil.move(s, d) |
| os.rmdir(subfolder_path) |
| |
| for item in os.listdir(extract_folder_path): |
| if ".pth" in item: |
| file_name = item.split(".pth")[0] |
| if file_name != model_name: |
| os.rename( |
| os.path.join(extract_folder_path, item), |
| os.path.join(extract_folder_path, model_name + ".pth"), |
| ) |
| else: |
| if "v2" not in item: |
| file_name = item.split("_nprobe_1_")[1].split("_v1")[0] |
| if file_name != model_name: |
| new_file_name = ( |
| item.split("_nprobe_1_")[0] |
| + "_nprobe_1_" |
| + model_name |
| + "_v1" |
| ) |
| os.rename( |
| os.path.join(extract_folder_path, item), |
| os.path.join( |
| extract_folder_path, new_file_name + ".index" |
| ), |
| ) |
| else: |
| file_name = item.split("_nprobe_1_")[1].split("_v2")[0] |
| if file_name != model_name: |
| new_file_name = ( |
| item.split("_nprobe_1_")[0] |
| + "_nprobe_1_" |
| + model_name |
| + "_v2" |
| ) |
| os.rename( |
| os.path.join(extract_folder_path, item), |
| os.path.join( |
| extract_folder_path, new_file_name + ".index" |
| ), |
| ) |
| |
| if success: |
| print(f"Model {model_name} downloaded!") |
| else: |
| print(f"Error downloading {model_name}") |
| sys.exit() |
| if extract_folder_path == "": |
| print("Zip file was not found.") |
| result = search_pth_index(extract_folder_path) |
| else: |
| message = "Error" |
| |
| except Exception as error: |
| print(f"Error: {error}") |
| |
|
|
