当前位置：首页 > news >正文

【工程实践/源码阅读】批量文件处理步骤以及如何并行处理

news 来源：原创 2025/5/28 23:27:55

一、背景

需要对3000份文件进行处理，内容、文件名不是很标准，而且文件类型多种多样，需要统一转成PDF，再从PDF解析为markdown。再各个处理的阶段，也会带来一些脏数据，比如doc转PDF的时候，就可能有部分doc转换失败。下面会罗列一些在文件处理阶段好用的函数。
在数据清洗过程，可能面临的问题有：

文件类型不一样
处理过程中带来的脏数据
文件名（含有非法字符导致路径问题）
批量处理如何分块？
如何并行处理

二、常用函数

1.各类型统计

完成目录下各类型统计，并建立分类后的目录
在这里插入图片描述
使用字典这个数据结构来存储类型以及类型下的文件路径
shutil.move(file, dest_path) 是 Python 中用于移动文件或目录的函数。它的作用是将 file 移动到 dest_path。如果 dest_path 是一个目录，则会将 file 移动到该目录中，并保留原文件名。

def classify_and_move_files_by_type(folder_path, output_folder):
    # 创建一个字典来存储文件类型及其对应的文件路径
    file_dict = defaultdict(list)

    # 遍历文件夹及其子文件夹
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            # 获取文件路径
            file_path = os.path.join(root, file)
            # 获取文件扩展名（类型）
            file_type = os.path.splitext(file)[1].lower()  # 转换为小写以便统一处理
            # 将文件路径添加到对应类型的列表中
            file_dict[file_type].append(file_path)

    # 创建输出文件夹（如果不存在）
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    # 将文件按类型分类并移动到新的文件夹中
    for file_type, files in file_dict.items():
        # 创建子文件夹（以文件类型命名）
        type_folder = os.path.join(output_folder, file_type if file_type else "no_extension")
        if not os.path.exists(type_folder):
            os.makedirs(type_folder)

        # 移动文件到对应的子文件夹
        for file in files:
            # 获取文件名
            file_name = os.path.basename(file)
            # 目标路径
            dest_path = os.path.join(type_folder, file_name)
            # 移动文件
            shutil.move(file, dest_path)
            print(f"Moved: {file} -> {dest_path}")

    print("文件分类和移动完成！")

2. 批量处理分块

此处是用的一个脚本，输入目录可以将目录下进行分块，便于并行处理

#!/bin/bash

# 定义源目录
SOURCE_DIR="/data/huyuqiang/简历分类/pdf"

# 检查源目录是否存在
if [ ! -d "$SOURCE_DIR" ]; then
    echo "Source directory not found: $SOURCE_DIR"
    exit 1
fi

# 创建 7 个子目录
for i in {1..7}; do
    mkdir -p "${SOURCE_DIR}/split_${i}"
    if [ ! -d "${SOURCE_DIR}/split_${i}" ]; then
        echo "Failed to create subdirectory: ${SOURCE_DIR}/split_${i}"
        exit 1
    fi
done

# 获取所有文件列表（排除子目录）
files=()
while IFS= read -r -d '' file; do
    files+=("$file")
done < <(find "$SOURCE_DIR" -maxdepth 1 -type f -print0)

# 检查是否有文件
if [ ${#files[@]} -eq 0 ]; then
    echo "No files found in $SOURCE_DIR"
    exit 1
fi

# 计算每份文件的数量
total_files=${#files[@]}
files_per_split=$(( (total_files + 6) / 7 ))  # 向上取整

# 将文件分配到子目录
count=0
split_index=1
for file in "${files[@]}"; do
    # 移动文件到子目录
    mv "$file" "${SOURCE_DIR}/split_${split_index}/"

    # 更新计数器和子目录索引
    count=$((count + 1))
    if [ $count -ge $files_per_split ]; then
        count=0
        split_index=$((split_index + 1))
    fi
done

echo "Files have been split into 7 subdirectories."

3.检查最大文件和最小文件

在数据处理过程中，往往会形成一些超大文件（这些文件中可能是乱码），这些文件如果不处理会影响后续的处理效率。所以需要再过程中查看一下输出。
用一个元组列表来保存文件路径对应的文件页数
PyPDF2.PdfReader(f) 是 PyPDF2 库中用于读取 PDF 文件的类。它可以从文件对象 f 中加载 PDF 内容，并提供一个接口来访问 PDF 的页面、元数据、书签等信息。
然后利用sorted对列表进行排序

def sort_pdf_files_by_pages(root_dir, output_file):
    """
    收集指定目录下的所有PDF文件路径和页数，并按页数排序后写入文件
    
    参数:
    root_dir (str): 要搜索的根目录路径
    output_file (str): 输出文件路径
    """
    import os
    import PyPDF2
    
    # 收集所有PDF文件路径和页数
    pdf_files = []
    for dirpath, _, filenames in os.walk(root_dir):
        for filename in filenames:
            if filename.endswith(".pdf"):
                file_path = os.path.join(dirpath, filename)
                try:
                    # 打开PDF文件并读取页数
                    with open(file_path, "rb") as f:
                        reader = PyPDF2.PdfReader(f)
                        num_pages = len(reader.pages)
                        pdf_files.append((file_path, num_pages))
                except Exception as e:
                    print(f"无法读取文件 {file_path}: {e}")

    # 按页数排序（从大到小）
    pdf_files_sorted = sorted(pdf_files, key=lambda x: x[1], reverse=True)

    # 将结果写入txt文件
    with open(output_file, "w", encoding="utf-8") as f:
        for i, (file_path, num_pages) in enumerate(pdf_files_sorted, 1):
            f.write(f"{i}. {file_path} - {num_pages} 页\n")

排序完查看一下大文件和小文件，是不是最大文件是处理后乱码变成脏数据，还有最小文件是不是空白页没有进行处理

3.删除大文件

这里是估计了一下超过9页的文件可能是一个脏数据，然后就删除
直接用os.remove(pdf_path)

def count_pdf_pages(pdf_path):
    """计算 PDF 文件的页数"""
    try:
        with open(pdf_path, 'rb') as file:
            reader = PyPDF2.PdfReader(file)
            return len(reader.pages)
    except Exception as e:
        print(f"Error reading {pdf_path}: {e}")
        return 0

def delete_large_pdfs(directory, max_pages=9):
    """删除页数大于 max_pages 的 PDF 文件"""
    for filename in os.listdir(directory):
        if filename.endswith(".pdf"):
            pdf_path = os.path.join(directory, filename)
            num_pages = count_pdf_pages(pdf_path)
            if num_pages > max_pages:
                print(f"Deleting {filename} (pages: {num_pages})")
                os.remove(pdf_path)