Python 多线程下载工具 - 支持 HTTP, FTP, Magnet
这是一个使用 Python 编写的命令行工具，用于下载文件。它支持 HTTP、FTP 和 Magnet 链接，并允许您使用多线程加速下载速度。
用法
python downloader.py [-h] -u <URL> -o <FILENAME> [-t <NUM_THREADS>] [-d <DIRECTORY>]
参数
-h, --help：显示帮助信息。-u <URL>, --url <URL>：指定下载的 URL，支持 HTTP、FTP、Magnet 三种方式。-o <FILENAME>, --output <FILENAME>：指定输出的文件名。-t <NUM_THREADS>, --threads <NUM_THREADS>：指定下载线程数，默认为 4。-d <DIRECTORY>, --dir <DIRECTORY>：指定输出文件的路径，默认为当前目录。
示例
- 下载文件：
python downloader.py -u https://download.pytorch.org/whl/cpu/torch-1.10.0%2Bcpu-cp38-cp38-linux_x86_64.whl -o torch.whl
- 下载文件并指定输出路径：
python downloader.py -u https://download.pytorch.org/whl/cpu/torch-1.10.0%2Bcpu-cp38-cp38-linux_x86_64.whl -o torch.whl -d /home/user/downloads
- 下载文件并指定线程数：
python downloader.py -u https://download.pytorch.org/whl/cpu/torch-1.10.0%2Bcpu-cp38-cp38-linux_x86_64.whl -o torch.whl -t 8
详细说明
该工具使用 requests 库进行下载，支持断点续传。通过 ThreadPoolExecutor 实现多线程下载，可指定线程数。下载过程中使用 tqdm 库显示进度条。
注意事项
- 当指定的输出文件名已经存在时，会在文件名后加上 '_1'、'_2'、'_3'……直到找到一个不存在的文件名。
- 当指定的输出路径不存在时，会尝试创建该路径。
- 当下载的 URL 不是以 HTTP、FTP、Magnet 开头时，会报错。
错误帮助
-
错误：'xxx' is not a directory。 原因：指定的输出路径不是一个目录。 解决：请指定一个正确的目录。
-
错误：'xxx' is not a valid url。 原因：指定的 URL 不是一个合法的 URL。 解决：请指定一个合法的 URL，支持 HTTP、FTP、Magnet 三种方式。
-
错误：Download failed: xxx。 原因：下载过程中发生了错误。 解决：请检查网络连接，或者尝试更改下载线程数。
-
错误：The url xxx does not start with http, ftp or magnet。 原因：指定的 URL 不是以 HTTP、FTP、Magnet 开头。 解决：请指定一个以 HTTP、FTP、Magnet 开头的 URL。
代码
import os
import argparse
import requests
import logging
import re
import transmissionrpc
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
def check_dir(path):
if not os.path.exists(path):
os.makedirs(path)
elif not os.path.isdir(path):
raise argparse.ArgumentTypeError(f'{path} is not a directory')
return path
def check_url(url):
pattern = r'^https?://|^ftp://|^magnet:'
if not re.match(pattern, url):
raise argparse.ArgumentTypeError(f'{url} is not a valid url')
return url
def download_file(url, filename, num_threads=4):
response = requests.head(url)
size = int(response.headers.get('Content-Length'))
chunk_size = int(size / num_threads) + 1
with open(filename, 'wb') as f:
with tqdm(total=size, unit='B', unit_scale=True, desc=os.path.basename(filename), ncols=80, position=0) as bar:
futures = []
for i in range(num_threads):
start = i * chunk_size
end = min(start + chunk_size, size)
futures.append((url, start, end, f))
with ThreadPoolExecutor(max_workers=num_threads) as executor:
for future in as_completed(executor.submit(download_range, *future, bar) for future in futures):
try:
future.result()
except Exception as e:
logging.error(f'Download failed: {e}')
logging.info(f'Download completed: {filename}')
def download_range(url, start, end, fileobj, bar):
headers = {'Range': f'bytes={start}-{end-1}'}
with requests.get(url, headers=headers, stream=True) as response:
for chunk in response.iter_content(chunk_size=1024):
if chunk:
fileobj.write(chunk)
bar.update(len(chunk))
def download_ftp(url, filename, num_threads=4):
with requests.get(url, stream=True) as response:
total_size = response.headers.get('Content-Length')
if total_size:
total_size = int(total_size.strip())
with tqdm(total=total_size, unit='B', unit_scale=True, desc=os.path.basename(filename), ncols=80, position=0) as bar:
with open(filename, 'wb') as f:
chunk_size = int(total_size / num_threads) + 1
futures = []
for i in range(num_threads):
start = i * chunk_size
end = min(start + chunk_size, total_size)
futures.append((url, start, end, f, bar))
with ThreadPoolExecutor(max_workers=num_threads) as executor:
for future in as_completed(executor.submit(download_range_ftp, *future) for future in futures):
try:
future.result()
except Exception as e:
logging.error(f'Download failed: {e}')
logging.info(f'Download completed: {filename}')
else:
with open(filename, 'wb') as f:
f.write(response.content)
logging.info(f'Download completed: {filename}')
def download_range_ftp(url, start, end, fileobj, bar):
headers = {'Range': f'bytes={start}-{end-1}'}
with requests.get(url, headers=headers, stream=True) as response:
for chunk in response.iter_content(chunk_size=1024):
if chunk:
fileobj.write(chunk)
bar.update(len(chunk))
def download_torrent(magnet, filename):
tc = transmissionrpc.Client('localhost', port=9091)
tc.add_torrent(magnet, download_dir=os.path.dirname(filename))
logging.info(f'Download started: {filename}')
def main():
parser = argparse.ArgumentParser(description='A command-line tool for downloading files.')
parser.add_argument('-u', '--url', metavar='<URL>', type=check_url, required=True, help='download url')
parser.add_argument('-o', '--output', metavar='<FILENAME>', type=str, required=True, help='output filename')
parser.add_argument('-t', '--threads', metavar='<NUM_THREADS>', type=int, default=4, help='number of threads for downloading')
parser.add_argument('-d', '--dir', metavar='<DIRECTORY>', type=check_dir, default='.', help='output directory')
args = parser.parse_args()
url, filename, num_threads, output_dir = args.url, args.output, args.threads, args.dir
filename = os.path.join(output_dir, filename)
if os.path.isfile(filename):
i = 1
while True:
new_filename = f'{os.path.splitext(filename)[0]}_{i}{os.path.splitext(filename)[1]}'
if not os.path.isfile(new_filename):
filename = new_filename
break
i += 1
if url.startswith('http'):
download_file(url, filename, num_threads)
elif url.startswith('ftp'):
download_ftp(url, filename, num_threads)
elif url.startswith('magnet'):
download_torrent(url, filename)
else:
logging.error(f'The url {url} does not start with http, ftp or magnet.')
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
main()
原文地址: https://www.cveoy.top/t/topic/oPPE 著作权归作者所有。请勿转载和采集!