Downloaded file will be saved under ` /textattack/ `. If it doesn't exist on disk, the zip file will be downloaded and extracted. Args: url (str): URL path from which to download. save_path (str): path to which to save the downloaded content.
(url, save_path, skip_if_cached=True)
| 68 | |
| 69 | |
| 70 | def download_from_url(url, save_path, skip_if_cached=True): |
| 71 | """Downloaded file will be saved under |
| 72 | `<cache_dir>/textattack/<save_path>`. If it doesn't exist on disk, the zip |
| 73 | file will be downloaded and extracted. |
| 74 | |
| 75 | Args: |
| 76 | url (str): URL path from which to download. |
| 77 | save_path (str): path to which to save the downloaded content. |
| 78 | skip_if_cached (bool): If `True`, skip downloading if content is already cached. |
| 79 | |
| 80 | Returns: |
| 81 | str: path to the downloaded folder or file on disk |
| 82 | """ |
| 83 | cache_dest_path = path_in_cache(save_path) |
| 84 | os.makedirs(os.path.dirname(cache_dest_path), exist_ok=True) |
| 85 | # Use a lock to prevent concurrent downloads. |
| 86 | cache_dest_lock_path = cache_dest_path + ".lock" |
| 87 | cache_file_lock = filelock.FileLock(cache_dest_lock_path) |
| 88 | cache_file_lock.acquire() |
| 89 | # Check if already downloaded. |
| 90 | if skip_if_cached and os.path.exists(cache_dest_path): |
| 91 | cache_file_lock.release() |
| 92 | return cache_dest_path |
| 93 | # If the file isn't found yet, download the zip file to the cache. |
| 94 | downloaded_file = tempfile.NamedTemporaryFile( |
| 95 | dir=TEXTATTACK_CACHE_DIR, suffix=".zip", delete=False |
| 96 | ) |
| 97 | http_get(url, downloaded_file) |
| 98 | # Move or unzip the file. |
| 99 | downloaded_file.close() |
| 100 | if zipfile.is_zipfile(downloaded_file.name): |
| 101 | unzip_file(downloaded_file.name, cache_dest_path) |
| 102 | else: |
| 103 | logger.info(f"Copying {downloaded_file.name} to {cache_dest_path}.") |
| 104 | shutil.copyfile(downloaded_file.name, cache_dest_path) |
| 105 | cache_file_lock.release() |
| 106 | # Remove the temporary file. |
| 107 | os.remove(downloaded_file.name) |
| 108 | logger.info(f"Successfully saved {url} to cache.") |
| 109 | return cache_dest_path |
| 110 | |
| 111 | |
| 112 | def unzip_file(path_to_zip_file, unzipped_folder_path): |
nothing calls this directly
no test coverage detected