Python实现批量下载文件
批量下载文件工具使用说明文档
概述
该工具是一个用 Python 编写的批量下载文件脚本。它从配置文件中读取下载任务列表,并并行下载文件,同时显示下载进度。该工具使用 requests
库进行 HTTP 请求,使用 tqdm
库显示下载进度,并使用 configparser
库读取配置文件。
依赖
在使用该工具之前,请确保已安装以下库:
requests
tqdm
可以使用以下命令安装所需的库:
pip install requests tqdm
配置文件
创建一个配置文件(例如 downloads.ini
),其中包含下载任务列表。配置文件使用 INI 格式,每个下载任务有一个单独的节,每个节包含 url
和 path
两个字段。
示例配置文件 downloads.ini
:
[file1]
url = http://example.com/file1.jpg
path = downloads/file1.jpg
[file2]
url = http://example.com/file2.jpg
path = downloads/file2.jpg
[file3]
url = http://example.com/file3.jpg
path = downloads/file3.jpg
运行脚本
将以下代码保存为 batch_download.py
:
import os
import requests
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
import configparser
import argparse
def download_file(task):
"""
下载单个文件并保存到指定路径,并显示下载进度
"""
url = task['url']
path = task['path']
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
with requests.get(url, stream=True) as r:
r.raise_for_status()
total_size = int(r.headers.get('content-length', 0))
with open(path, 'wb') as f, tqdm(
total=total_size, unit='B', unit_scale=True, desc=os.path.basename(path)
) as pbar:
for chunk in r.iter_content(chunk_size=8192):
f.write(chunk)
pbar.update(len(chunk))
print(f'{path} 下载完成')
except Exception as e:
print(f'下载 {url} 失败: {e}')
def read_config(config_path):
"""
从配置文件读取下载任务列表
"""
config = configparser.ConfigParser()
config.read(config_path)
tasks = []
for section in config.sections():
url = config[section]['url']
path = config[section]['path']
tasks.append({'url': url, 'path': path})
return tasks
def main(config_path):
"""
主函数,读取配置文件并并行下载文件
"""
tasks = read_config(config_path)
with ThreadPoolExecutor(max_workers=5) as executor:
executor.map(download_file, tasks)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='批量下载文件工具')
parser.add_argument('config', type=str, help='配置文件路径')
args = parser.parse_args()
main(args.config)
使用方法
-
创建配置文件:按照上述示例创建
downloads.ini
文件,并根据需要修改下载任务。 -
运行脚本:在命令行中运行以下命令,指定配置文件路径:
python batch_download.py downloads.ini
代码功能说明
-
导入库
import os import requests from concurrent.futures import ThreadPoolExecutor from tqdm import tqdm import configparser import argparse
-
download_file
函数:下载单个文件并保存到指定路径,同时显示下载进度。def download_file(task): url = task['url'] path = task['path'] try: os.makedirs(os.path.dirname(path), exist_ok=True) with requests.get(url, stream=True) as r: r.raise_for_status() total_size = int(r.headers.get('content-length', 0)) with open(path, 'wb') as f, tqdm( total=total_size, unit='B', unit_scale=True, desc=os.path.basename(path) ) as pbar: for chunk in r.iter_content(chunk_size=8192): f.write(chunk) pbar.update(len(chunk)) print(f'{path} 下载完成') except Exception as e: print(f'下载 {url} 失败: {e}')
-
read_config
函数:从配置文件读取下载任务列表。def read_config(config_path): config = configparser.ConfigParser() config.read(config_path) tasks = [] for section in config.sections(): url = config[section]['url'] path = config[section]['path'] tasks.append({'url': url, 'path': path}) return tasks
-
main
函数:读取配置文件并并行下载文件。def main(config_path): tasks = read_config(config_path) with ThreadPoolExecutor(max_workers=5) as executor: executor.map(download_file, tasks)
-
命令行参数解析:使用
argparse
解析命令行参数,指定配置文件路径。if __name__ == '__main__': parser = argparse.ArgumentParser(description='批量下载文件工具') parser.add_argument('config', type=str, help='配置文件路径') args = parser.parse_args() main(args.config)
注意事项
- 请确保配置文件中的 URL 是有效的下载链接。
- 如果下载过程中出现错误,脚本会输出相应的错误信息。
原文地址:https://blog.csdn.net/Mr_JingFu/article/details/140311345
免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!