基于os.walk批量创建目录并高效提取Zip文件的优化方案问询

问题背景 #

根目录下包含N个项目文件夹，每个项目文件夹内有一个Zip文件，该Zip包含dpd.zip与lib.ear两个文件。需要实现以下功能：

• 遍历所有项目文件夹
• 在每个项目内创建core/zip、core/ear目录
• 提取项目内的Zip文件到core目录
• 将dpd.zip移至zip子目录并解压，lib.ear移至ear子目录并解压
• 任务完成后脚本自动删除

原代码在处理dpd.zip和lib.ear时需遍历整个core目录，项目数量庞大时效率极低，以下是优化方案：

优化后的代码 #

import os
import shutil
import zipfile

def extract_archive(file_path, extract_to):
    """通用解压函数，支持.zip和.ear（ear本质是zip格式）"""
    with zipfile.ZipFile(file_path, 'r') as zip_ref:
        zip_ref.extractall(extract_to)

def process_project(project_dir):
    # 构造目标目录路径
    core_dir = os.path.join(project_dir, "core")
    zip_dir = os.path.join(core_dir, "zip")
    ear_dir = os.path.join(core_dir, "ear")

    # 一键创建多级目录，已存在则跳过
    os.makedirs(zip_dir, exist_ok=True)
    os.makedirs(ear_dir, exist_ok=True)

    # 快速定位项目内的Zip文件
    project_zip = None
    for item in os.scandir(project_dir):
        if item.is_file() and item.name.endswith(".zip"):
            project_zip = item.path
            break

    if not project_zip:
        print(f"警告：项目{project_dir}未找到Zip文件，跳过")
        return

    # 解压项目Zip到core目录
    extract_archive(project_zip, core_dir)

    # 直接通过已知文件名构造路径，无需遍历目录
    dpd_zip_path = os.path.join(core_dir, "dpd.zip")
    lib_ear_path = os.path.join(core_dir, "lib.ear")

    # 处理dpd.zip
    if os.path.exists(dpd_zip_path):
        shutil.move(dpd_zip_path, zip_dir)
        extract_archive(os.path.join(zip_dir, "dpd.zip"), zip_dir)
    else:
        print(f"警告：项目{project_dir}的core目录未找到dpd.zip")

    # 处理lib.ear（ear是zip格式，直接用zipfile解压）
    if os.path.exists(lib_ear_path):
        shutil.move(lib_ear_path, ear_dir)
        extract_archive(os.path.join(ear_dir, "lib.ear"), ear_dir)
    else:
        print(f"警告：项目{project_dir}的core目录未找到lib.ear")

if __name__ == "__main__":
    # 只遍历当前目录下的直接项目文件夹，排除隐藏目录和脚本自身
    for item in os.scandir("."):
        if item.is_dir() and not item.name.startswith(".") and item.name != os.path.basename(__file__):
            process_project(item.path)

    # 脚本执行完成后自删除
    try:
        os.remove(__file__)
        print("脚本执行完成，已自动删除")
    except Exception as e:
        print(f"脚本执行完成，但自动删除失败：{e}")

核心优化点 #

1. 跳过无意义遍历：直接通过已知文件名构造dpd.zip和lib.ear的路径，省去遍历core目录的开销，大幅提升批量处理速度
2. 高效目录扫描：用os.scandir替代os.listdir，性能更优，尤其适合大量目录场景
3. 简化目录创建：os.makedirs一键创建多级目录，避免重复判断和创建操作
4. 精准范围控制：只处理当前目录下的直接子文件夹，避免os.walk带来的深层遍历冗余
5. 容错性提升：对缺失文件的情况输出警告，避免脚本中途崩溃

内容的提问来源于stack exchange，提问作者Caveira