海南响应式网站建设方案游戏广告推广平台

一、背景

---

需要对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" ]; thenecho "Source directory not found: $SOURCE_DIR"exit 1
fi# 创建 7 个子目录
for i in {1..7}; domkdir -p "${SOURCE_DIR}/split_${i}"if [ ! -d "${SOURCE_DIR}/split_${i}" ]; thenecho "Failed to create subdirectory: ${SOURCE_DIR}/split_${i}"exit 1fi
done# 获取所有文件列表（排除子目录）
files=()
while IFS= read -r -d '' file; dofiles+=("$file")
done < <(find "$SOURCE_DIR" -maxdepth 1 -type f -print0)# 检查是否有文件
if [ ${#files[@]} -eq 0 ]; thenecho "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 ]; thencount=0split_index=$((split_index + 1))fi
doneecho "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 osimport 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 0def 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)

https://gitcode.com/qq_43920838/module_code.git