《使用Python下载文件的三种方法介绍》
在Python编程中,文件下载是常见的网络操作需求。无论是从HTTP服务器获取数据、下载大型文件,还是实现自动化任务,掌握高效的下载方法至关重要。本文将详细介绍三种主流的Python文件下载技术:使用标准库`urllib`、第三方库`requests`以及流式下载大文件的方法,并对比它们的适用场景和性能特点。
一、使用urllib.request下载文件
`urllib`是Python标准库的一部分,无需额外安装即可使用。它提供了基础的HTTP请求功能,适合简单的下载任务。
1.1 基本下载方法
通过`urllib.request.urlretrieve()`函数可以直接将远程文件保存到本地:
import urllib.request
url = "https://example.com/sample.txt"
local_path = "downloaded_file.txt"
try:
urllib.request.urlretrieve(url, local_path)
print(f"文件已下载到 {local_path}")
except Exception as e:
print(f"下载失败: {e}")
这种方法简单直接,但缺乏进度显示和错误处理的精细控制。
1.2 添加进度条
要实现下载进度显示,需要自定义`urlretrieve`的回调函数:
import urllib.request
import sys
def download_progress(block_num, block_size, total_size):
downloaded = block_num * block_size
if total_size > 0:
percent = downloaded / total_size * 100
sys.stdout.write(f"\r下载进度: {percent:.1f}%")
sys.stdout.flush()
url = "https://example.com/large_file.zip"
local_path = "large_file.zip"
try:
urllib.request.urlretrieve(
url,
local_path,
reporthook=download_progress
)
print("\n下载完成")
except Exception as e:
print(f"下载失败: {e}")
这种方法通过`reporthook`参数实现进度反馈,但需要手动计算百分比。
1.3 完整示例:带错误处理的下载
import urllib.request
import urllib.error
import os
def download_file(url, save_path):
try:
# 检查目录是否存在
os.makedirs(os.path.dirname(save_path), exist_ok=True)
# 添加请求头模拟浏览器
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
}
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req) as response:
if response.status == 200:
with open(save_path, "wb") as f:
f.write(response.read())
print(f"文件成功保存到 {save_path}")
else:
print(f"服务器返回错误: {response.status}")
except urllib.error.HTTPError as e:
print(f"HTTP错误: {e.code} - {e.reason}")
except urllib.error.URLError as e:
print(f"URL错误: {e.reason}")
except Exception as e:
print(f"未知错误: {e}")
# 使用示例
download_file(
"https://example.com/protected_file.pdf",
"./downloads/protected_file.pdf"
)
这个示例展示了如何添加请求头、处理HTTP错误以及创建必要的目录结构。
二、使用requests库下载文件
`requests`库是Python中最流行的HTTP客户端库,以其简洁的API和强大的功能著称。需要先安装:
pip install requests
2.1 基本下载方法
import requests
url = "https://example.com/sample.jpg"
local_path = "sample.jpg"
try:
response = requests.get(url)
response.raise_for_status() # 检查请求是否成功
with open(local_path, "wb") as f:
f.write(response.content)
print(f"文件已保存到 {local_path}")
except requests.exceptions.RequestException as e:
print(f"下载失败: {e}")
这种方法比`urllib`更简洁,自动处理了重定向和HTTP错误。
2.2 流式下载大文件
对于大文件,可以使用流式下载避免内存问题:
import requests
import os
def download_large_file(url, save_path, chunk_size=8192):
try:
os.makedirs(os.path.dirname(save_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))
downloaded = 0
with open(save_path, "wb") as f:
for chunk in r.iter_content(chunk_size=chunk_size):
if chunk: # 过滤掉keep-alive新块
f.write(chunk)
downloaded += len(chunk)
# 计算并显示进度
if total_size > 0:
percent = downloaded / total_size * 100
print(f"\r下载进度: {percent:.1f}%", end="")
print("\n下载完成")
except requests.exceptions.RequestException as e:
print(f"下载失败: {e}")
流式下载特别适合下载视频、ISO镜像等大型文件,因为它不会一次性将整个文件加载到内存中。
2.3 完整示例:带进度条和断点续传
import requests
import os
from tqdm import tqdm # 需要安装: pip install tqdm
def download_with_progress(url, save_path):
try:
# 检查是否已部分下载
downloaded = 0
if os.path.exists(save_path):
downloaded = os.path.getsize(save_path)
headers = {"Range": f"bytes={downloaded}-"} if downloaded > 0 else {}
with requests.get(url, headers=headers, stream=True) as r:
r.raise_for_status()
total_size = int(r.headers.get("content-length", 0))
if downloaded > 0:
total_size += downloaded
mode = "ab" if downloaded > 0 else "wb"
with open(save_path, mode) as f, tqdm(
desc="下载进度",
total=total_size,
unit="iB",
unit_scale=True,
initial=downloaded,
unit_divisor=1024,
) as bar:
for chunk in r.iter_content(chunk_size=1024):
f.write(chunk)
bar.update(len(chunk))
print("\n下载完成")
except requests.exceptions.RequestException as e:
print(f"下载失败: {e}")
# 使用示例
download_with_progress(
"https://example.com/huge_video.mp4",
"./videos/huge_video.mp4"
)
这个示例使用了`tqdm`库来显示美观的进度条,并实现了断点续传功能,即如果下载中断,可以从上次的位置继续下载。
三、使用wget模块下载文件
`wget`是一个Python实现的命令行工具,模仿了Unix的wget命令。需要先安装:
pip install wget
3.1 基本下载方法
import wget
url = "https://example.com/sample.pdf"
local_path = "sample.pdf"
try:
filename = wget.download(url, out=local_path)
print(f"\n文件已保存到 {filename}")
except Exception as e:
print(f"下载失败: {e}")
`wget`模块会自动显示下载进度条,适合快速实现下载功能。
3.2 高级功能:批量下载
`wget`可以方便地实现批量下载:
import wget
import os
def batch_download(url_list, save_dir):
try:
os.makedirs(save_dir, exist_ok=True)
for i, url in enumerate(url_list, 1):
filename = os.path.join(save_dir, f"file_{i}.dat")
print(f"正在下载第 {i} 个文件...")
try:
wget.download(url, out=filename)
print(f"文件 {i} 下载完成")
except Exception as e:
print(f"文件 {i} 下载失败: {e}")
except Exception as e:
print(f"批量下载失败: {e}")
# 示例URL列表
urls = [
"https://example.com/data1.dat",
"https://example.com/data2.dat",
"https://example.com/data3.dat"
]
batch_download(urls, "./batch_downloads")
3.3 完整示例:带重试机制的下载
import wget
import time
from requests.exceptions import ConnectionError
def download_with_retry(url, save_path, max_retries=3, delay=5):
retries = 0
while retries
这个示例实现了自动重试机制,当网络不稳定时可以自动重试下载,适合在不可靠的网络环境下使用。
四、三种方法对比与选择建议
方法 | 优点 | 缺点 | 适用场景 |
---|---|---|---|
urllib | 标准库,无需安装 | API较复杂,功能较少 | 简单下载,不想安装第三方库时 |
requests | API简洁,功能强大 | 需要额外安装 | 大多数下载场景,特别是需要进度显示或大文件下载时 |
wget | 自动进度显示,简单易用 | 功能相对有限 | 快速实现下载,批量下载 |
五、最佳实践建议
1. 对于小型项目或只需要简单下载功能的场景,`urllib`是不错的选择,因为它不需要额外安装依赖。
2. 对于大多数实际应用,特别是需要显示下载进度、处理大文件或实现断点续传的场景,`requests`库提供了最全面和灵活的解决方案。
3. 如果需要快速实现下载功能,特别是批量下载,`wget`模块提供了最简洁的API。
4. 无论使用哪种方法,都应该添加适当的错误处理,包括网络错误、文件写入错误等。
5. 对于大文件下载,务必使用流式下载方法,避免内存不足的问题。
6. 考虑添加重试机制,特别是在网络不稳定的环境下。
7. 对于需要认证的下载,确保正确处理cookies和会话。
六、扩展应用:下载管理器实现
结合上述技术,我们可以实现一个简单的下载管理器:
import requests
import os
import threading
import queue
from tqdm import tqdm
class DownloadManager:
def __init__(self, max_workers=4):
self.task_queue = queue.Queue()
self.max_workers = max_workers
self.workers = []
def add_task(self, url, save_path):
self.task_queue.put((url, save_path))
def worker(self):
while True:
url, save_path = self.task_queue.get()
try:
self._download_file(url, save_path)
except Exception as e:
print(f"下载 {url} 失败: {e}")
finally:
self.task_queue.task_done()
def _download_file(self, url, save_path):
os.makedirs(os.path.dirname(save_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(save_path, "wb") as f, tqdm(
desc=os.path.basename(save_path),
total=total_size,
unit="iB",
unit_scale=True,
unit_divisor=1024,
) as bar:
for chunk in r.iter_content(chunk_size=1024):
f.write(chunk)
bar.update(len(chunk))
print(f"\n{url} 下载完成")
def start(self):
for _ in range(self.max_workers):
t = threading.Thread(target=self.worker)
t.daemon = True
t.start()
self.workers.append(t)
def wait_completion(self):
self.task_queue.join()
def shutdown(self):
for _ in range(self.max_workers):
self.task_queue.put((None, None)) # 发送终止信号
for t in self.workers:
t.join()
# 使用示例
if __name__ == "__main__":
dm = DownloadManager(max_workers=2)
dm.start()
tasks = [
("https://example.com/file1.zip", "./downloads/file1.zip"),
("https://example.com/file2.zip", "./downloads/file2.zip"),
("https://example.com/file3.zip", "./downloads/file3.zip"),
]
for url, path in tasks:
dm.add_task(url, path)
dm.wait_completion()
dm.shutdown()
这个下载管理器实现了多线程下载、进度显示和任务队列管理,可以根据需要扩展更多功能,如暂停/继续、速度限制等。
七、常见问题解决
7.1 下载中文文件名文件
当下载的文件名包含中文时,可能会遇到编码问题。解决方法:
import requests
import urllib.parse
url = "https://example.com/中文文件.pdf"
filename = "中文文件.pdf"
# 对URL进行编码(如果URL中包含中文)
encoded_url = urllib.parse.quote(url, safe="/:?=")
# 或者在保存时处理文件名编码
with open(filename.encode("utf-8").decode("latin1"), "wb") as f: # 不推荐,仅演示
pass # 更好的方式是确保系统支持UTF-8文件名
最佳实践是确保你的Python脚本和操作系统都支持UTF-8编码。
7.2 处理重定向
`requests`库会自动处理重定向,但有时你可能需要知道最终URL:
import requests
url = "https://example.com/shortlink"
response = requests.get(url, allow_redirects=True)
print(f"最终URL: {response.url}")
7.3 限制下载速度
对于大文件下载,可能需要限制速度以避免占用过多带宽:
import requests
import time
class ThrottledStream:
def __init__(self, response, bytes_per_second):
self.response = response
self.bytes_per_second = bytes_per_second
self.last_read = time.time()
self.bytes_read = 0
def read(self, size=-1):
now = time.time()
elapsed = now - self.last_read
# 计算可以读取的字节数
max_bytes = int(elapsed * self.bytes_per_second) - self.bytes_read
if max_bytes 0 and len(chunk) 0.1:
additional_chunk = self.response.raw.read(size - len(chunk))
chunk += additional_chunk
self.bytes_read += len(additional_chunk)
return chunk
url = "https://example.com/large_file.zip"
save_path = "large_file.zip"
bytes_per_second = 1024 * 100 # 限制为100KB/s
response = requests.get(url, stream=True)
throttled_stream = ThrottledStream(response, bytes_per_second)
with open(save_path, "wb") as f:
while True:
chunk = throttled_stream.read(8192)
if not chunk:
break
f.write(chunk)
更简单的方法是使用第三方库如`requests-throttle`。
八、总结
本文详细介绍了Python中三种主要的文件下载方法:使用标准库`urllib`、功能强大的`requests`库以及简洁的`wget`模块。每种方法都有其适用的场景和优缺点。
对于简单的下载任务,`urllib`提供了无需安装的解决方案;对于大多数实际应用,`requests`库因其简洁的API和强大的功能成为首选;而`wget`模块则适合需要快速实现下载功能的场景。
在实际开发中,应根据项目需求选择合适的方法,并考虑添加错误处理、进度显示、断点续传等功能以提高用户体验。对于更复杂的需求,可以基于这些基本方法构建自己的下载管理器或扩展功能。
掌握这些文件下载技术不仅可以帮助你完成日常的开发任务,还能为构建更复杂的网络应用程序打下坚实的基础。
关键词:Python下载文件、urllib下载、requests下载、wget下载、流式下载、断点续传、多线程下载、下载进度显示
简介:本文详细介绍了Python中三种主要的文件下载方法:使用标准库urllib、第三方库requests和wget模块。涵盖了基本下载、流式下载大文件、带进度条的下载、断点续传、多线程下载等高级功能,并提供了完整的代码示例和最佳实践建议。