Python下载URL压缩文件并正确提取内容的教程

import requests
import zipfile
import tempfile
import os
import pandas as pd # 假设最终目标是CSV，解压后可能需要用到

# 示例URL，指向一个ZIP压缩包
# 注意：实际应用中请替换为你的目标URL
TARGET_URL = "https://prod-dcd-datasets-cache-zipfiles.s3.eu-west-1.amazonaws.com/mpjzbtfgfr-1.zip"
CHUNK_SIZE = 32 * 1024  # 定义每次读取的块大小，提高效率

def download_and_extract_zip(url: str, extract_path: str = ".") -> list:
    """
    从指定的URL下载ZIP文件，并将其内容解压到指定路径。

    Args:
        url (str): ZIP文件的URL。
        extract_path (str): 解压文件的目标目录。默认为当前工作目录。

    Returns:
        list: 成功解压的文件名列表。
    """
    extracted_files = []
    try:
        print(f"开始从 {url} 下载文件...")
        # 使用requests以流模式获取文件内容
        with requests.get(url, stream=True) as response:
            response.raise_for_status()  # 检查HTTP请求是否成功 (2xx状态码)

            # 使用tempfile创建临时文件来存储下载的ZIP内容
            with tempfile.TemporaryFile() as temp_zip_file:
                # 迭代响应内容块并写入临时文件
                for chunk in response.iter_content(chunk_size=CHUNK_SIZE):
                    if chunk:  # 过滤掉保持连接的空块
                        temp_zip_file.write(chunk)
                temp_zip_file.flush() # 确保所有数据都写入临时文件

                # 将文件指针移到开头，以便zipfile模块读取
                temp_zip_file.seek(0)

                # 使用zipfile模块打开并解压临时文件
                with zipfile.ZipFile(temp_zip_file) as zip_ref:
                    print(f"成功下载ZIP文件。正在解压到 {os.path.abspath(extract_path)}...")
                    # 获取ZIP文件中包含的所有文件名
                    file_names_in_zip = zip_ref.namelist()
                    print("ZIP文件中包含的文件:", file_names_in_zip)

                    # 解压所有文件到指定路径
                    zip_ref.extractall(path=extract_path)
                    print("文件解压完成。")
                    extracted_files = [os.path.join(extract_path, f) for f in file_names_in_zip]

                    # 示例：如果知道解压后是CSV文件，可以进一步处理
                    if any(name.endswith(".csv") for name in file_names_in_zip):
                        for file_name in file_names_in_zip:
                            if file_name.endswith(".csv"):
                                full_path = os.path.join(extract_path, file_name)
                                try:
                                    df = pd.read_csv(full_path)
                                    print(f"\n成功读取CSV文件: {full_path}")
                                    print(f"CSV文件前5行:\n{df.head()}")
                                except Exception as e:
                                    print(f"警告: 无法读取CSV文件 {full_path}: {e}")

    except requests.exceptions.RequestException as e:
        print(f"下载请求错误: {e}")
    except zipfile.BadZipFile:
        print("错误: 下载的文件不是一个有效的ZIP文件。")
    except Exception as e:
        print(f"发生未知错误: {e}")

    return extracted_files

# 执行下载和解压操作
if __name__ == "__main__":
    # 指定解压目录，可以根据需要修改
    output_directory = "./downloaded_data"
    os.makedirs(output_directory, exist_ok=True) # 确保目录存在

    downloaded_files = download_and_extract_zip(TARGET_URL, output_directory)
    if downloaded_files:
        print("\n所有下载并解压的文件:")
        for f in downloaded_files:
            print(f)