位置: 文档库 > Python > 使用python下载文件的三种方法介绍

使用python下载文件的三种方法介绍

DeltaPulse80 上传于 2020-12-20 11:17

《使用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模块。涵盖了基本下载、流式下载大文件、带进度条的下载、断点续传多线程下载等高级功能,并提供了完整的代码示例和最佳实践建议。

《使用python下载文件的三种方法介绍.doc》
将本文的Word文档下载到电脑,方便收藏和打印
推荐度:
点击下载文档