downloads.py
ultralytics\utils\downloads.py
目錄
downloads.py
1.所需的庫和模塊
2.def is_url(url, check=False):?
3.def delete_dsstore(path, files_to_delete=(".DS_Store", "__MACOSX")):?
4.def zip_directory(directory, compress=True, exclude=(".DS_Store", "__MACOSX"), progress=True):?
5.def unzip_file(file, path=None, exclude=(".DS_Store", "__MACOSX"), exist_ok=False, progress=True):?
6.def check_disk_space(url="https://ultralytics.com/assets/coco8.zip", path=Path.cwd(), sf=1.5, hard=True):?
7.def get_google_drive_file_info(link):?
8.def safe_download(url, file=None, dir=None, unzip=True, delete=False, curl=False, retry=3, min_bytes=1e0, exist_ok=False, progress=True,):?
9.def get_github_assets(repo="ultralytics/assets", version="latest", retry=False):?
10.def attempt_download_asset(file, repo="ultralytics/assets", release="v8.3.0", **kwargs):?
11.def download(url, dir=Path.cwd(), unzip=True, delete=False, curl=False, threads=1, retry=3, exist_ok=False):?
1.所需的庫和模塊
# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/licenseimport re
import shutil
import subprocess
from itertools import repeat
from multiprocessing.pool import ThreadPool
from pathlib import Path
from urllib import parse, requestimport requests
import torchfrom ultralytics.utils import LOGGER, TQDM, checks, clean_url, emojis, is_online, url2file# Define Ultralytics GitHub assets maintained at https://github.com/ultralytics/assets
# 這段代碼定義了多個與GitHub倉庫資產相關的變量，用于指定和管理不同版本的YOLO模型文件和其他相關資產。
# 定義GitHub倉庫路徑。
# 定義了一個變量 GITHUB_ASSETS_REPO ，表示GitHub倉庫的路徑。這個倉庫用于存儲YOLO模型文件和其他資產。
GITHUB_ASSETS_REPO = "ultralytics/assets"
# 定義GitHub資產文件名列表。
# 定義了一個變量 GITHUB_ASSETS_NAMES ，它是一個包含所有YOLO模型文件和其他資產文件名的列表。
# 使用列表推導式生成文件名列表，涵蓋了不同版本的YOLO模型文件（如YOLOv3、YOLOv5、YOLOv8等）。
# 文件名包括不同的變體（如 -cls 、 -seg 、 -pose 等）和分辨率（如 6 ）。
# 還包括一些特定的文件名，如 mobile_sam.pt 和 calibration_image_sample_data_20x128x128x3_float32.npy.zip 。
GITHUB_ASSETS_NAMES = ([f"yolov8{k}{suffix}.pt" for k in "nsmlx" for suffix in ("", "-cls", "-seg", "-pose", "-obb", "-oiv7")]+ [f"yolo11{k}{suffix}.pt" for k in "nsmlx" for suffix in ("", "-cls", "-seg", "-pose", "-obb")]+ [f"yolov5{k}{resolution}u.pt" for k in "nsmlx" for resolution in ("", "6")]+ [f"yolov3{k}u.pt" for k in ("", "-spp", "-tiny")]+ [f"yolov8{k}-world.pt" for k in "smlx"]+ [f"yolov8{k}-worldv2.pt" for k in "smlx"]+ [f"yolov9{k}.pt" for k in "tsmce"]+ [f"yolov10{k}.pt" for k in "nsmblx"]+ [f"yolo_nas_{k}.pt" for k in "sml"]+ [f"sam_{k}.pt" for k in "bl"]+ [f"FastSAM-{k}.pt" for k in "sx"]+ [f"rtdetr-{k}.pt" for k in "lx"]+ ["mobile_sam.pt"]+ ["calibration_image_sample_data_20x128x128x3_float32.npy.zip"]
)
# 定義GitHub資產文件名的莖（stem）。
# 定義了一個變量 GITHUB_ASSETS_STEMS ，它是一個包含所有YOLO模型文件和其他資產文件名的莖（stem）的列表。
# 使用 Path(k).stem 提取每個文件名的莖（即去除擴展名的部分）。
GITHUB_ASSETS_STEMS = [Path(k).stem for k in GITHUB_ASSETS_NAMES]
# 這段代碼的主要功能是。定義GitHub倉庫路徑：指定存儲YOLO模型文件和其他資產的GitHub倉庫路徑。定義GitHub資產文件名列表：生成一個包含所有YOLO模型文件和其他資產文件名的列表。定義GitHub資產文件名的莖：提取每個文件名的莖，方便后續處理。通過這些定義，可以方便地管理和引用GitHub倉庫中的YOLO模型文件和其他資產，確保代碼的靈活性和可維護性。2.def is_url(url, check=False):?
# 這段代碼定義了一個函數 is_url ，用于檢查給定的字符串是否是一個有效的 URL，并且可以選擇性地檢查該 URL 是否在線可訪問。
# 定義了一個函數 is_url ，接收兩個參數。
# 1.url ：要檢查的 URL，可以是字符串或任何可以轉換為字符串的對象。
# 2.check ：一個布爾值，默認為 False 。如果設置為 True ，函數會進一步檢查 URL 是否在線可訪問。
def is_url(url, check=False):# 驗證給定的字符串是否為 URL，并可選擇檢查 URL 是否在線存在。"""Validates if the given string is a URL and optionally checks if the URL exists online.Args:url (str): The string to be validated as a URL.check (bool, optional): If True, performs an additional check to see if the URL exists online.Defaults to False.Returns:(bool): Returns True for a valid URL. If 'check' is True, also returns True if the URL exists online.Returns False otherwise.Example:```pythonvalid = is_url("https://www.example.com")```"""# 開始一個 try 塊，用于捕獲可能出現的異常。try:# 將輸入的 url 轉換為字符串。這確保了即使輸入是其他類型（如 Path 對象或字節序列），也能正確處理。url = str(url)# result = urlparse(urlstring, scheme='', allow_fragments=True)# urlparse() 函數是 Python 標準庫 urllib.parse 模塊中的一個函數，用于解析 URL（統一資源定位符）并將其分解為組件。這個函數在處理網絡地址時非常有用，因為它可以將復雜的 URL 分解成易于管理的部分。# 參數 ：# urlstring : 要解析的 URL 字符串。# scheme : （可選）如果提供，將用于覆蓋 URL 中的方案部分。# allow_fragments : （可選）一個布爾值，指示是否允許解析 URL 的片段部分（即 # 后面的部分）。默認為 True 。# 返回值 :# urlparse() 函數返回一個 ParseResult 對象，該對象包含以下屬性 :# scheme : URL 的方案部分（例如 http 、 https ）。# netloc : 網絡位置部分（例如域名和端口）。# path : URL 的路徑部分。# params : URL 的參數部分（ ? 后面的部分）。# query : URL 的查詢部分（ ? 后面的部分，不包括 # ）。# fragment : URL 的片段部分（ # 后面的部分）。# urlparse() 函數是處理 URL 的基礎工具，常用于網絡編程、Web 開發和任何需要解析或構造 URL 的場景。# 使用 urllib.parse.urlparse 函數解析 URL。 urlparse 會將 URL 分解為幾個部分，包括協議（scheme）、網絡位置（netloc）、路徑（path）等。result = parse.urlparse(url)# 使用 assert 語句檢查解析結果中的協議（ scheme ）和網絡位置（ netloc ）是否都存在。 如果 scheme 和 netloc 都存在，說明這是一個有效的 URL。 如果其中一個不存在， assert 會拋出 AssertionError 。assert all([result.scheme, result.netloc]) # check if is url# 如果 check 參數為 True ，則執行以下代碼，進一步檢查 URL 是否在線可訪問。if check:# 使用 urllib.request.urlopen 打開 URL，并獲取響應。 with 語句確保響應對象在使用后正確關閉。with request.urlopen(url) as response:# 檢查響應的狀態碼是否為 200（HTTP 狀態碼 200 表示請求成功）。如果狀態碼為 200，說明 URL 在線可訪問，返回 True ；否則返回 False 。return response.getcode() == 200 # check if exists online# 如果 check 參數為 False ，或者 check 參數為 True 且 URL 在線可訪問，返回 True 。return True# 捕獲 try 塊中可能出現的任何異常。except Exception:# 如果捕獲到異常，說明輸入的 URL 不是有效的 URL，或者在檢查在線狀態時出現問題，返回 False 。return False
# is_url 函數的主要功能是。檢查 URL 格式：通過解析 URL，檢查其是否包含協議（如 http 、 https ）和網絡位置（如域名或 IP 地址）。檢查 URL 是否在線：如果 check 參數為 True ，進一步檢查 URL 是否在線可訪問（即 HTTP 狀態碼是否為 200）。
# 注意事項 ：
# 異常處理 ： except Exception 捕獲所有異常，這可能會隱藏一些具體的錯誤信息。在實際使用中，可以根據需要捕獲更具體的異常類型。
# 網絡請求 ：當 check=True 時，函數會發起網絡請求。這可能會導致性能問題，特別是在頻繁調用時。建議在需要時才啟用 check 參數。
# 安全性 ：在處理用戶輸入的 URL 時，需要注意防范潛在的安全問題，如注入攻擊。3.def delete_dsstore(path, files_to_delete=(".DS_Store", "__MACOSX")):?
# 這段代碼定義了一個名為 delete_dsstore 的函數，用于刪除指定路徑及其子目錄中特定的文件（如 .DS_Store 和 __MACOSX 文件夾）。
# 定義了一個函數 delete_dsstore ，它接受兩個參數。
# 1.path ：表示要進行文件刪除操作的目錄路徑。
# 2.files_to_delete ：一個元組，用于指定需要刪除的文件或文件夾名稱，默認值為 (".DS_Store", "__MACOSX") 。這意味著如果沒有指定其他文件類型，函數會默認刪除 .DS_Store 文件和 __MACOSX 文件夾。
def delete_dsstore(path, files_to_delete=(".DS_Store", "__MACOSX")):# 刪除指定目錄下的所有“.DS_store”文件。# 注意：# “.DS_store”文件由 Apple 操作系統創建，包含有??關文件夾和文件的元數據。它們是隱藏的系統文件，在不同操作系統之間傳輸文件時可能會導致問題。"""Deletes all ".DS_store" files under a specified directory.Args:path (str, optional): The directory path where the ".DS_store" files should be deleted.files_to_delete (tuple): The files to be deleted.Example:```pythonfrom ultralytics.utils.downloads import delete_dsstoredelete_dsstore("path/to/dir")```Note:".DS_store" files are created by the Apple operating system and contain metadata about folders and files. Theyare hidden system files and can cause issues when transferring files between different operating systems."""# 遍歷 files_to_delete 元組中的每個文件或文件夾名稱。例如，如果 files_to_delete 是默認值，那么會依次處理 .DS_Store 和 __MACOSX 。for file in files_to_delete:# 使用 Path 類（來自 pathlib 模塊）來處理路徑操作。# Path(path).rglob(file) ：在指定的 path 路徑及其所有子目錄中遞歸查找與 file 匹配的文件或文件夾路徑。# list() ：將匹配到的路徑對象轉換為一個列表，存儲在變量 matches 中。matches = list(Path(path).rglob(file))# 使用 LOGGER （是一個日志記錄器對象）記錄日志信息。 日志內容顯示正在刪除的文件類型（ file ）以及匹配到的文件路徑列表（ matches ），方便調試和跟蹤。LOGGER.info(f"Deleting {file} files: {matches}") # 刪除 {file} 文件：{matches} 。# 遍歷 matches 列表中的每個路徑對象 f 。for f in matches:# 調用 f.unlink() 方法刪除當前路徑對象 f 所指向的文件或文件夾。 unlink() 是 pathlib 模塊中用于刪除文件或文件夾的方法。f.unlink()
# 這段代碼通過遞歸搜索指定路徑及其子目錄，查找并刪除指定的文件或文件夾（默認為 .DS_Store 和 __MACOSX ）。它利用了 pathlib 模塊的路徑操作功能，并通過日志記錄功能提供了操作的透明性。此函數在處理跨平臺文件系統時非常有用，尤其是清理 macOS 系統在文件傳輸過程中生成的隱藏文件和文件夾。4.def zip_directory(directory, compress=True, exclude=(".DS_Store", "__MACOSX"), progress=True):?
# 這段代碼定義了一個名為 zip_directory 的函數，用于將指定目錄及其子目錄中的文件打包成一個 ZIP 文件，同時支持壓縮、排除特定文件以及顯示進度條等功能。
# 定義了一個函數 zip_directory ，接受以下參數。
# 1.directory ：要打包的目錄路徑。
# 2.compress ：布爾值，表示是否對文件進行壓縮，默認為 True 。
# 3.exclude ：一個元組，用于指定需要排除的文件名，默認為 (".DS_Store", "__MACOSX") 。
# 4.progress ：布爾值，表示是否顯示進度條，默認為 True 。
def zip_directory(directory, compress=True, exclude=(".DS_Store", "__MACOSX"), progress=True):# 壓縮目錄內容，排除包含排除列表中的字符串的文件。生成的 zip 文件以目錄命名并放在旁邊。"""Zips the contents of a directory, excluding files containing strings in the exclude list. The resulting zip file isnamed after the directory and placed alongside it.Args:directory (str | Path): The path to the directory to be zipped.compress (bool): Whether to compress the files while zipping. Default is True.exclude (tuple, optional): A tuple of filename strings to be excluded. Defaults to ('.DS_Store', '__MACOSX').progress (bool, optional): Whether to display a progress bar. Defaults to True.Returns:(Path): The path to the resulting zip file.Example:```pythonfrom ultralytics.utils.downloads import zip_directoryfile = zip_directory("path/to/dir")```"""# 導入了 zipfile 模塊中的常量和類。# ZIP_DEFLATED ：表示使用 DEFLATE 壓縮算法。# ZIP_STORED ：表示不進行壓縮，直接存儲文件。# ZipFile ：用于操作 ZIP 文件的類。from zipfile import ZIP_DEFLATED, ZIP_STORED, ZipFile# 調用了之前定義的 delete_dsstore 函數，刪除指定目錄及其子目錄中的 .DS_Store 和 __MACOSX 文件夾（默認行為）。這一步確保在打包之前清理這些不必要的文件。delete_dsstore(directory)# 將傳入的 directory 路徑轉換為 pathlib.Path 對象，方便后續的路徑操作。directory = Path(directory)# 檢查 directory 是否是一個存在的目錄。if not directory.is_dir():# 如果不是，拋出 FileNotFoundError 異常，并提示目錄不存在。raise FileNotFoundError(f"Directory '{directory}' does not exist.") # 目錄“{directory}”不存在。# Unzip with progress bar# 使用列表推導式 生成需要打包的文件列表 。# directory.rglob("*") ：遞歸查找目錄及其子目錄中的所有文件和文件夾。# f.is_file() ：確保只選擇文件，排除文件夾。# all(x not in f.name for x in exclude) ：確保文件名中不包含 exclude 列表中的任何字符串（例如 .DS_Store 或 __MACOSX ）。files_to_zip = [f for f in directory.rglob("*") if f.is_file() and all(x not in f.name for x in exclude)]# 使用 Path.with_suffix 方法，將目錄路徑的擴展名替換為 .zip ，生成 ZIP 文件的路徑。zip_file = directory.with_suffix(".zip")# 根據 compress 參數的值選擇壓縮方式。# 如果 compress=True ，使用 ZIP_DEFLATED （壓縮）。# 如果 compress=False ，使用 ZIP_STORED （不壓縮）。compression = ZIP_DEFLATED if compress else ZIP_STORED# ZipFile(file, mode='r', compression=ZIP_STORED, allowZip64=True, compresslevel=None, *, strict_timestamps=True)# ZipFile 類是Python標準庫 zipfile 模塊中的一個類，用于讀取和寫入ZIP文件。# 參數解釋 ：# file ：可以是一個文件名（字符串），也可以是一個類文件對象（如 io.BytesIO ）。如果是一個文件名， ZipFile 將打開這個文件進行讀寫操作；如果是一個類文件對象， ZipFile 將使用這個對象進行操作。# mode ：指定打開文件的模式，默認為 'r' ，表示只讀模式。其他常用模式包括 ： 'w' 寫入模式，如果文件已存在，將被覆蓋。 'a' 追加模式，如果文件已存在，將在文件末尾追加內容。 'r+' 讀寫模式，文件必須已存在。# compression ：指定壓縮算法，默認為 ZIP_STORED ，表示不壓縮。常用的壓縮算法包括 ： ZIP_STORED 不壓縮，直接存儲文件。 ZIP_DEFLATED 使用DEFLATE算法壓縮文件，這是最常用的壓縮算法。 ZIP_BZIP2 使用BZIP2算法壓縮文件。 ZIP_LZMA 使用LZMA算法壓縮文件。# allowZip64 ：布爾值，默認為 True ，表示是否允許使用ZIP64擴展。ZIP64擴展允許創建和讀取大于4GB的ZIP文件。# compresslevel ：指定壓縮級別，僅在使用 ZIP_DEFLATED 、 ZIP_BZIP2 或 ZIP_LZMA 壓縮算法時有效。取值范圍通常為0（無壓縮）到9（最高壓縮）。# strict_timestamps ：布爾值，默認為 True ，表示是否嚴格處理文件時間戳。如果為 True ，將確保文件時間戳在ZIP文件中準確表示。# 常用方法 ：# write(filename, arcname=None, compress_type=None, compresslevel=None) ：將文件 filename 寫入ZIP文件中。 arcname 可以指定文件在ZIP文件中的名稱， compress_type 可以指定壓縮算法， compresslevel 可以指定壓縮級別。# writestr(zinfo_or_arcname, data, compress_type=None, compresslevel=None) ：將字符串 data 寫入ZIP文件中。 zinfo_or_arcname 可以是一個 ZipInfo 對象或文件名， compress_type 和 compresslevel 的含義與 write 方法相同。# extract(member, path=None, pwd=None) ：從ZIP文件中提取一個成員文件到指定路徑 path 。 member 可以是一個文件名或 ZipInfo 對象， pwd 可以指定解壓密碼。# extractall(path=None, members=None, pwd=None) ：從ZIP文件中提取所有成員文件到指定路徑 path 。 members 可以是一個文件名列表， pwd 可以指定解壓密碼。# close() ：關閉ZIP文件，釋放資源。# ZipFile 類提供了豐富的功能來處理ZIP文件，包括創建、寫入、讀取和提取文件。通過合理使用其參數和方法，可以方便地進行文件的壓縮和解壓操作。# 使用 ZipFile 類創建一個 ZIP 文件，模式為 "w" （寫入模式），并指定壓縮方式。 使用 with 語句確保 ZIP 文件在操作完成后正確關閉。with ZipFile(zip_file, "w", compression) as f:# 遍歷 files_to_zip 列表中的每個文件。 使用 TQDM （是一個進度條庫）顯示進度條。 desc 進度條的描述信息。 unit 進度條的單位（這里是文件）。 disable 根據 progress 參數的值決定是否啟用進度條。for file in TQDM(files_to_zip, desc=f"Zipping {directory} to {zip_file}...", unit="file", disable=not progress):# 將當前文件 file 寫入 ZIP 文件中。# file.relative_to(directory) ：將文件路徑轉換為相對于目錄的相對路徑，避免在 ZIP 文件中包含絕對路徑。f.write(file, file.relative_to(directory))# 返回生成的 ZIP 文件的路徑。return zip_file # return path to zip file
# 這段代碼實現了一個功能強大的目錄打包功能，支持以下特性。文件清理：在打包前自動刪除指定的文件（如 .DS_Store 和 __MACOSX ）。壓縮選項：可以選擇是否對文件進行壓縮。文件排除：支持排除指定的文件名。進度條顯示：在打包過程中顯示進度條，增強用戶體驗。路徑處理：使用 pathlib 模塊處理路徑，代碼更加簡潔且跨平臺兼容。該函數適用于需要將目錄打包為 ZIP 文件的場景，尤其適合處理包含大量文件的目錄，同時提供靈活的配置選項。5.def unzip_file(file, path=None, exclude=(".DS_Store", "__MACOSX"), exist_ok=False, progress=True):?
# 這段代碼定義了一個名為 unzip_file 的函數，用于解壓ZIP文件到指定目錄。它提供了多種功能，包括排除特定文件、處理路徑安全問題、檢查目標目錄是否已存在等。
# 定義了 unzip_file 函數，它接受以下參數 ：
# 1.file ：ZIP文件的路徑。
# 2.path ：解壓目標目錄，默認為ZIP文件的父目錄。
# 3.exclude ：一個元組，包含需要排除的文件名或目錄名，默認為 (".DS_Store", "__MACOSX") 。
# 4.exist_ok ：一個布爾值，指示如果目標目錄已存在且不為空時是否跳過解壓，默認為 False 。
# 5.progress ：一個布爾值，指示是否顯示解壓進度條，默認為 True 。
def unzip_file(file, path=None, exclude=(".DS_Store", "__MACOSX"), exist_ok=False, progress=True):# 將 *.zip 文件解壓到指定路徑，排除包含排除列表中的字符串的文件。# 如果 zip 文件不包含單個頂級目錄，則該函數將創建一個與 zip 文件同名（不帶擴展名）的新目錄來提取其內容。# 如果未提供路徑，則該函數將使用 zip 文件的父目錄作為默認路徑。# 引發：# BadZipFile：如果提供的文件不存在或不是有效的 zip 文件。"""Unzips a *.zip file to the specified path, excluding files containing strings in the exclude list.If the zipfile does not contain a single top-level directory, the function will create a newdirectory with the same name as the zipfile (without the extension) to extract its contents.If a path is not provided, the function will use the parent directory of the zipfile as the default path.Args:file (str | Path): The path to the zipfile to be extracted.path (str, optional): The path to extract the zipfile to. Defaults to None.exclude (tuple, optional): A tuple of filename strings to be excluded. Defaults to ('.DS_Store', '__MACOSX').exist_ok (bool, optional): Whether to overwrite existing contents if they exist. Defaults to False.progress (bool, optional): Whether to display a progress bar. Defaults to True.Raises:BadZipFile: If the provided file does not exist or is not a valid zipfile.Returns:(Path): The path to the directory where the zipfile was extracted.Example:```pythonfrom ultralytics.utils.downloads import unzip_filedir = unzip_file("path/to/file.zip")```"""# 這段代碼是 unzip_file 函數的一部分，用于檢查ZIP文件是否存在，并設置默認的解壓路徑。# 導入必要的模塊。# 從 zipfile 模塊中導入以下類。# BadZipFile ：表示ZIP文件損壞或無效的異常。# ZipFile ：用于操作ZIP文件的類。# is_zipfile ：一個函數，用于檢查文件是否是一個有效的ZIP文件。from zipfile import BadZipFile, ZipFile, is_zipfile# 檢查文件是否存在且是有效的ZIP文件。# 使用 Path(file).exists() 檢查文件是否存在。 使用 is_zipfile(file) 檢查文件是否是一個有效的ZIP文件。if not (Path(file).exists() and is_zipfile(file)):# 如果文件不存在或不是一個有效的ZIP文件，拋出 BadZipFile 異常，并提供錯誤信息。raise BadZipFile(f"File '{file}' does not exist or is a bad zip file.") # 文件“{file}”不存在或是一個壞的 zip 文件。# 設置默認解壓路徑。# 如果未指定解壓路徑（ path 為 None ），則將默認解壓路徑設置為ZIP文件的父目錄。if path is None:# 使用 Path(file).parent 獲取ZIP文件的父目錄路徑。path = Path(file).parent # default path# 這段代碼的主要功能是。檢查文件是否存在且是有效的ZIP文件：確保文件存在且是一個有效的ZIP文件。如果文件不存在或不是一個有效的ZIP文件，拋出 BadZipFile 異常。設置默認解壓路徑：如果未指定解壓路徑，則默認為ZIP文件的父目錄。通過這些檢查，可以確保在解壓文件之前，文件是有效的，并且解壓路徑是合理的。這有助于避免因文件損壞或路徑錯誤導致的問題。# 這段代碼是 unzip_file 函數的一部分，用于解壓ZIP文件的內容，并根據ZIP文件的結構決定解壓的目標路徑。# Unzip the file contents# 使用 ZipFile 類打開指定的ZIP文件，并將其存儲在變量 zipObj 中。 with 語句確保文件在操作完成后正確關閉。with ZipFile(file) as zipObj:# 使用列表推導式從ZIP文件中獲取所有文件名（ zipObj.namelist() ）。 過濾掉包含在 exclude 列表中的文件或目錄名。 exclude 是一個包含需要排除的文件名或目錄名的元組，默認為 (".DS_Store", "__MACOSX") 。files = [f for f in zipObj.namelist() if all(x not in f for x in exclude)]# 使用集合推導式提取所有文件的頂級目錄名稱。 Path(f).parts[0] 獲取每個文件路徑的第一部分，即頂級目錄名。top_level_dirs = {Path(f).parts[0] for f in files}# Decide to unzip directly or unzip into a directory# 判斷ZIP文件中是否只有一個頂級目錄。 如果是，則將 unzip_as_dir 設置為 True 。 如果ZIP文件中有多個頂級文件或目錄，則將 unzip_as_dir 設置為 False 。unzip_as_dir = len(top_level_dirs) == 1 # (len(files) > 1 and not files[0].endswith("/"))# 如果ZIP文件中只有一個頂級目錄（ unzip_as_dir 為 True ）。if unzip_as_dir:# Zip has 1 top-level directory# 將 解壓路徑 設置為指定的 path 。extract_path = path # i.e. ../datasets# 將 目標路徑 設置為 path 與頂級目錄名的組合。path = Path(path) / list(top_level_dirs)[0] # i.e. extract coco8/ dir to ../datasets/# 如果ZIP文件中有多個頂級文件或目錄（ unzip_as_dir 為 False ）。else:# Zip has multiple files at top level# 將 解壓路徑 設置為 path 與ZIP文件名（去除擴展名）的組合。path = extract_path = Path(path) / Path(file).stem # i.e. extract multiple files to ../datasets/coco8/# 這段代碼的主要功能是。過濾文件列表：排除指定的文件或目錄。提取頂級目錄：提取ZIP文件中的頂級目錄名稱。決定解壓方式：根據ZIP文件的結構決定是直接解壓到指定路徑，還是創建一個以ZIP文件名命名的目錄。設置解壓路徑：根據決定的解壓方式設置最終的解壓路徑。通過這些邏輯，可以靈活地處理不同結構的ZIP文件，確保解壓后的文件組織合理。# 這段代碼是 unzip_file 函數的一部分，用于檢查目標目錄是否已存在且包含文件，并根據需要解壓ZIP文件。# Check if destination directory already exists and contains files# 檢查目標目錄是否已存在且包含文件。# 檢查目標目錄 path 是否存在且包含文件。if path.exists() and any(path.iterdir()) and not exist_ok:# If it exists and is not empty, return the path without unzipping# 如果目標目錄已存在且不為空，并且 exist_ok 參數為 False ，則記錄一條警告信息。LOGGER.warning(f"WARNING ?? Skipping {file} unzip as destination directory {path} is not empty.") # 警告 ?? 跳過 {file} 解壓縮，因為目標目錄 {path} 不為空。# 返回目標路徑，跳過解壓操作。return path# 解壓文件。# 使用 TQDM （一個進度條庫）顯示解壓進度條。# 遍歷ZIP文件中的所有文件 files ，并為每個文件提供描述信息和進度單位。# 如果 progress 參數為 False ，則禁用進度條。for f in TQDM(files, desc=f"Unzipping {file} to {Path(path).resolve()}...", unit="file", disable=not progress):# Ensure the file is within the extract_path to avoid path traversal security vulnerability# 檢查文件路徑是否包含 .. ，這可能表示路徑遍歷攻擊（path traversal）。if ".." in Path(f).parts:# 如果文件路徑不安全，記錄一條警告信息。LOGGER.warning(f"Potentially insecure file path: {f}, skipping extraction.") # 可能不安全的文件路徑：{f}，跳過提取。# 并跳過該文件的解壓。continue# 使用 zipObj.extract 方法將文件 f 解壓到目標路徑 extract_path 。zipObj.extract(f, extract_path)# 返回解壓后的目錄路徑。return path # return unzip dir# 這段代碼的主要功能是。檢查目標目錄是否已存在且包含文件：如果目標目錄已存在且不為空，并且 exist_ok 為 False ，則跳過解壓操作。解壓文件：使用 TQDM 顯示解壓進度條（如果啟用）。檢查文件路徑是否安全，避免路徑遍歷攻擊。解壓文件到目標路徑。返回解壓目錄：返回解壓后的目錄路徑。通過這些邏輯，可以確保在解壓文件時處理常見的問題，如目標目錄已存在和路徑安全問題，從而提高代碼的健壯性和安全性。
# 這段代碼的主要功能是。檢查ZIP文件是否存在：確保文件存在且是一個有效的ZIP文件。設置默認解壓路徑：如果未指定解壓路徑，則默認為ZIP文件的父目錄。解壓文件：排除指定的文件或目錄。根據ZIP文件的結構決定解壓方式。檢查目標目錄是否已存在：如果目標目錄已存在且不為空，并且 exist_ok 為 False ，則跳過解壓。解壓文件：確保文件路徑安全，避免路徑遍歷漏洞。顯示解壓進度條（如果啟用）。返回解壓目錄：返回解壓后的目錄路徑。通過這個函數，可以方便地解壓ZIP文件，同時處理多種常見問題，如路徑安全和目標目錄存在性檢查。6.def check_disk_space(url="https://ultralytics.com/assets/coco8.zip", path=Path.cwd(), sf=1.5, hard=True):?
# 這段代碼定義了一個名為 check_disk_space 的函數，用于檢查是否有足夠的磁盤空間來下載指定URL的文件。# current_working_directory = Path.cwd()
# 在Python中， cwd() 函數是 pathlib 模塊中的一個方法，用于獲取當前工作目錄。這個方法是 Path 類的一個實例方法， Path 類是 pathlib 模塊中用于處理文件系統路徑的類。
# 功能描述 :
# Path.cwd() 方法返回一個 Path 對象，該對象代表當前工作目錄的路徑。
# 返回值 ：
# Path.cwd() 方法返回的是 Path 對象，這個對象提供了許多方法來操作路徑，例如 .resolve() 可以獲取路徑的絕對路徑， .as_posix() 可以將路徑轉換為跨平臺的字符串形式等。
# 異常處理 ：
# Path.cwd() 方法通常不會拋出異常，因為它只是返回當前工作目錄的路徑。但是，如果系統出現問題，導致無法確定當前工作目錄，可能會拋出異常，這種情況下應該進行異常處理。# 定義了 check_disk_space 函數，它接受以下參數 ：
# 1.url ：文件的URL。
# 2.path ：檢查磁盤空間的路徑，默認為當前工作目錄（ Path.cwd() ）。
# 3.sf ：安全因子，默認為1.5，表示需要的空間是文件大小的1.5倍。
# 4.hard ：是否嚴格檢查磁盤空間，默認為 True 。
def check_disk_space(url="https://ultralytics.com/assets/coco8.zip", path=Path.cwd(), sf=1.5, hard=True):# 檢查是否有足夠的磁盤空間來下載和存儲文件。"""Check if there is sufficient disk space to download and store a file.Args:url (str, optional): The URL to the file. Defaults to 'https://ultralytics.com/assets/coco8.zip'.path (str | Path, optional): The path or drive to check the available free space on.sf (float, optional): Safety factor, the multiplier for the required free space. Defaults to 1.5.hard (bool, optional): Whether to throw an error or not on insufficient disk space. Defaults to True.Returns:(bool): True if there is sufficient disk space, False otherwise."""# 檢查URL的有效性。try:# 使用 requests.head 發送一個HEAD請求到指定的URL，獲取響應。r = requests.head(url) # response# 檢查響應狀態碼是否小于400（表示請求成功）。assert r.status_code < 400, f"URL error for {url}: {r.status_code} {r.reason}" # check response {url} 的 URL 錯誤：{r.status_code} {r.reason}。# 如果請求失敗（例如網絡問題或URL無效）。except Exception:# 捕獲異常并返回 True ，表示默認有足夠的磁盤空間。return True # requests issue, default to True# Check file size# 檢查文件大小。# 定義1 GiB的字節數（ 1 << 30 ）。gib = 1 << 30 # bytes per GiB# 從響應頭中獲取 Content-Length ，表示文件的大小（以字節為單位），并將其轉換為GiB。data = int(r.headers.get("Content-Length", 0)) / gib # file size (GB)# shutil.disk_usage(path)# shutil.disk_usage(path) 是 Python 標準庫 shutil 模塊中的一個函數，用于獲取指定路徑的磁盤使用情況統計信息。# 參數 ：# path ：一個表示文件系統路徑的字符串或路徑對象。在 Windows 上，這個路徑必須代表一個目錄；在 Unix 系統上，它可以是文件或目錄。# 返回值 ：# 該函數返回一個命名元組（namedtuple），包含以下屬性 ：# total ：表示文件系統的總空間量，單位為字節。# used ：表示已使用的磁盤空間量，單位為字節。# free ：表示可用的磁盤空間量，單位為字節。# 注意事項 ：# 在 Unix 文件系統中， path 必須指向一個已掛載文件系統分區中的路徑。在這些平臺上，CPython 不會嘗試從未掛載的文件系統中獲取磁盤使用信息。# 從 Python 3.8 開始，在 Windows 上， path 可以是一個文件或目錄。 shutil.disk_usage() 提供的信息可以幫助開發者監控磁盤空間使用情況，或者在需要時優化磁盤使用。# 使用 shutil.disk_usage 獲取指定路徑的磁盤使用情況（總空間、已用空間和剩余空間），并將其轉換為GiB。total, used, free = (x / gib for x in shutil.disk_usage(path)) # bytes# 計算所需的空間（文件大小乘以安全因子 sf ）。如果所需空間小于剩余空間。if data * sf < free:# 則返回 True ，表示有足夠的磁盤空間。return True # sufficient space# Insufficient space# 構造一條警告信息，說明當前的剩余空間不足，并提示用戶需要釋放的額外空間。text = (f"WARNING ?? Insufficient free disk space {free:.1f} GB < {data * sf:.3f} GB required, " # 警告 ?? 可用磁盤空間不足 {free:.1f} GB < {data * sf:.3f} GB required，f"Please free {data * sf - free:.1f} GB additional disk space and try again." # 請釋放 {data * sf - free:.1f} GB 額外磁盤空間并重試。)# 如果 hard=True 。if hard:# 拋出 MemoryError 異常，提示用戶磁盤空間不足。raise MemoryError(text)# 如果 hard=False ，記錄一條警告信息。LOGGER.warning(text)# 返回 False ，表示磁盤空間不足。return False
# 這段代碼的主要功能是。檢查URL的有效性：發送HEAD請求，確保URL有效且可訪問。檢查文件大小：從響應頭中獲取文件的大小。檢查磁盤空間：獲取指定路徑的磁盤使用情況。判斷是否有足夠的空間：如果所需空間小于剩余空間，返回 True 。處理不足的空間：如果空間不足，根據 hard 參數的值，拋出異常或記錄警告信息。通過這個函數，可以確保在下載文件之前有足夠的磁盤空間，避免因空間不足導致的下載失敗。7.def get_google_drive_file_info(link):?
# 這段代碼定義了一個名為 get_google_drive_file_info 的函數，用于從Google Drive鏈接中提取文件信息，包括文件ID和文件名。它還處理了Google Drive的下載警告和配額限制。
# 定義了 get_google_drive_file_info 函數，它接受一個參數。
# 1.link ：Google Drive文件的共享鏈接。
def get_google_drive_file_info(link):# 檢索可共享 Google Drive 文件鏈接的直接下載鏈接和文件名。"""Retrieves the direct download link and filename for a shareable Google Drive file link.Args:link (str): The shareable link of the Google Drive file.Returns:(str): Direct download URL for the Google Drive file.(str): Original filename of the Google Drive file. If filename extraction fails, returns None.Example:```pythonfrom ultralytics.utils.downloads import get_google_drive_file_infolink = "https://drive.google.com/file/d/1cqT-cJgANNrhIHCrEufUYhQ4RqiWG_lJ/view?usp=drive_link"url, filename = get_google_drive_file_info(link)```"""# 從鏈接中提取文件ID。假設鏈接格式為 https://drive.google.com/file/d/<file_id>/view ，通過分割字符串提取 <file_id> 部分。file_id = link.split("/d/")[1].split("/view")[0]# 構造Google Drive的下載URL。drive_url = f"https://drive.google.com/uc?export=download&id={file_id}"# 初始化文件名為 None ，稍后從響應頭中提取。filename = None# Start session# 使用 requests.Session 啟動一個會話。with requests.Session() as session:# 發送一個GET請求到 drive_url ，并啟用流式傳輸（ stream=True ）。response = session.get(drive_url, stream=True)# 檢查響應內容中是否包含“quota exceeded”（配額超出）。if "quota exceeded" in str(response.content.lower()):# 如果配額超出，拋出 ConnectionError 異常，并提示用戶手動下載文件。raise ConnectionError(emojis(f"? Google Drive file download quota exceeded. " # ? Google Drive 文件下載配額已超出。f"Please try again later or download this file manually at {link}." # 請稍后重試或從 {link} 手動下載此文件。))# 遍歷響應中的Cookie，查找以 download_warning 開頭的鍵。for k, v in response.cookies.items():# 如果找到。if k.startswith("download_warning"):# 將 confirm 參數添加到下載URL中，以繞過Google Drive的下載警告。drive_url += f"&confirm={v}" # v is token# 從響應頭中提取 Content-Disposition 字段。if cd := response.headers.get("content-disposition"):# 使用正則表達式 re.findall 提取文件名。filename = re.findall('filename="(.+)"', cd)[0]# 返回 更新后的下載URL 和 文件名 。return drive_url, filename
# 這段代碼的主要功能是。提取文件ID：從Google Drive共享鏈接中提取文件ID。構造下載URL：構造用于下載文件的URL。檢查配額限制：檢查是否達到Google Drive的下載配額，如果達到，提示用戶手動下載。處理下載警告：如果存在下載警告，通過添加 confirm 參數繞過警告。提取文件名：從響應頭中提取文件名。返回文件信息：返回更新后的下載URL和文件名。通過這個函數，可以方便地從Google Drive鏈接中提取文件信息，并處理常見的下載問題，如配額限制和下載警告。8.def safe_download(url, file=None, dir=None, unzip=True, delete=False, curl=False, retry=3, min_bytes=1e0, exist_ok=False, progress=True,):?
# 這段代碼定義了一個名為 safe_download 的函數，用于安全地下載文件，并提供多種功能，如解壓、刪除下載文件、重試機制等。
# 定義了 safe_download 函數，它接受以下參數 ：
# 1.url ：文件的URL。
# 2.file ：目標文件名，默認為 None 。
# 3.dir ：目標目錄，默認為 None 。
# 4.unzip ：是否解壓下載的文件，默認為 True 。
# 5.delete ：下載后是否刪除原始文件，默認為 False 。
# 6.curl ：是否使用 curl 進行下載，默認為 False 。
# 7.retry ：下載失敗時的重試次數，默認為3。
# 8.min_bytes ：文件的最小字節數，用于檢查文件是否下載完整，默認為 1e0 （1字節）。
# 9.exist_ok ：如果目標文件已存在且不為空，是否跳過下載，默認為 False 。
# 10.progress ：是否顯示下載進度條，默認為 True 。
def safe_download(url,file=None,dir=None,unzip=True,delete=False,curl=False,retry=3,min_bytes=1e0,exist_ok=False,progress=True,
):# 從 URL 下載文件，帶有重試、解壓和刪除已下載文件的選項。"""Downloads files from a URL, with options for retrying, unzipping, and deleting the downloaded file.Args:url (str): The URL of the file to be downloaded.file (str, optional): The filename of the downloaded file.If not provided, the file will be saved with the same name as the URL.dir (str, optional): The directory to save the downloaded file.If not provided, the file will be saved in the current working directory.unzip (bool, optional): Whether to unzip the downloaded file. Default: True.delete (bool, optional): Whether to delete the downloaded file after unzipping. Default: False.curl (bool, optional): Whether to use curl command line tool for downloading. Default: False.retry (int, optional): The number of times to retry the download in case of failure. Default: 3.min_bytes (float, optional): The minimum number of bytes that the downloaded file should have, to be considereda successful download. Default: 1E0.exist_ok (bool, optional): Whether to overwrite existing contents during unzipping. Defaults to False.progress (bool, optional): Whether to display a progress bar during the download. Default: True.Example:```pythonfrom ultralytics.utils.downloads import safe_downloadlink = "https://ultralytics.com/assets/bus.jpg"path = safe_download(link)```"""# 檢查是否為Google Drive鏈接。# 檢查URL是否以`https://drive.google.com/`開頭。gdrive = url.startswith("https://drive.google.com/") # check if the URL is a Google Drive link# 如果是，則調用`get_google_drive_file_info`函數獲取文件信息，包括更新后的URL和文件名。if gdrive:# def get_google_drive_file_info(link): -> 用于從Google Drive鏈接中提取文件信息，包括文件ID和文件名。它還處理了Google Drive的下載警告和配額限制。返回 更新后的下載URL 和 文件名 。 -> return drive_url, filenameurl, file = get_google_drive_file_info(url)# 如果未指定目標目錄 dir ，則默認為當前目錄（ "." ）。 如果未指定文件名 file ，則調用 url2file 函數從URL中提取文件名。 構造目標文件路徑 f 。# def url2file(url): -> 用于將URL轉換為文件名。使用 clean_url(url) 清理URL，去除查詢參數（如 ?auth ）。 將清理后的URL字符串轉換為 Path 對象。 使用 Path 對象的 .name 屬性提取文件名部分。 -> return Path(clean_url(url)).namef = Path(dir or ".") / (file or url2file(url)) # URL converted to filename# 檢查文件是否存在。如果URL中不包含 :// 且路徑指向的文件存在。if "://" not in str(url) and Path(url).is_file(): # URL exists ('://' check required in Windows Python<3.10)# 則直接使用該路徑作為目標文件路徑。f = Path(url) # filename# 這段代碼是 safe_download 函數的一部分，用于處理文件下載邏輯。如果目標文件不存在，它會準備下載文件，并確保目標目錄存在且有足夠的磁盤空間。# 檢查目標文件 f 是否存在。如果文件不存在，則繼續執行下載邏輯。elif not f.is_file(): # URL and file do not exist# 如果 gdrive 為 True （即URL是Google Drive鏈接），直接使用 url 。 否則，調用 clean_url 函數清理URL，去除查詢參數。 替換 https://ultralytics.com/assets/ 為 https://github.com/ultralytics/assets/releases/download/v0.0.0/ ，確保URL格式正確。uri = (url if gdrive else clean_url(url)).replace( # cleaned and aliased url"https://github.com/ultralytics/assets/releases/download/v0.0.0/","https://ultralytics.com/assets/", # assets alias)# 構造下載描述信息，說明正在從 uri 下載文件到目標路徑 f 。desc = f"Downloading {uri} to '{f}'" # 正在將 {uri} 下載至 '{f}'。# 使用 LOGGER.info 記錄下載信息。LOGGER.info(f"{desc}...")# 使用 mkdir 方法確保目標文件的父目錄存在。如果父目錄不存在，則創建所有必要的父目錄。# parents=True ：創建所有必要的父目錄。# exist_ok=True ：如果目錄已存在，不會拋出異常。f.parent.mkdir(parents=True, exist_ok=True) # make directory if missing# 調用 check_disk_space 函數，檢查目標路徑 f.parent 是否有足夠的磁盤空間來下載文件。 如果磁盤空間不足， check_disk_space 函數會拋出異常或記錄警告信息。# def check_disk_space(url="https://ultralytics.com/assets/coco8.zip", path=Path.cwd(), sf=1.5, hard=True):# -> 用于檢查是否有足夠的磁盤空間來下載指定URL的文件。 捕獲異常并返回 True ，表示默認有足夠的磁盤空間。返回 True ，表示有足夠的磁盤空間。返回 False ，表示磁盤空間不足。# -> return True / return True / return Falsecheck_disk_space(url, path=f.parent)# 這段代碼的主要功能是。檢查文件是否存在：如果目標文件不存在，則繼續執行下載邏輯。清理和替換URL：清理URL，去除查詢參數，并確保URL格式正確。打印下載描述信息：記錄下載信息，說明正在從哪個URL下載文件到哪個目標路徑。確保目標目錄存在：? 確保目標文件的父目錄存在，如果不存在則創建。檢查磁盤空間：檢查目標路徑是否有足夠的磁盤空間來下載文件。通過這些步驟，可以確保在下載文件之前，目標路徑存在且有足夠的磁盤空間，從而避免因路徑不存在或磁盤空間不足導致的下載失敗。# 這段代碼是 safe_download 函數的一部分，用于實現文件下載的重試機制，并根據配置選擇不同的下載方法。# 使用 for 循環實現重試機制，最多嘗試 retry + 1 次。for i in range(retry + 1):# 使用 try 塊捕獲下載過程中可能出現的異常。try:# 如果 curl 為 True 或重試次數大于0，則使用 curl 進行下載。if curl or i > 0: # curl download with retry, continue# 如果 progress 為 False ，則設置 curl 的靜默模式（ -sS ）。s = "sS" * (not progress) # silent# 使用 subprocess.run 調用 curl 命令，參數包括 。# "-#" ：顯示進度條（如果 progress 為 True ）。# f"-{s}L" ：跟隨重定向（ -L ）并設置靜默模式（ -sS ）。# url ：下載的URL。# "-o", f ：指定輸出文件路徑。# "--retry", "3" ：設置重試次數為3。# "-C", "-" ：從上次失敗的地方繼續下載。r = subprocess.run(["curl", "-#", f"-{s}L", url, "-o", f, "--retry", "3", "-C", "-"]).returncode# 檢查 curl 的返回碼 r ，確保下載成功（返回碼為0）assert r == 0, f"Curl return value {r}" # Curl 返回值 {r} 。# 如果 curl 為 False ，則選擇使用 torch 或 urllib 進行下載。else: # urllib downloadmethod = "torch"# 如果 method 為 "torch" ，則使用 torch.hub.download_url_to_file 下載文件。if method == "torch":torch.hub.download_url_to_file(url, f, progress=progress)# 否則，使用 urllib.request.urlopen 下載文件，并使用 TQDM 顯示進度條。else:# urllib.request.urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, *, cafile=None, capath=None, cadefault=False, context=None)# urllib.request.urlopen() 是 Python 標準庫 urllib 模塊中的一個函數，它用于打開一個 URL 并返回一個類似文件對象的東西，可以用來讀取從 URL 獲取的數據。這個函數是 urllib.request 模塊的一部分，該模塊提供了用于處理 URL 的函數和類。# 參數 ：# url : 要打開的 URL 字符串。# data : 可選參數，用于發送 POST 請求的數據。如果是字節字符串，將作為請求體發送。如果是字典或序列的元組，將被編碼為表單數據并發送。# timeout : 可選參數，指定等待服務器響應的超時時間（以秒為單位）。默認值為 socket._GLOBAL_DEFAULT_TIMEOUT 。# cafile : 可選參數，指定包含受信任的 CA 證書的文件路徑，用于驗證 SSL 證書。# capath : 可選參數，指定包含多個 CA 證書的目錄路徑。# cadefault : 布爾值，如果為 True ，則使用默認的 CA 證書。# context : 可選參數，用于指定 SSL 上下文。# 返回值 ：# 返回一個 urllib.response.addinfourl 對象，該對象繼承自 io.IOBase ，提供了一個文件類的接口來讀取服務器的響應。這個對象包含了響應的狀態碼、響應頭等信息。# 使用 urllib.request.urlopen 打開指定的URL，獲取響應對象 response 。 使用 TQDM （一個進度條庫）顯示下載進度條。with request.urlopen(url) as response, TQDM(# response.getheader(name)# response.getheader() 是 Python 標準庫 http.client 模塊中的一個方法，用于從 HTTP 響應中獲取指定的頭信息。# 參數 ：# name ：要獲取的響應頭的名稱（如 "Content-Length" 、 "Content-Type" 等）。# "Content-Length" ：響應頭表示響應體的長度，以字節為單位。它告訴客戶端響應體的大小，以便客戶端可以預先分配足夠的緩沖區來接收數據。# "Content-Type" ：響應頭表示響應體的媒體類型（MIME類型）。它告訴客戶端響應體的內容格式，以便客戶端可以正確解析和顯示響應內容。# 返回值 ：# 返回指定響應頭的值。如果響應中不包含該頭，則返回 None 。# 功能 ：# 從 HTTP 響應中獲取指定的頭信息。如果響應中包含多個相同名稱的頭，則返回第一個頭的值。# 從響應頭中獲取 Content-Length 字段，表示文件的總大小（以字節為單位）。 如果 Content-Length 字段不存在，則默認值為0。total=int(response.getheader("Content-Length", 0)),# 進度條的描述信息，說明正在下載的文件。desc=desc,# 是否禁用進度條。如果 progress 為 False ，則禁用進度條。disable=not progress,# 進度條的單位，設置為 "B" 表示字節。unit="B",# 是否自動選擇合適的單位（如KB、MB、GB）。unit_scale=True,# 單位轉換的基數，設置為1024（表示1KB = 1024字節）。unit_divisor=1024,) as pbar:# 使用 open 函數以二進制寫模式打開目標文件 f 。with open(f, "wb") as f_opened:# 打開HTTP連接并獲取響應 ：使用 urllib.request.urlopen 或 requests.get 打開指定的URL，獲取HTTP響應對象。# 獲取文件的總大小 ：從響應頭中獲取 Content-Length 字段，表示文件的總大小（以字節為單位）。# 逐塊讀取數據 ：使用 for data in response: 逐塊讀取響應體中的數據。 每一塊數據是一個字節字符串（ bytes ）。# 寫入文件并更新進度條 ：將每一塊數據寫入目標文件。 使用 tqdm 更新進度條，顯示下載進度。# 遍歷響應對象 response 中的數據塊 data 。# 這行代碼通常用于從HTTP響應對象中逐塊讀取數據。這種用法在處理HTTP響應時非常常見，尤其是在下載文件或處理大量數據時。# response 是一個HTTP響應對象，通常是通過 urllib.request.urlopen 或其他HTTP客戶端庫（如 requests ）獲取的。# for data in response: 會逐塊讀取響應體中的數據。每一塊數據通常是一個字節字符串（ bytes ）。# 這種逐塊讀取的方式可以有效地處理大文件，避免一次性將整個文件內容加載到內存中，從而節省內存。# for data in response: 這行代碼的主要功能是。逐塊讀取HTTP響應數據：逐塊讀取響應體中的數據，每一塊數據是一個字節字符串。處理大文件：逐塊讀取的方式可以有效處理大文件，避免一次性將整個文件內容加載到內存中，從而節省內存。寫入文件并更新進度條：將每一塊數據寫入目標文件，并使用進度條庫（如 tqdm ）顯示下載進度。通過這種方式，可以高效地下載文件，并提供用戶友好的進度反饋。for data in response:# 使用 f_opened.write(data) 將數據塊寫入目標文件。f_opened.write(data)# 使用 pbar.update(len(data)) 更新進度條，表示已寫入的數據量。pbar.update(len(data))# 這段代碼的主要功能是。重試機制：最多嘗試 retry + 1 次下載。選擇下載方法：根據配置選擇使用 curl 、 torch 或 urllib 進行下載。顯示下載進度：如果 progress 為 True ，顯示下載進度條。檢查文件是否下載成功：檢查文件是否存在且大小大于 min_bytes ，確保文件下載完整。如果文件下載不完整，則刪除部分下載的文件。通過這些邏輯，可以確保文件下載過程的健壯性和可靠性，同時提供靈活的配置選項。# 這段代碼是 safe_download 函數的一部分，用于檢查文件是否成功下載。如果文件存在且大小符合要求，則認為下載成功；否則，刪除部分下載的文件。# 使用 Path.exists() 方法檢查目標文件 f 是否存在。if f.exists():# 使用 Path.stat().st_size 獲取文件的大小（以字節為單位）。 檢查文件大小是否大于 min_bytes （最小字節數，默認為1字節）。 如果文件大小大于 min_bytes ，則認為文件下載成功，退出循環。if f.stat().st_size > min_bytes:break # success# 如果文件大小不符合要求，使用 Path.unlink() 方法刪除部分下載的文件，以避免保留不完整的文件。f.unlink() # remove partial downloads# 這段代碼的主要功能是。檢查文件是否存在：確保目標文件 f 是否存在。檢查文件大小是否符合要求：檢查文件大小是否大于 min_bytes ，確保文件下載完整。刪除部分下載的文件：如果文件大小不符合要求，刪除部分下載的文件，避免保留不完整的文件。通過這些步驟，可以確保下載的文件是完整的，從而提高代碼的健壯性和可靠性。# 這段代碼是 safe_download 函數的一部分，用于處理下載過程中可能出現的異常。它根據異常類型和重試次數決定如何處理下載失敗的情況。# 使用 try-except 塊捕獲下載過程中可能出現的異常，并將其存儲在變量 e 中。except Exception as e:# 如果是第一次嘗試（ i == 0 ）且當前環境不在線（ is_online() 返回 False ）。if i == 0 and not is_online():# 則拋出 ConnectionError 異常，提示用戶環境不在線。使用 emojis 函數添加表情符號，使錯誤信息更直觀。使用 from e 將原始異常 e 作為上下文信息傳遞，方便調試。raise ConnectionError(emojis(f"? Download failure for {uri}. Environment is not online.")) from e # ? {uri} 下載失敗。環境不在線。# 如果重試次數已達到或超過最大重試次數（ i >= retry ）。elif i >= retry:# 則拋出 ConnectionError 異常，提示用戶重試次數已達到上限。 同樣使用 emojis 函數添加表情符號，并將原始異常 e 作為上下文信息傳遞。raise ConnectionError(emojis(f"? Download failure for {uri}. Retry limit reached.")) from e # ? {uri} 下載失敗。已達到重試限制。# 如果上述條件都不滿足，則記錄一條警告信息，說明當前下載失敗，并提示將進行下一次重試。使用 LOGGER.warning 記錄警告信息，并使用 emojis 函數添加表情符號，使警告信息更直觀。LOGGER.warning(f"?? Download failure, retrying {i + 1}/{retry} {uri}...") # ?? 下載失敗，重試 {i + 1}/{retry} {uri}...# 這段代碼的主要功能是。捕獲下載過程中的異常：捕獲所有可能的異常，并根據異常類型和重試次數決定如何處理。檢查網絡狀態：如果第一次嘗試失敗且當前環境不在線，拋出 ConnectionError 異常，提示用戶環境不在線。檢查重試次數：如果重試次數已達到或超過最大重試次數，拋出 ConnectionError 異常，提示用戶重試次數已達到上限。記錄警告并重試：如果上述條件都不滿足，記錄一條警告信息，說明當前下載失敗，并提示將進行下一次重試。通過這些邏輯，可以確保在下載過程中遇到問題時，能夠提供清晰的錯誤信息，并根據情況決定是否重試，從而提高代碼的健壯性和用戶體驗。# 這段代碼是 safe_download 函數的一部分，用于在文件下載完成后，根據需要解壓文件，并處理解壓后的文件。# 檢查是否需要解壓文件（ unzip 為 True ）。# 檢查文件是否存在（ f.exists() ）。# 檢查文件的后綴是否為ZIP、TAR或GZ文件（ f.suffix in {"", ".zip", ".tar", ".gz"} ）。if unzip and f.exists() and f.suffix in {"", ".zip", ".tar", ".gz"}:# 導入 zipfile 模塊中的 is_zipfile 函數，用于檢查文件是否為ZIP文件。from zipfile import is_zipfile# 確定解壓目錄 unzip_dir 。如果提供了 dir 參數，則解壓到指定目錄。 如果未提供 dir 參數，則解壓到文件的父目錄。unzip_dir = (dir or f.parent).resolve() # unzip to dir if provided else unzip in place# 使用 is_zipfile 檢查文件是否為ZIP文件。if is_zipfile(f):# 如果是ZIP文件，調用 unzip_file 函數解壓文件，并返回解壓后的目錄路徑。# def unzip_file(file, path=None, exclude=(".DS_Store", "__MACOSX"), exist_ok=False, progress=True):# -> 用于解壓ZIP文件到指定目錄。它提供了多種功能，包括排除特定文件、處理路徑安全問題、檢查目標目錄是否已存在等。返回目標路徑，跳過解壓操作。返回解壓后的目錄路徑。# -> return path / return path # return unzip dirunzip_dir = unzip_file(file=f, path=unzip_dir, exist_ok=exist_ok, progress=progress) # unzip# 如果文件是TAR或GZ文件，使用 subprocess.run 調用 tar 命令解壓文件。elif f.suffix in {".tar", ".gz"}:LOGGER.info(f"Unzipping {f} to {unzip_dir}...") # 正在將 {f} 解壓縮至 {unzip_dir}...# 這行代碼使用 subprocess.run 調用了系統的 tar 命令來解壓TAR或GZ文件。# subprocess.run ：這是Python標準庫中的一個函數，用于在子進程中運行命令。# ["tar", ...] ：這是要運行的命令及其參數的列表。 "tar" 是命令名稱，后面的元素是命令的參數。# "xf" ：這是 tar 命令的選項，表示解壓文件。 x 表示解壓， f 表示指定文件。# "xfz" ：這是 tar 命令的選項，表示解壓GZ文件。 x 表示解壓， f 表示指定文件， z 表示文件是gzip壓縮的。# f ：這是要解壓的文件路徑。# "--directory" ：這是 tar 命令的選項，指定解壓的目標目錄。# unzip_dir ：這是解壓的目標目錄路徑。# check=True ：這個參數指示 subprocess.run 在命令執行失敗時拋出異常。如果命令執行成功，返回值的 returncode 屬性為0；如果命令執行失敗， returncode 屬性為非0值。subprocess.run(["tar", "xf" if f.suffix == ".tar" else "xfz", f, "--directory", unzip_dir], check=True)# 如果 delete 為 True ，則刪除原始文件（ZIP、TAR或GZ文件）。if delete:f.unlink() # remove zip# 返回解壓后的目錄路徑，方便后續操作。return unzip_dir# 這段代碼的主要功能是。檢查是否需要解壓文件：確保文件存在且是ZIP、TAR或GZ文件。確定解壓目錄：如果提供了目標目錄，則解壓到指定目錄；否則，解壓到文件的父目錄。解壓ZIP文件：使用 unzip_file 函數解壓ZIP文件。解壓TAR或GZ文件：使用 tar 命令解壓TAR或GZ文件。刪除原始文件：如果 delete 為 True ，刪除解壓后的原始文件。返回解壓后的目錄路徑：返回解壓后的目錄路徑，方便后續操作。通過這些步驟，可以確保下載的文件在需要時被正確解壓，并且可以選擇性地刪除原始文件，從而提高代碼的靈活性和用戶體驗。
# 這段代碼的主要功能是。檢查是否為Google Drive鏈接：如果是，調用 get_google_drive_file_info 獲取文件信息。確定目標文件路徑：根據提供的目錄和文件名或從URL中提取文件名。檢查文件是否存在：如果文件已存在，跳過下載。下載文件：使用 curl 或 torch.hub.download_url_to_file 或 urllib.request.urlopen 下載文件。提供重試機制和進度條。檢查文件是否下載成功：檢查文件大小是否大于 min_bytes ，確保文件下載完整。解壓文件：如果文件是ZIP、TAR或GZ文件，解壓到指定目錄。如果 delete 為 True ，刪除解壓后的原始文件。返回解壓后的目錄路徑：返回解壓后的目錄路徑，方便后續操作。通過這個函數，可以安全地下載文件，并處理多種常見問題，如文件存在性檢查、解壓和刪除原始文件等。
# def safe_download(url, file=None, dir=None, unzip=True, delete=False, curl=False, retry=3, min_bytes=1e0, exist_ok=False, progress=True,):
# -> 用于安全地下載文件，并提供多種功能，如解壓、刪除下載文件、重試機制等。返回解壓后的目錄路徑，方便后續操作。
# -> return unzip_dir9.def get_github_assets(repo="ultralytics/assets", version="latest", retry=False):?
# 這段代碼定義了一個名為 get_github_assets 的函數，用于從GitHub倉庫中獲取指定版本的資產（如模型文件）。
# 定義了 get_github_assets 函數，它接受以下參數 ：
# 1.repo ：GitHub倉庫的名稱，默認為 "ultralytics/assets" 。
# 2.version ：要獲取資產的版本，默認為 "latest" 。
# 3.retry ：是否在失敗時重試，默認為 False 。
def get_github_assets(repo="ultralytics/assets", version="latest", retry=False):# 從 GitHub 存儲庫中檢索指定版本的標簽和資產。如果未指定版本，該函數將獲取最新版本的資產。"""Retrieve the specified version's tag and assets from a GitHub repository. If the version is not specified, thefunction fetches the latest release assets.Args:repo (str, optional): The GitHub repository in the format 'owner/repo'. Defaults to 'ultralytics/assets'.version (str, optional): The release version to fetch assets from. Defaults to 'latest'.retry (bool, optional): Flag to retry the request in case of a failure. Defaults to False.Returns:(tuple): A tuple containing the release tag and a list of asset names.Example:```pythontag, assets = get_github_assets(repo="ultralytics/assets", version="latest")```"""# 如果 version 不是 "latest" ，則將其格式化為 "tags/<version>" ，例如 "tags/v6.2" 。if version != "latest":version = f"tags/{version}" # i.e. tags/v6.2# 構造GitHub API的URL。url = f"https://api.github.com/repos/{repo}/releases/{version}"# 使用 requests.get 發送GET請求到GitHub API，獲取指定版本的資產信息。r = requests.get(url) # github api# 檢查響應狀態碼是否為200（成功）。# 如果狀態碼不是200且不是因為GitHub的速率限制（ rate limit exceeded ），并且 retry 為 True 。if r.status_code != 200 and r.reason != "rate limit exceeded" and retry: # failed and not 403 rate limit exceeded# 則再次發送GET請求。r = requests.get(url) # try again# 如果響應狀態碼不是200，記錄一條警告信息。if r.status_code != 200:LOGGER.warning(f"?? GitHub assets check failure for {url}: {r.status_code} {r.reason}") # ?? GitHub 資產檢查 {url} 失敗：{r.status_code} {r.reason}。# 返回 空的版本號 和 資產列表 。return "", []# 將響應內容解析為JSON格式。data = r.json()# 從JSON數據中提取 tag_name （版本號）和 assets （資產列表）。 使用列表推導式提取 每個資產的名稱 。return data["tag_name"], [x["name"] for x in data["assets"]] # tag, assets i.e. ['yolo11n.pt', 'yolov8s.pt', ...]
# 這段代碼的主要功能是。處理版本號：如果版本號不是 "latest" ，則格式化為 "tags/<version>" 。構造GitHub API URL：構造用于獲取資產信息的GitHub API URL。發送GET請求：使用 requests.get 發送GET請求，獲取指定版本的資產信息。處理請求失敗：如果請求失敗且不是因為速率限制，并且啟用了重試機制，則再次發送請求。檢查響應狀態：如果響應狀態碼不是200，記錄警告信息并返回空結果。解析響應數據：解析響應內容，提取版本號和資產列表。通過這個函數，可以方便地從GitHub倉庫中獲取指定版本的資產信息，適用于需要動態獲取模型文件或其他資源的場景。10.def attempt_download_asset(file, repo="ultralytics/assets", release="v8.3.0", **kwargs):?
# 這段代碼定義了一個名為 attempt_download_asset 的函數，用于嘗試下載指定的資產文件（如模型文件）。它會檢查文件是否已存在，如果不存在，則嘗試從指定的URL或GitHub倉庫下載。
# 定義了 attempt_download_asset 函數，它接受以下參數 ：
# 1.file ：要下載的文件名或URL。
# 2.repo ：GitHub倉庫的名稱，默認為 "ultralytics/assets" 。
# 3.release ：要下載的版本，默認為 "v8.3.0" 。
# 4.**kwargs ：其他關鍵字參數，傳遞給 safe_download 函數。
def attempt_download_asset(file, repo="ultralytics/assets", release="v8.3.0", **kwargs):# 如果在本地找不到文件，則嘗試從 GitHub 發布資產中下載文件。該函數首先在本地檢查文件，然后嘗試從指定的 GitHub 存儲庫發布中下載。"""Attempt to download a file from GitHub release assets if it is not found locally. The function checks for the filelocally first, then tries to download it from the specified GitHub repository release.Args:file (str | Path): The filename or file path to be downloaded.repo (str, optional): The GitHub repository in the format 'owner/repo'. Defaults to 'ultralytics/assets'.release (str, optional): The specific release version to be downloaded. Defaults to 'v8.3.0'.**kwargs (any): Additional keyword arguments for the download process.Returns:(str): The path to the downloaded file.Example:```pythonfile_path = attempt_download_asset("yolo11n.pt", repo="ultralytics/assets", release="latest")```"""# 從 ultralytics.utils 模塊中導入 SETTINGS ，這是一個包含全局設置的字典。from ultralytics.utils import SETTINGS # scoped for circular import# YOLOv3/5u updates# 將 file 轉換為字符串。file = str(file)# 調用 check_yolov5u_filename 函數，將舊版YOLOv5文件名替換為更新后的YOLOv5u文件名。# def check_yolov5u_filename(file: str, verbose: bool = True): -> 用于將舊版YOLOv5文件名替換為更新后的YOLOv5u文件名。它還提供了一個提示信息，建議用戶使用新的文件名，因為YOLOv5u模型在性能上有所改進。返回更新后的文件名。 -> return filefile = checks.check_yolov5u_filename(file)# 去除文件名首尾的空白字符和單引號。file = Path(file.strip().replace("'", ""))# 檢查文件是否已存在。如果存在，返回文件路徑。if file.exists():return str(file)# 如果文件不存在，檢查 SETTINGS["weights_dir"] 目錄下是否存在該文件。如果存在，返回文件路徑。elif (SETTINGS["weights_dir"] / file).exists():return str(SETTINGS["weights_dir"] / file)# 這段代碼是 attempt_download_asset 函數的一部分，用于處理指定URL的情況。如果 file 是一個URL，它會嘗試下載該URL指向的文件。# 處理URL指定的情況。else:# URL specified# parse.unquote(s, encoding='utf-8', errors='replace')# parse.unquote() 是 Python 標準庫 urllib.parse 模塊中的一個函數，用于對 URL 編碼的字符串進行解碼，將百分號編碼（%XX）轉換回普通字符。# 參數 :# s ：要解碼的 URL 編碼字符串。# encoding ：（可選）用于解碼的字符編碼，默認為 'utf-8' 。# errors ：（可選）指定如何處理解碼錯誤，默認為 'replace' ，意味著將無法解碼的字符替換為一個替代字符（通常是 ? ）。# 返回值 :# 返回解碼后的字符串。# 函數邏輯 :# 解碼百分號編碼 ：將字符串中的 %XX 序列轉換為對應的字符。# 字符編碼轉換 ：將原始的百分比編碼字符串（通常為 ASCII）轉換為指定的編碼。# 在例子中， unquote 函數將 URL 編碼的字符串 "Hello%2C%20World%21" 解碼為普通字符串 "Hello, World!" 。# 注意事項 ：# 當處理來自用戶的 URL 編碼數據時，使用 unquote 函數可以確保正確地解釋這些數據。# 如果 URL 包含非 ASCII 字符，確保指定正確的 encoding 參數，否則解碼可能會失敗或產生意外結果。# 如果遇到無法解碼的百分號序列， errors 參數決定了如何處理這些錯誤。常見的選項包括 'strict' （拋出 UnicodeDecodeError ）、 'replace' （用替代字符替換無法解碼的字符）和 'ignore' （忽略無法解碼的字符）。# 將 file 轉換為字符串，并使用 parse.unquote 解碼URL中的特殊字符（如將 %2F 解碼為 / ）。 使用 Path 對象的 .name 屬性提取文件名。name = Path(parse.unquote(str(file))).name # decode '%2F' to '/' etc.# 構造下載URL的基本路徑。download_url = f"https://github.com/{repo}/releases/download"# 檢查 file 是否以 http:/ 或 https:/ 開頭，表示它是一個HTTP URL。if str(file).startswith(("http:/", "https:/")): # download# 修正URL格式，將 (":/") 替換為 ("://") 。這是因為 Pathlib 可能會錯誤地將 :// 轉換為 :/ 。url = str(file).replace(":/", "://") # Pathlib turns :// -> :/# 調用 url2file 函數，從URL中提取文件名。這個函數通常會處理URL中的認證信息（如 ?auth... ）。# def url2file(url): -> 用于將URL轉換為文件名。使用 clean_url(url) 清理URL，去除查詢參數（如 ?auth ）。 將清理后的URL字符串轉換為 Path 對象。 使用 Path 對象的 .name 屬性提取文件名部分。 -> return Path(clean_url(url)).namefile = url2file(name) # parse authentication https://url.com/file.txt?auth...# 使用 Path.is_file() 檢查文件是否已存在。if Path(file).is_file():# 如果文件已存在，記錄一條信息，說明文件已本地存在。LOGGER.info(f"Found {clean_url(url)} locally at {file}") # file already exists 在本地的 {file} 處找到了 {clean_url(url)}。# 如果文件不存在。else:# 調用 safe_download 函數下載文件。# min_bytes=1e5 ：設置文件的最小字節數為100,000字節，確保下載的文件大小符合要求。# **kwargs ：將其他關鍵字參數傳遞給 safe_download 函數。# def safe_download(url, file=None, dir=None, unzip=True, delete=False, curl=False, retry=3, min_bytes=1e0, exist_ok=False, progress=True,):# -> 用于安全地下載文件，并提供多種功能，如解壓、刪除下載文件、重試機制等。返回解壓后的目錄路徑，方便后續操作。# -> return unzip_dirsafe_download(url=url, file=file, min_bytes=1e5, **kwargs)# 這段代碼的主要功能是。處理指定URL的情況：如果 file 是一個URL，解析URL并提取文件名。修正URL格式：修正URL格式，確保其正確性。解析文件名：從URL中提取文件名，處理認證信息。檢查文件是否已存在：如果文件已存在，記錄一條信息，說明文件已本地存在。下載文件：如果文件不存在，調用 safe_download 函數下載文件。通過這些步驟，可以確保在需要時下載指定的文件，同時避免重復下載已存在的文件，提高代碼的效率和用戶體驗。# 從GitHub倉庫下載。# 如果 repo 是 GITHUB_ASSETS_REPO 且文件名在 GITHUB_ASSETS_NAMES 中，直接從指定的版本下載文件。elif repo == GITHUB_ASSETS_REPO and name in GITHUB_ASSETS_NAMES:safe_download(url=f"{download_url}/{release}/{name}", file=file, min_bytes=1e5, **kwargs)# 這段代碼是 attempt_download_asset 函數的一部分，用于從GitHub倉庫中下載指定的資產文件。如果指定的版本中沒有找到文件，則嘗試從最新版本中下載。else:# 調用 get_github_assets 函數，嘗試從指定的GitHub倉庫 repo 和版本 release 中獲取資產列表。# tag ：版本標簽（如 v8.3.0 ）。# assets ：資產文件列表（如 ['yolo11n.pt', 'yolov8s.pt', ...] ）。# def get_github_assets(repo="ultralytics/assets", version="latest", retry=False):# -> 用于從GitHub倉庫中獲取指定版本的資產（如模型文件）。返回 空的版本號 和 資產列表 。從JSON數據中提取 tag_name （版本號）和 assets （資產列表）。 使用列表推導式提取 每個資產的名稱 。# -> return "", [] / return data["tag_name"], [x["name"] for x in data["assets"]] # tag, assets i.e. ['yolo11n.pt', 'yolov8s.pt', ...]tag, assets = get_github_assets(repo, release)# 如果指定版本中沒有資產，嘗試從最新版本獲取。# 如果指定版本中沒有找到資產（ assets 為空）。if not assets:# 則調用 get_github_assets 函數，嘗試從最新版本中獲取資產列表。tag, assets = get_github_assets(repo) # latest release# 檢查文件名 name 是否在資產列表 assets 中。if name in assets:# 如果文件名在資產列表中，調用 safe_download 函數下載文件。# url ：構造下載URL，格式為 <download_url>/<tag>/<name> 。# file ：目標文件路徑。# min_bytes=1e5 ：設置文件的最小字節數為100,000字節，確保下載的文件大小符合要求。# **kwargs ：將其他關鍵字參數傳遞給 safe_download 函數。safe_download(url=f"{download_url}/{tag}/{name}", file=file, min_bytes=1e5, **kwargs)# 這段代碼的主要功能是。從指定版本獲取資產：調用 get_github_assets 函數，嘗試從指定的版本中獲取資產列表。如果指定版本中沒有資產，嘗試從最新版本獲取：如果指定版本中沒有找到資產，嘗試從最新版本中獲取資產列表。檢查文件是否在資產列表中：檢查文件名是否在資產列表中。下載文件：如果文件名在資產列表中，調用 safe_download 函數下載文件。通過這些步驟，可以確保在需要時從GitHub倉庫中下載指定的文件，同時處理指定版本中沒有找到文件的情況，提高代碼的健壯性和靈活性。# 返回文件的路徑。return str(file)
# 這段代碼的主要功能是。檢查文件是否存在：如果文件已存在，直接返回文件路徑。處理URL指定的情況：如果 file 是一個URL，解析URL并下載文件。從GitHub倉庫下載：如果 file 是一個文件名，從指定的GitHub倉庫和版本下載文件。返回文件路徑：返回文件的路徑。通過這個函數，可以方便地嘗試下載指定的資產文件，同時處理文件已存在的情況，確保代碼的健壯性和靈活性。
# def attempt_download_asset(file, repo="ultralytics/assets", release="v8.3.0", **kwargs): -> 用于嘗試下載指定的資產文件（如模型文件）。它會檢查文件是否已存在，如果不存在，則嘗試從指定的URL或GitHub倉庫下載。返回文件的路徑。 -> return str(file)11.def download(url, dir=Path.cwd(), unzip=True, delete=False, curl=False, threads=1, retry=3, exist_ok=False):?
# 這段代碼定義了一個名為 download 的函數，用于從指定的 URL 下載文件，并支持多種功能，包括多線程下載、自動解壓、刪除源文件、使用 curl 下載等。
# 定義了一個函數 download ，接受以下參數 ：
# 1.url ：文件的下載鏈接，可以是一個字符串（單個 URL），也可以是一個列表（多個 URL）。
# 2.dir ：下載文件的目標目錄，默認為當前工作目錄（ Path.cwd() ）。
# 3.unzip ：布爾值，表示是否自動解壓下載的 ZIP 文件，默認為 True 。
# 4.delete ：布爾值，表示是否刪除下載后的源文件，默認為 False 。
# 5.curl ：布爾值，表示是否使用 curl 命令進行下載，默認為 False 。
# 6.threads ：整數，表示下載時使用的線程數，默認為 1 。
# 7.retry ：整數，表示下載失敗時的重試次數，默認為 3 。
# 8.exist_ok ：布爾值，表示在創建目標目錄時，如果目錄已存在是否拋出異常，默認為 False 。
def download(url, dir=Path.cwd(), unzip=True, delete=False, curl=False, threads=1, retry=3, exist_ok=False):# 從指定的 URL 下載文件到給定目錄。如果指定多個線程，則支持并發下載。"""Downloads files from specified URLs to a given directory. Supports concurrent downloads if multiple threads arespecified.Args:url (str | list): The URL or list of URLs of the files to be downloaded.dir (Path, optional): The directory where the files will be saved. Defaults to the current working directory.unzip (bool, optional): Flag to unzip the files after downloading. Defaults to True.delete (bool, optional): Flag to delete the zip files after extraction. Defaults to False.curl (bool, optional): Flag to use curl for downloading. Defaults to False.threads (int, optional): Number of threads to use for concurrent downloads. Defaults to 1.retry (int, optional): Number of retries in case of download failure. Defaults to 3.exist_ok (bool, optional): Whether to overwrite existing contents during unzipping. Defaults to False.Example:```pythondownload("https://ultralytics.com/assets/example.zip", dir="path/to/dir", unzip=True)```"""# 將 dir 參數轉換為 pathlib.Path 對象，方便后續路徑操作。dir = Path(dir)# 使用 Path.mkdir 方法創建目標目錄。# parents=True ：如果目標目錄的父目錄不存在，會遞歸創建。# exist_ok=True ：如果目錄已存在，不會拋出異常。dir.mkdir(parents=True, exist_ok=True) # make directory# 判斷是否使用多線程下載。如果 threads 參數大于 1 ，則使用多線程。if threads > 1:# pool = ThreadPool(processes=5)# 在Python中， multiprocessing.pool.ThreadPool 是 multiprocessing 模塊中的 pool 子模塊提供的類之一，它用于創建一個線程池，以便并行執行多個線程任務。這個類是 multiprocessing 庫的一部分，該庫提供了一種方式來并行化程序，利用多核處理器的能力。# 參數 ：# processes ：參數指定線程池中的線程數量。# 返回 ：# pool ：創建一個線程池實例。# 方法 ThreadPool 提供了以下方法 ：# apply_async(func, args=(), kwds={}) ：異步地將一個函數 func 應用到 args 和 kwds 參數上，并返回一個 AsyncResult 對象。# map(func, iterable, chunksize=1) ：將一個函數 func 映射到一個迭代器 iterable 的所有元素上，并返回一個迭代器。# close() ：關閉線程池，不再接受新的任務。# join() ：等待線程池中的所有任務完成。# terminate() ：立即終止線程池中的所有任務。# ThreadPool 是 multiprocessing 庫中用于并行執行 I/O 密集型任務的工具，它允許程序利用多核處理器的能力來提高性能。需要注意的是， multiprocessing 庫中的進程是系統級的進程，與線程相比，它們有更大的開銷，但也能提供更好的并行性能。# 使用 concurrent.futures.ThreadPool 創建一個線程池，線程數由 threads 參數指定。with ThreadPool(threads) as pool:# 使用 pool.map 方法將下載任務分配給線程池中的線程。pool.map(# 定義了一個匿名函數，調用 safe_download 函數進行實際的下載操作。# def safe_download(url, file=None, dir=None, unzip=True, delete=False, curl=False, retry=3, min_bytes=1e0, exist_ok=False, progress=True,):# -> 用于安全地下載文件，并提供多種功能，如解壓、刪除下載文件、重試機制等。返回解壓后的目錄路徑，方便后續操作。# -> return unzip_dirlambda x: safe_download(url=x[0],dir=x[1],unzip=unzip,delete=delete,curl=curl,retry=retry,exist_ok=exist_ok,# 如果線程數為 1 ，則顯示進度條。progress=threads <= 1,),# 將 url 和重復的目標目錄路徑 dir 組合成一個迭代器，為每個線程提供下載任務。zip(url, repeat(dir)),)# 關閉線程池，不再接受新的任務。pool.close()# 等待所有線程完成任務。pool.join()# 如果 threads 參數小于或等于 1 ，則不使用多線程，逐個下載文件。else:# 如果 url 是一個字符串或 Path 對象，則將其轉換為一個列表（包含單個 URL）。如果 url 已經是一個列表，則直接遍歷。for u in [url] if isinstance(url, (str, Path)) else url:# 調用 safe_download 函數逐個下載文件。safe_download(url=u, dir=dir, unzip=unzip, delete=delete, curl=curl, retry=retry, exist_ok=exist_ok)
# 這段代碼實現了一個功能豐富的文件下載工具，支持以下特性。多線程下載：通過 ThreadPool 實現多線程下載，提高下載效率。自動解壓：下載完成后自動解壓 ZIP 文件。刪除源文件：可以選擇是否刪除下載后的源文件。使用 curl 下載：支持使用 curl 命令進行下載，適用于某些特殊場景。重試機制：支持下載失敗時的重試機制。進度條顯示：在單線程下載時顯示進度條。目錄創建：自動創建目標目錄，確保下載路徑有效。該函數適用于需要從網絡下載文件的場景，尤其是需要處理多個文件或需要高效下載的場景。通過靈活的參數配置，可以滿足不同的下載需求。# curl 是一個非常強大的命令行工具，用于在不同類型的網絡協議之間傳輸數據。它支持多種協議，包括 HTTP、HTTPS、FTP、FTPS、SMTP、IMAP 等。 curl 的名字來源于“Client URL”，即客戶端 URL 工具。
# 主要用途 ：
# 文件下載 ：從服務器下載文件。
# 文件上傳 ：將文件上傳到服務器。
# API 調用 ：發送 HTTP 請求，與 RESTful API 或其他 Web 服務進行交互。
# 測試網絡連接 ：檢查服務器的響應情況。
# 模擬瀏覽器行為 ：發送帶有特定頭信息或數據的請求。
# 基本語法 ：
# curl [選項] [URL]
# 常用選項 ：
# -o <file> ：將輸出保存到指定文件。 curl -o example.txt http://example.com/file.txt
# -O ：將輸出保存到與遠程文件同名的本地文件。 curl -O http://example.com/file.txt
# -L ：跟隨 HTTP 重定向。 curl -L http://example.com/redirect
# -X <method> ：指定 HTTP 請求方法（如 GET 、 POST 、 PUT 、 DELETE ）。 curl -X POST http://example.com/api -d "key=value"
# -d <data> ：發送 HTTP POST 請求的數據。 curl -X POST http://example.com/api -d "key=value"
# -H <header> ：添加自定義 HTTP 頭。 curl -H "Content-Type: application/json" -d '{"key":"value"}' http://example.com/api
# -u <user:password> ：用于 HTTP 基本認證。 curl -u user:password http://example.com/protected
# -v ：顯示詳細的請求和響應信息，用于調試。 curl -v http://example.com
# 優點 ：
# 跨平臺 ： curl 可以在 Linux、macOS 和 Windows 上使用。
# 功能強大 ：支持多種網絡協議和復雜的請求。
# 輕量級 ：無需安裝龐大的圖形界面工具。
# 易于集成 ：可以在腳本中使用，方便自動化任務。
# 總結 ： curl 是一個非常實用的網絡工具，廣泛用于開發、測試和自動化任務中。它簡單易用，功能強大，是每個開發者和系統管理員的必備工具之一。?
![BUU24 [GXYCTF2019]BabyUpload 1](http://pic.xiahunao.cn/BUU24 [GXYCTF2019]BabyUpload 1)
