Python实现文件下载的方法汇总与适用场景介绍

import urllib.request

url = "https://example.com/file.zip"
filename = "downloaded_file.zip"

urllib.request.urlretrieve(url, filename)
print(f"文件已保存为: {filename}")

# 进阶：添加请求头
headers = {"User-Agent": "Mozilla/5.0"}
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req) as response:
    with open(filename, 'wb') as f:
        f.write(response.read())

import requests

url = "https://example.com/large_file.iso"
filename = "large_file.iso"

# 简单下载
response = requests.get(url)
with open(filename, 'wb') as f:
    f.write(response.content)

# 流式下载大文件
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    with open(filename, 'wb') as f:
        for chunk in r.iter_content(chunk_size=8192): 
            f.write(chunk)

import wget

url = "https://example.com/image.jpg"
filename = wget.download(url)
print(f"\n下载完成: {filename}")

# 指定保存路径
wget.download(url, out="/path/to/save/image.jpg")

import http.client

conn = http.client.HTTPSConnection("example.com")
conn.request("GET", "/file.pdf")
response = conn.getresponse()

with open("document.pdf", 'wb') as f:
    f.write(response.read())

conn.close()

import aiohttp
import asyncio

async def download_file(url, filename):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            with open(filename, 'wb') as f:
                while True:
                    chunk = await response.content.read(8192)
                    if not chunk:
                        break
                    f.write(chunk)
    print(f"异步下载完成: {filename}")

urls = [
    ("https://example.com/file1.zip", "file1.zip"),
    ("https://example.com/file2.zip", "file2.zip")
]

async def main():
    tasks = [download_file(url, name) for url, name in urls]
    await asyncio.gather(*tasks)

asyncio.run(main())

import pycurl
from io import BytesIO

buffer = BytesIO()
c = pycurl.Curl()
c.setopt(c.URL, "https://example.com/data.json")
c.setopt(c.WRITEDATA, buffer)
c.perform()
c.close()

body = buffer.getvalue()
with open("data.json", 'wb') as f:
    f.write(body)

import urllib3

http = urllib3.PoolManager()
url = "https://example.com/video.mp4"
response = http.request("GET", url, preload_content=False)

with open("video.mp4", 'wb') as f:
    for chunk in response.stream(1024):
        f.write(chunk)

response.release_conn()

import socket

def download_via_socket(url, port=80, filename="output.bin"):
    # 解析URL（简化版，实际应使用urllib.parse）
    host = url.split('/')[2]
    path = '/' + '/'.join(url.split('/')[3:])
    
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((host, port))
    request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\n\r\n"
    s.send(request.encode())
    
    with open(filename, 'wb') as f:
        while True:
            data = s.recv(1024)
            if not data:
                break
            f.write(data)
    s.close()

​​​​​​​download_via_socket("http://example.com/file")

import requests
from multiprocessing import Pool

def download(args):
    url, filename = args
    response = requests.get(url, stream=True)
    with open(filename, 'wb') as f:
        for chunk in response.iter_content(8192):
            f.write(chunk)
    return filename

urls = [
    ("https://example.com/file1.zip", "file1.zip"),
    ("https://example.com/file2.zip", "file2.zip")
]

with Pool(4) as p:  # 4个进程
    results = p.map(download, urls)
    print(f"下载完成: {results}")

import scrapy
from scrapy.crawler import CrawlerProcess

class FileDownloadSpider(scrapy.Spider):
    name = "filedownload"
    start_urls = ["https://example.com/downloads"]
    
    def parse(self, response):
        for href in response.css('a.download-link::attr(href)').getall():
            yield scrapy.Request(
                response.urljoin(href),
                callback=self.save_file
            )
    
    def save_file(self, response):
        path = response.url.split('/')[-1]
        with open(path, 'wb') as f:
            f.write(response.body)
        self.log(f"保存文件: {path}")

process = CrawlerProcess()
process.crawl(FileDownloadSpider)
process.start()

import requests
import os

def download_with_resume(url, filename):
    headers = {}
    if os.path.exists(filename):
        downloaded = os.path.getsize(filename)
        headers = {'Range': f'bytes={downloaded}-'}
    
    with requests.get(url, headers=headers, stream=True) as r:
        mode = 'ab' if headers else 'wb'
        with open(filename, mode) as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)

download_with_resume("https://example.com/large_file.iso", "large_file.iso")

# requests示例（默认验证证书）
requests.get("https://example.com", verify=True)

max_size = 1024 * 1024 * 100  # 100MB
response = requests.get(url, stream=True)
downloaded = 0
with open(filename, 'wb') as f:
    for chunk in response.iter_content(8192):
        downloaded += len(chunk)
        if downloaded > max_size:
            raise ValueError("文件超过最大限制")
        f.write(chunk)

import re
def sanitize_filename(filename):
    return re.sub(r'[\\/*?:"<>|]', "", filename)