Python下载URL文件：解析与处理压缩包内容

import requests
import zipfile
import tempfile
import os # 用于获取当前工作目录

# 待下载的ZIP文件URL
ZIP_URL = "https://prod-dcd-datasets-cache-zipfiles.s3.eu-west-1.amazonaws.com/mpjzbtfgfr-1.zip"
# 定义下载时的分块大小，有助于处理大文件
CHUNK_SIZE = 32 * 1024 # 32 KB

def download_and_extract_zip(url: str, chunk_size: int = CHUNK_SIZE, extract_path: str = None):
    """
    从指定URL下载ZIP文件，并将其内容解压到指定路径。

    Args:
        url (str): ZIP文件的URL。
        chunk_size (int): 下载时每次读取的数据块大小。
        extract_path (str, optional): 文件解压的目标路径。如果为None，则解压到当前工作目录。
    """
    if extract_path is None:
        extract_path = os.getcwd()

    print(f"开始从 {url} 下载ZIP文件...")
    try:
        # 使用requests.get进行流式下载，stream=True表示不立即下载全部内容
        with requests.get(url, stream=True) as response:
            response.raise_for_status()  # 检查HTTP请求是否成功 (状态码2xx)

            # 使用tempfile.TemporaryFile创建一个临时文件对象，用于存储下载的ZIP内容
            # 这避免了将整个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)

                # 下载完成后，将文件指针重置到文件开头，以便zipfile可以读取
                temp_zip_file.flush()
                temp_zip_file.seek(0)

                # 使用zipfile库打开临时文件，并解压其内容
                with zipfile.ZipFile(temp_zip_file) as zip_ref:
                    print(f"ZIP文件中包含的文件: {zip_ref.namelist()}")
                    print(f"正在解压文件到: {extract_path}")
                    zip_ref.extractall(path=extract_path)
                    print("文件解压完成。")

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

# 调用函数执行下载和解压
if __name__ == "__main__":
    download_and_extract_zip(ZIP_URL)

    # 假设已知解压后的CSV文件名为 'CRC_clusters_neighborhoods_markers.csv'
    # 可以在解压后进一步处理该文件
    extracted_csv_name = "CRC_clusters_neighborhoods_markers.csv"
    extracted_csv_path = os.path.join(os.getcwd(), extracted_csv_name)
    if os.path.exists(extracted_csv_path):
        print(f"\n成功找到并处理文件: {extracted_csv_path}")
        # 这里可以添加使用pandas读取CSV文件的逻辑
        # import pandas as pd
        # df = pd.read_csv(extracted_csv_path)
        # print(df.head())
    else:
        print(f"\n未找到预期的解压文件: {extracted_csv_path}")