Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import os | |
| import time | |
| import urllib.request | |
| from pathlib import Path | |
| from urllib.parse import urlparse, parse_qs, unquote | |
| CHUNK_SIZE = 1638400 | |
| TOKEN_FILE = Path.home() / '.civitai' / 'config' | |
| USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3' | |
| def get_token(): | |
| try: | |
| with open(TOKEN_FILE, 'r') as file: | |
| token = file.read() | |
| return token | |
| except Exception: | |
| return None | |
| def store_token(token: str): | |
| TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True) | |
| with open(TOKEN_FILE, 'w') as file: | |
| file.write(token) | |
| def download_file(url: str, output_path: str, token: str): | |
| headers = { | |
| 'Authorization': f'Bearer {token}', | |
| 'User-Agent': USER_AGENT, | |
| } | |
| class NoRedirection(urllib.request.HTTPErrorProcessor): | |
| def http_response(self, request, response): | |
| return response | |
| https_response = http_response | |
| request = urllib.request.Request(url, headers=headers) | |
| opener = urllib.request.build_opener(NoRedirection) | |
| response = opener.open(request) | |
| if response.status in [301, 302, 303, 307, 308]: | |
| redirect_url = response.getheader('Location') | |
| parsed_url = urlparse(redirect_url) | |
| query_params = parse_qs(parsed_url.query) | |
| content_disposition = query_params.get('response-content-disposition', [None])[0] | |
| if content_disposition: | |
| filename = unquote(content_disposition.split('filename=')[1].strip('"')) | |
| else: | |
| return None, f"ERROR: Unable to determine filename" | |
| response = urllib.request.urlopen(redirect_url) | |
| elif response.status == 404: | |
| return None, f"ERROR: File not found" | |
| else: | |
| return None, f"ERROR: No redirect found, something went wrong" | |
| total_size = response.getheader('Content-Length') | |
| if total_size is not None: | |
| total_size = int(total_size) | |
| output_file = os.path.join(output_path, filename) | |
| with open(output_file, 'wb') as f: | |
| downloaded = 0 | |
| start_time = time.time() | |
| while True: | |
| chunk_start_time = time.time() | |
| buffer = response.read(CHUNK_SIZE) | |
| chunk_end_time = time.time() | |
| if not buffer: | |
| break | |
| downloaded += len(buffer) | |
| f.write(buffer) | |
| chunk_time = chunk_end_time - chunk_start_time | |
| if chunk_time > 0: | |
| speed = len(buffer) / chunk_time / (1024 ** 2) # Speed in MB/s | |
| if total_size is not None: | |
| progress = downloaded / total_size | |
| gr.Progress.update(f'Downloading: {filename} [{progress*100:.2f}%] - {speed:.2f} MB/s') | |
| end_time = time.time() | |
| time_taken = end_time - start_time | |
| hours, remainder = divmod(time_taken, 3600) | |
| minutes, seconds = divmod(remainder, 60) | |
| if hours > 0: | |
| time_str = f'{int(hours)}h {int(minutes)}m {int(seconds)}s' | |
| elif minutes > 0: | |
| time_str = f'{int(minutes)}m {int(seconds)}s' | |
| else: | |
| time_str = f'{int(seconds)}s' | |
| return output_file, f"Download completed. File saved as: {filename}. Downloaded in {time_str}" | |
| def run_downloader(url, output_path, token_input): | |
| token = get_token() or token_input | |
| if not token: | |
| return None, "ERROR: Token not provided." | |
| store_token(token) | |
| file_path, message = download_file(url, output_path, token) | |
| return file_path, message | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# CivitAI Downloader") | |
| with gr.Row(): | |
| url_input = gr.Textbox(label="CivitAI Download URL") | |
| output_path_input = gr.Textbox(label="Output Path", value="./") # Default value set to current directory | |
| token_input = gr.Textbox(label="CivitAI API Token (Optional)") | |
| download_button = gr.Button("Download") | |
| output_file = gr.File(label="Downloaded Model") | |
| message_output = gr.Textbox(label="Status Message", interactive=False) | |
| download_button.click( | |
| run_downloader, | |
| inputs=[url_input, output_path_input, token_input], | |
| outputs=[output_file, message_output] | |
| ) | |
| demo.launch() | |