数据净化大师：从复杂格式到干净文本的工业级解析实战

攻克PDF、扫描件、混合排版，为LLM训练构建零污染的语料生产线

前言

在大模型微调训练中，数据质量直接决定模型性能天花板。一个在1000个样本上训练的模型，如果数据干净，效果可能超过用10000个脏数据训练的模型。特别是对于强调"手写LoRA"的深度技术岗位，解析（Parsing）能力是构建高质量训练数据的基石。

本文将从工业级实践角度，深入讲解：

• PDF的全策略解析（文本型、扫描件、表格混合）
• OCR的精度提升与后处理修复
• 文档结构的语义保留与噪声清除
• 解析质量的自动化评估体系

---

第一部分：解析的根本挑战与应对框架

1.1 为什么解析如此困难？

数据源	挑战	对训练的影响
PDF	非结构化、多种生成方式、表格乱码	产生乱码token，破坏模型学习
扫描件	图像质量差、字体识别错误、排版错乱	引入错误语义，污染微调效果
混合排版	标题/正文/注释/页眉页脚混杂	模型学到噪声模式
编码问题	GBK/UTF-8/PDF专有编码	产生不可见字符，增加困惑度

1.2 工业级解析五步法

┌──────────────────────────────────────────────────────────────┐
│                    工业级解析标准流程                         │
├──────────────────────────────────────────────────────────────┤
│ Step 1: 格式检测  →  识别文档类型（文本型/扫描件/混合）     │
│ Step 2: 策略路由  →  选择最优解析方案                        │
│ Step 3: 内容提取  →  执行解析，保留结构                      │
│ Step 4: 后处理清洗 →  去除噪声，修复错误                     │
│ Step 5: 质量验证  →  自动化评估，过滤低质量数据              │
└──────────────────────────────────────────────────────────────┘

---

第二部分：PDF解析——从简单到复杂的四层递进

2.1 Layer 1：文本型PDF的基础提取

import pdfplumber
import pypdf
from typing import List, Dict, Any, Optional
import re

class PDFTextExtractor:
    """
    PDF文本提取器 - 支持多种策略
    """
    
    @staticmethod
    def extract_with_pdfplumber(pdf_path: str) -> Dict[str, Any]:
        """
        使用pdfplumber提取（推荐）
        优势：保留字符级坐标、字体信息
        """
        result = {
            'text': '',
            'pages': [],
            'metadata': {},
            'tables': [],
            'errors': []
        }
        
        try:
            with pdfplumber.open(pdf_path) as pdf:
                result['metadata'] = pdf.metadata
                
                for page_num, page in enumerate(pdf.pages):
                    try:
                        # 提取文本
                        page_text = page.extract_text()
                        if page_text:
                            result['pages'].append({
                                'page': page_num + 1,
                                'text': page_text
                            })
                            result['text'] += page_text + '\n\n'
                        
                        # 提取表格
                        tables = page.extract_tables()
                        for table in tables:
                            if table and len(table) > 1:
                                result['tables'].append({
                                    'page': page_num + 1,
                                    'data': table
                                })
                                
                    except Exception as e:
                        result['errors'].append(f"Page {page_num+1}: {str(e)}")
                        
        except Exception as e:
            result['errors'].append(f"PDF Open Error: {str(e)}")
        
        return result
    
    @staticmethod
    def extract_with_pypdf(pdf_path: str) -> str:
        """
        使用pypdf提取（速度快，适合简单文档）
        """
        try:
            reader = pypdf.PdfReader(pdf_path)
            text = ''
            for page in reader.pages:
                page_text = page.extract_text()
                if page_text:
                    text += page_text + '\n\n'
            return text
        except Exception as e:
            print(f"pypdf提取失败: {str(e)}")
            return ''
    
    @staticmethod
    def detect_pdf_type(pdf_path: str) -> str:
        """
        智能检测PDF类型
        返回: 'text' | 'scanned' | 'mixed'
        """
        try:
            with pdfplumber.open(pdf_path) as pdf:
                # 检查前3页
                text_count = 0
                for i, page in enumerate(pdf.pages):
                    if i >= 3:
                        break
                    text = page.extract_text()
                    if text and len(text.strip()) > 50:
                        text_count += 1
                
                if text_count >= 2:
                    return 'text'
                elif text_count > 0:
                    return 'mixed'
                else:
                    return 'scanned'
        except:
            return 'scanned'  # 无法打开按扫描件处理

2.2 Layer 2：扫描件OCR提取

import pytesseract
from pdf2image import convert_from_path
from PIL import Image
import numpy as np
import cv2

class OCRProcessor:
    """
    扫描件OCR处理器
    支持预处理增强、多语言识别、结果修正
    """
    
    def __init__(self, language: str = 'chi_sim+eng', dpi: int = 300):
        self.language = language
        self.dpi = dpi
        self.psm = 6  # 默认：假设为统一文本块
    
    def preprocess_image(self, image: Image.Image) -> Image.Image:
        """
        图像预处理 - 提升OCR精度
        """
        # 转换为numpy数组
        img_array = np.array(image.convert('L'))  # 灰度
        
        # 1. 自适应阈值二值化
        binary = cv2.adaptiveThreshold(
            img_array, 255,
            cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
            cv2.THRESH_BINARY, 11, 2
        )
        
        # 2. 降噪
        denoised = cv2.fastNlMeansDenoising(binary, h=10)
        
        # 3. 旋转校正（检测文本方向）
        from deskew import determine_skew
        try:
            angle = determine_skew(denoised)
            if abs(angle) > 0.5:
                h, w = denoised.shape
                center = (w // 2, h // 2)
                rotation_matrix = cv2.getRotationMatrix2D(center, angle, 1.0)
                denoised = cv2.warpAffine(
                    denoised, rotation_matrix, (w, h),
                    flags=cv2.INTER_CUBIC,
                    borderMode=cv2.BORDER_REPLICATE
                )
        except:
            pass
        
        return Image.fromarray(denoised, mode='L')
    
    def extract_text_from_pdf(
        self,
        pdf_path: str,
        preprocess: bool = True,
        pages: Optional[List[int]] = None
    ) -> Dict[str, Any]:
        """
        从扫描件PDF提取文本
        """
        result = {
            'text': '',
            'pages': [],
            'confidence_scores': [],
            'errors': []
        }
        
        try:
            # 转换PDF为图像
            images = convert_from_path(
                pdf_path,
                dpi=self.dpi,
                first_page=pages[0] if pages else None,
                last_page=pages[-1] if pages else None
            )
            
            for i, image in enumerate(images):
                try:
                    if preprocess:
                        processed_image = self.preprocess_image(image)
                    else:
                        processed_image = image
                    
                    # 执行OCR
                    ocr_data = pytesseract.image_to_data(
                        processed_image,
                        lang=self.language,
                        config=f'--psm {self.psm} --oem 3',
                        output_type=pytesseract.Output.DICT
                    )
                    
                    # 提取文本
                    page_text = ''
                    confidence_scores = []
                    
                    for j, conf in enumerate(ocr_data['conf']):
                        if int(conf) > 30:  # 过滤低置信度
                            text = ocr_data['text'][j]
                            if text and text.strip():
                                page_text += text + ' '
                                confidence_scores.append(int(conf))
                    
                    result['pages'].append({
                        'page': i + 1,
                        'text': page_text.strip()
                    })
                    result['text'] += page_text + '\n\n'
                    
                    # 计算平均置信度
                    if confidence_scores:
                        avg_conf = sum(confidence_scores) / len(confidence_scores)
                        result['confidence_scores'].append(avg_conf)
                        if avg_conf < 60:
                            result['errors'].append(
                                f"Page {i+1}: 低置信度 ({avg_conf:.1f}%)"
                            )
                    
                except Exception as e:
                    result['errors'].append(f"Page {i+1} OCR失败: {str(e)}")
                    continue
                    
        except Exception as e:
            result['errors'].append(f"PDF转换失败: {str(e)}")
        
        return result
    
    def post_process_ocr_text(self, text: str) -> str:
        """
        OCR文本后处理
        修正常见识别错误
        """
        # 常见OCR错误修正映射
        corrections = {
            '0': 'O',  # 数字0误识别为字母O
            '1': 'I',  # 数字1误识别为I
            '5': 'S',  # 5误识别为S
            '：': ':',  # 中文冒号转英文
            '。': '.',
            '，': ',',
            '（': '(',
            '）': ')',
            '“': '"',
            '”': '"',
            '‘': "'",
            '’': "'"
        }
        
        for wrong, right in corrections.items():
            text = text.replace(wrong, right)
        
        # 修复多余空格
        text = re.sub(r' +', ' ', text)
        
        # 修复换行
        text = re.sub(r'\n{3,}', '\n\n', text)
        
        return text.strip()

2.3 Layer 3：表格与复杂结构提取

class TableExtractor:
    """
    表格提取与结构化
    支持：有框线表格、无框线表格、合并单元格
    """
    
    @staticmethod
    def extract_tables_with_pdfplumber(pdf_path: str) -> List[Dict[str, Any]]:
        """
        使用pdfplumber提取表格
        """
        tables = []
        
        with pdfplumber.open(pdf_path) as pdf:
            for page_num, page in enumerate(pdf.pages):
                # 提取表格
                page_tables = page.extract_tables({
                    'vertical_strategy': 'lines',
                    'horizontal_strategy': 'lines',
                    'snap_tolerance': 3,
                    'join_tolerance': 3
                })
                
                for table_data in page_tables:
                    if table_data and len(table_data) > 1:
                        # 清理表格数据
                        cleaned_table = TableExtractor._clean_table(table_data)
                        if cleaned_table:
                            tables.append({
                                'page': page_num + 1,
                                'data': cleaned_table,
                                'markdown': TableExtractor._table_to_markdown(cleaned_table)
                            })
        
        return tables
    
    @staticmethod
    def _clean_table(table_data: List[List[str]]) -> List[List[str]]:
        """
        清理表格数据：去除空行、合并单元格处理
        """
        cleaned = []
        for row in table_data:
            # 过滤全空行
            if all(cell is None or str(cell).strip() == '' for cell in row):
                continue
            
            # 清洗每个单元格
            cleaned_row = []
            for cell in row:
                if cell is None:
                    cleaned_row.append('')
                else:
                    # 移除多余换行和空格
                    cell_clean = re.sub(r'\s+', ' ', str(cell)).strip()
                    cleaned_row.append(cell_clean)
            
            cleaned.append(cleaned_row)
        
        return cleaned
    
    @staticmethod
    def _table_to_markdown(table_data: List[List[str]]) -> str:
        """
        将表格转换为Markdown格式
        便于LLM理解表格结构
        """
        if not table_data or len(table_data) < 2:
            return ''
        
        # 提取表头
        headers = table_data[0]
        rows = table_data[1:]
        
        # 构建Markdown表
        markdown = '| ' + ' | '.join(headers) + ' |\n'
        markdown += '| ' + ' | '.join(['---'] * len(headers)) + ' |\n'
        
        for row in rows:
            # 补齐列数
            while len(row) < len(headers):
                row.append('')
            markdown += '| ' + ' | '.join(row) + ' |\n'
        
        return markdown
    
    @staticmethod
    def table_to_natural_language(table_data: List[List[str]]) -> str:
        """
        将表格转换为自然语言描述
        适合作为LLM训练数据
        """
        if not table_data or len(table_data) < 2:
            return ''
        
        headers = table_data[0]
        rows = table_data[1:]
        
        description = "表格内容如下：\n"
        description += f"列名：{', '.join(headers)}\n\n"
        
        for i, row in enumerate(rows, 1):
            # 补齐列数
            while len(row) < len(headers):
                row.append('')
            
            row_desc = f"第{i}行："
            row_desc += '，'.join([f"{headers[j]}为{row[j]}" for j in range(len(headers)) if row[j]])
            description += row_desc + '\n'
        
        return description

2.4 Layer 4：unstructured全自动解析

from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title

class UnstructuredParser:
    """
    使用unstructured库进行全自动解析
    自动识别标题、段落、表格、图片
    """
    
    @staticmethod
    def parse_pdf(
        pdf_path: str,
        strategy: str = 'hi_res'
    ) -> List[Dict[str, Any]]:
        """
        使用unstructured进行PDF解析
        
        Args:
            strategy: 'fast' | 'hi_res' | 'ocr_only'
        """
        elements = partition_pdf(
            filename=pdf_path,
            strategy=strategy,
            extract_images_in_pdf=True,
            infer_table_structure=True,
            chunking_strategy='by_title',
            max_characters=1500,
            new_after_n_chars=1200,
            combine_text_under_n_chars=200,
            languages=['chi_sim', 'eng']
        )
        
        parsed = []
        for elem in elements:
            parsed.append({
                'type': elem.category,  # Title, NarrativeText, Table, etc.
                'text': str(elem),
                'metadata': elem.metadata.to_dict() if hasattr(elem, 'metadata') else {}
            })
        
        return parsed
    
    @staticmethod
    def extract_clean_text_from_unstructured(pdf_path: str) -> str:
        """
        提取纯文本（忽略表格和图片）
        """
        elements = partition_pdf(pdf_path, strategy='fast')
        
        text = ''
        for elem in elements:
            category = elem.category
            # 只提取文本内容
            if category not in ['Table', 'Image', 'Figure']:
                text += str(elem) + '\n\n'
        
        return text.strip()

---

第三部分：解析后清洗——从原始文本到训练语料

3.1 多层级清洗策略

class TextCleaner:
    """
    多层级文本清洗器
    """
    
    def __init__(self):
        self.rules = []
    
    def add_rule(self, name: str, func):
        """添加自定义清洗规则"""
        self.rules.append((name, func))
    
    def clean(self, text: str) -> Dict[str, Any]:
        """
        执行完整清洗流程
        """
        result = {
            'original': text,
            'cleaned': text,
            'transformations': [],
            'statistics': {
                'original_length': len(text),
                'removed_characters': 0,
                'line_count': text.count('\n')
            }
        }
        
        for name, func in self.rules:
            new_text = func(result['cleaned'])
            if new_text != result['cleaned']:
                result['transformations'].append(name)
            result['cleaned'] = new_text
        
        result['statistics']['cleaned_length'] = len(result['cleaned'])
        result['statistics']['removed_characters'] = (
            result['statistics']['original_length'] - result['statistics']['cleaned_length']
        )
        
        return result
    
    @staticmethod
    def clean_header_footer(text: str) -> str:
        """
        移除页眉页脚
        基于位置或模式识别
        """
        lines = text.split('\n')
        if len(lines) <= 10:
            return text
        
        # 检测并移除页眉页脚（基于重复模式）
        # 常见模式：页码、日期、文档标题
        pattern = r'^(Page\s+\d+|\d+/\d+|\d+\s*页|第\s*\d+\s*页|Copyright|©|www\.|http)'
        
        cleaned_lines = []
        for line in lines:
            # 移除匹配页眉页脚的行
            if re.search(pattern, line, re.IGNORECASE):
                continue
            cleaned_lines.append(line)
        
        return '\n'.join(cleaned_lines)
    
    @staticmethod
    def remove_page_breaks(text: str) -> str:
        """移除分页符"""
        text = re.sub(r'[\f\v]', '', text)
        text = re.sub(r'-{5,}', '', text)  # 移除分隔线
        return text
    
    @staticmethod
    def normalize_whitespace(text: str) -> str:
        """统一空白字符"""
        # 合并多个空格
        text = re.sub(r'[ \t]+', ' ', text)
        # 合并多个换行
        text = re.sub(r'\n{3,}', '\n\n', text)
        return text.strip()
    
    @staticmethod
    def fix_encoding_errors(text: str) -> str:
        """修复常见编码错误"""
        replacements = {
            'Â': ' ',
            '’': "'",
            '“': '"',
            'â€': '"',
            'é': 'é',
            'è': 'è',
            'ç': 'ç'
        }
        for wrong, right in replacements.items():
            text = text.replace(wrong, right)
        return text
    
    @staticmethod
    def remove_noise_patterns(text: str) -> str:
        """
        移除常见噪声模式
        """
        noise_patterns = [
            r'http[s]?://[^\s]+',  # URL
            r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',  # 邮箱
            r'\d{4}年\d{1,2}月\d{1,2}日',  # 日期
            r'\d{4}-\d{2}-\d{2}',  # ISO日期
            r'\d+\.\d+\.\d+\.\d+',  # IP地址
        ]
        
        for pattern in noise_patterns:
            text = re.sub(pattern, '', text)
        
        return text

class PipelineCleaner:
    """
    组合清洗器 - 预设工业级清洗规则
    """
    
    @staticmethod
    def create_industrial_cleaner():
        """
        创建工业级清洗器
        包含所有必要的清洗规则
        """
        cleaner = TextCleaner()
        cleaner.add_rule('fix_encoding', TextCleaner.fix_encoding_errors)
        cleaner.add_rule('remove_page_breaks', TextCleaner.remove_page_breaks)
        cleaner.add_rule('clean_header_footer', TextCleaner.clean_header_footer)
        cleaner.add_rule('remove_noise', TextCleaner.remove_noise_patterns)
        cleaner.add_rule('normalize_whitespace', TextCleaner.normalize_whitespace)
        return cleaner
    
    @staticmethod
    def clean_document(text: str) -> str:
        """
        一站式文档清洗
        """
        cleaner = PipelineCleaner.create_industrial_cleaner()
        result = cleaner.clean(text)
        return result['cleaned']

3.2 解析质量评估体系

class ParsingQualityEvaluator:
    """
    解析质量自动评估器
    """
    
    def __init__(self):
        self.metrics = {}
    
    def evaluate(self, original_text: str, parsed_text: str) -> Dict[str, Any]:
        """
        评估解析质量
        
        返回多个维度的指标
        """
        # 1. 文本长度损失
        len_loss = (len(original_text) - len(parsed_text)) / len(original_text) if original_text else 0
        
        # 2. 字符保留率
        char_retention = len(set(original_text) & set(parsed_text)) / len(set(original_text)) if original_text else 0
        
        # 3. 特殊字符比例（异常信号）
        special_chars = re.findall(r'[^\w\s\u4e00-\u9fff，。！？、\n]', parsed_text)
        special_ratio = len(special_chars) / len(parsed_text) if parsed_text else 0
        
        # 4. 空白字符比例
        whitespace_ratio = len(re.findall(r'\s', parsed_text)) / len(parsed_text) if parsed_text else 0
        
        # 5. 平均句子长度
        sentences = re.split(r'[。！？\.\?\!]', parsed_text)
        avg_sentence_len = sum(len(s) for s in sentences) / len(sentences) if sentences else 0
        
        # 6. 乱码检测
        gibberish_score = self._detect_gibberish(parsed_text)
        
        # 7. 完整性检查（基于关键词覆盖率）
        keyword_coverage = self._check_keyword_coverage(original_text, parsed_text)
        
        # 8. 字符集分布（检查是否包含不常见字符）
        char_set_dist = self._char_set_distribution(parsed_text)
        
        result = {
            'length_loss': len_loss,
            'char_retention': char_retention,
            'special_ratio': special_ratio,
            'whitespace_ratio': whitespace_ratio,
            'avg_sentence_length': avg_sentence_len,
            'gibberish_score': gibberish_score,
            'keyword_coverage': keyword_coverage,
            'char_set_distribution': char_set_dist,
            'quality_score': 0
        }
        
        # 计算综合质量分 (0-100)
        result['quality_score'] = self._calculate_quality_score(result)
        result['pass'] = result['quality_score'] > 70
        
        return result
    
    def _detect_gibberish(self, text: str) -> float:
        """
        检测乱码比例
        返回0-1之间的分数，越高表示乱码越严重
        """
        # 检测不常见Unicode字符
        rare_chars = re.findall(r'[\uFFFD\u0000-\u0008\u000B-\u000C\u000E-\u001F]', text)
        rare_ratio = len(rare_chars) / len(text) if text else 0
        
        # 检测连续特殊字符
        special_sequence = re.findall(r'[^a-zA-Z0-9\u4e00-\u9fff\s]{3,}', text)
        seq_length = sum(len(s) for s in special_sequence)
        seq_ratio = seq_length / len(text) if text else 0
        
        return min(1, rare_ratio + seq_ratio / 2)
    
    def _check_keyword_coverage(self, original: str, parsed: str) -> float:
        """
        检查关键词覆盖率
        从原文提取关键词，检查是否在解析文本中出现
        """
        # 提取关键词（简单方法：取高频词）
        def extract_keywords(text: str, n: int = 10):
            words = re.findall(r'[\u4e00-\u9fff]{2,}|[a-zA-Z]{3,}', text)
            from collections import Counter
            return [word for word, _ in Counter(words).most_common(n)]
        
        original_keywords = extract_keywords(original)
        if not original_keywords:
            return 1.0
        
        parsed_text_lower = parsed.lower()
        covered = sum(1 for kw in original_keywords if kw.lower() in parsed_text_lower)
        return covered / len(original_keywords)
    
    def _char_set_distribution(self, text: str) -> Dict[str, float]:
        """
        字符集分布
        """
        total = len(text)
        if total == 0:
            return {}
        
        chinese = len(re.findall(r'[\u4e00-\u9fff]', text)) / total
        english = len(re.findall(r'[a-zA-Z]', text)) / total
        digits = len(re.findall(r'\d', text)) / total
        punctuation = len(re.findall(r'[，。！？、；：（）【】《》"\'.,!?;:()\[\]{}]', text)) / total
        spaces = len(re.findall(r'\s', text)) / total
        other = 1 - chinese - english - digits - punctuation - spaces
        
        return {
            'chinese': chinese,
            'english': english,
            'digits': digits,
            'punctuation': punctuation,
            'spaces': spaces,
            'other': other
        }
    
    def _calculate_quality_score(self, metrics: Dict[str, Any]) -> float:
        """
        计算综合质量分
        """
        score = 100.0
        
        # 长度损失扣分
        if metrics['length_loss'] > 0.3:
            score -= 20
        elif metrics['length_loss'] > 0.1:
            score -= 10
        
        # 乱码扣分
        score -= metrics['gibberish_score'] * 30
        
        # 特殊字符扣分
        if metrics['special_ratio'] > 0.05:
            score -= 15
        elif metrics['special_ratio'] > 0.02:
            score -= 5
        
        # 关键词覆盖率扣分
        score -= (1 - metrics['keyword_coverage']) * 20
        
        # 句子长度异常扣分
        if metrics['avg_sentence_length'] < 10:
            score -= 5
        
        return max(0, score)

---

第四部分：完整实战——解析与清洗流水线

class DocumentParsingPipeline:
    """
    完整文档解析流水线
    从PDF到干净训练语料
    """
    
    def __init__(self, output_dir: str = "./parsed_data"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
        
        self.pdf_extractor = PDFTextExtractor()
        self.ocr_processor = OCRProcessor()
        self.cleaner = PipelineCleaner()
        self.evaluator = ParsingQualityEvaluator()
        self.unstructured_parser = UnstructuredParser()
    
    def parse_document(
        self,
        pdf_path: str,
        force_ocr: bool = False,
        extract_tables: bool = True,
        min_quality_score: float = 70.0
    ) -> Dict[str, Any]:
        """
        解析单个文档
        """
        # 1. 检测PDF类型
        pdf_type = PDFTextExtractor.detect_pdf_type(pdf_path)
        print(f"📄 检测到PDF类型: {pdf_type}")
        
        # 2. 选择解析策略
        if force_ocr or pdf_type in ['scanned', 'mixed']:
            # 使用OCR
            print("🔍 执行OCR解析...")
            result = self.ocr_processor.extract_text_from_pdf(
                pdf_path, preprocess=True
            )
            text = '\n'.join([page['text'] for page in result['pages']])
            parsed_metadata = {
                'method': 'ocr',
                'confidence_scores': result['confidence_scores'],
                'errors': result['errors']
            }
        else:
            # 使用pdfplumber
            print("📖 执行文本提取...")
            result = self.pdf_extractor.extract_with_pdfplumber(pdf_path)
            text = result['text']
            parsed_metadata = {
                'method': 'pdfplumber',
                'tables': result.get('tables', []),
                'errors': result.get('errors', [])
            }
        
        # 3. 清洗文本
        print("🧹 执行文本清洗...")
        cleaned_text = self.cleaner.clean_document(text)
        
        # 4. 提取表格（可选）
        tables = []
        if extract_tables and pdf_type in ['text', 'mixed']:
            tables = TableExtractor.extract_tables_with_pdfplumber(pdf_path)
        
        # 5. 质量评估
        print("📊 执行质量评估...")
        quality = self.evaluator.evaluate(text, cleaned_text)
        
        # 6. 生成结果
        result = {
            'file_name': os.path.basename(pdf_path),
            'pdf_type': pdf_type,
            'original_text': text,
            'cleaned_text': cleaned_text,
            'parsed_metadata': parsed_metadata,
            'tables': tables,
            'quality': quality,
            'is_valid': quality['quality_score'] >= min_quality_score
        }
        
        return result
    
    def process_directory(
        self,
        directory_path: str,
        output_name: str = "training_corpus.jsonl"
    ) -> List[Dict[str, Any]]:
        """
        批量处理目录中的文档
        """
        pdf_files = []
        for root, _, files in os.walk(directory_path):
            for f in files:
                if f.lower().endswith('.pdf'):
                    pdf_files.append(os.path.join(root, f))
        
        print(f"📚 找到 {len(pdf_files)} 个PDF文件")
        
        valid_documents = []
        invalid_documents = []
        
        for pdf_path in pdf_files:
            print(f"\n{'='*60}")
            print(f"📄 处理: {os.path.basename(pdf_path)}")
            
            try:
                result = self.parse_document(pdf_path)
                
                if result['is_valid']:
                    valid_documents.append({
                        'source': pdf_path,
                        'content': result['cleaned_text'],
                        'metadata': {
                            'pdf_type': result['pdf_type'],
                            'quality_score': result['quality']['quality_score'],
                            'tables': result['tables']
                        }
                    })
                    print(f"   ✅ 有效 (质量分: {result['quality']['quality_score']:.1f})")
                else:
                    invalid_documents.append({
                        'source': pdf_path,
                        'reason': f"质量分 {result['quality']['quality_score']:.1f} < 阈值"
                    })
                    print(f"   ❌ 无效 (质量分: {result['quality']['quality_score']:.1f})")
                    
            except Exception as e:
                invalid_documents.append({
                    'source': pdf_path,
                    'reason': f"解析异常: {str(e)}"
                })
                print(f"   ❌ 异常: {str(e)}")
        
        # 输出统计
        print(f"\n{'='*60}")
        print(f"📊 处理完成")
        print(f"   ✅ 有效文档: {len(valid_documents)}")
        print(f"   ❌ 无效文档: {len(invalid_documents)}")
        print(f"   📈 成功率: {len(valid_documents)/len(pdf_files)*100:.1f}%")
        
        # 保存结果
        output_path = os.path.join(self.output_dir, output_name)
        import json
        with open(output_path, 'w', encoding='utf-8') as f:
            for doc in valid_documents:
                f.write(json.dumps(doc, ensure_ascii=False) + '\n')
        
        # 保存失败日志
        log_path = os.path.join(self.output_dir, "failed_documents.log")
        with open(log_path, 'w', encoding='utf-8') as f:
            for doc in invalid_documents:
                f.write(f"{doc['source']}: {doc['reason']}\n")
        
        return valid_documents

# ========== 使用示例 ==========
if __name__ == "__main__":
    pipeline = DocumentParsingPipeline(output_dir="./cleaned_corpus")
    
    # 处理整个目录
    valid_docs = pipeline.process_directory(
        directory_path="./raw_pdfs",
        output_name="training_corpus.jsonl"
    )
    
    # 查看质量报告
    import pandas as pd
    quality_scores = [doc['metadata']['quality_score'] for doc in valid_docs]
    print(f"\n📊 质量分布:")
    print(f"   平均: {sum(quality_scores)/len(quality_scores):.1f}")
    print(f"   最高: {max(quality_scores):.1f}")
    print(f"   最低: {min(quality_scores):.1f}")

---

总结：解析工程的四个核心原则

原则	说明	实践要点
策略路由	不同文档用不同工具	先检测类型，再选择最优方案
分层清洗	编码 → 页眉页脚 → 噪声 → 空白	从粗到细，层层递进
质量门槛	低质量数据直接丢弃	设置阈值，宁缺毋滥
人工抽样	自动化 + 人工抽查	定期抽检，防止系统性错误

核心洞察：在LLM训练中，数据质量是模型效果的乘法因子——再好的LoRA架构，喂进去的都是低质量数据，结果也不会好。优秀的解析能力，是构建高质量训练数据的基石。

---

原创声明：本文为CSDN博主原创文章，基于工业级数据清洗实践经验总结，欢迎交流讨论！

最后更新：2026年6月