3.1 文件的功能

上一章我们聊了模型调用，把发动机准备好了，这一章我们来看看数据从原始文件到干净文本的全过程，data_preprocess.py这个文件负责完成这些任务，它是整个项目的消化系统——不管你是 PDF、Word、CSV 还是 TXT，它都能给你读进来，把里面的文字和表格提取出来，再洗干净、切分开，最后吐出一个规规矩矩的 JSON 文件，方便后面建向量库。

简单说就是：把非结构化数据转换成结构化数据，具体的说是把不同类型的文档变成大模型能看懂的向量数据。这里面要主要的地方很多，尤其是编码问题、表格提取、长文档切分，我一个个讲。

3.2  文件读取与统一接口

一开始设计的时候我想：不能每种文件类型写一套调用逻辑，那样主函数会乱成一锅粥。所以搞了一个统一入口——extract_and_convert(file_path)，这样只管把文件路径传进来，它自己判断类型，调用对应的处理函数，最后返回一个统一格式的字典。

def extract_and_convert(file_path):
    file_type = detect_file_type(file_path)
    if file_type == 'pdf':
        result = extract_from_pdf(file_path)
    elif file_type == 'doc':
        result = extract_from_doc(file_path)
    elif file_type == 'csv':
        result = extract_from_csv(file_path)
    else:
        result = extract_from_txt(file_path)
    # 添加通用元数据
    result['metadata']['file_name'] = Path(file_path).name
    return result

返回的字典长这样：

{
    'raw_text': '...',           # 提取的纯文本
    'tables': [table1, table2],  # 表格的原始数据
    'metadata': {'type': 'pdf', 'pages': 5, 'file_name': 'xxx.pdf'}
}

后面不管是建向量库还是做数据切分，都只用接收处理输出这一种格式

3.3 PDF处理

读取 PDF 我用的是 pdfplumber，比起PyPDF2，pdfplumber 对表格的结构识别很准，可以保留位置信息，让处理表格数据成为可能

with pdfplumber.open(file_path) as pdf:
    for page in pdf.pages:
        page_text = page.extract_text() or ''   # 提取文字
        tables = page.extract_tables()          # 提取表格
        for table in tables:
            # 清理空单元格，保留非空行
            clean_table = []
            for row in table:
                clean_row = [cell if cell else '' for cell in row]
                if any(clean_row):
                    clean_table.append(clean_row)
            # 把表格存到 tables 列表，同时在文本里加个标记
            result['tables'].append(clean_table)
            page_text += f"\n\n[表格 {len(result['tables'])}]\n"
            for row in clean_table[:3]:
                page_text += ' | '.join(str(c)[:20] for c in row) + '\n'

对于双栏PDF，提取后文本顺序会乱，可用pdfplumber的page.within_bbox截取左半页和右半页分别提取，或后期用规则重排行序

这里有个细节：表格里可能有很多空行，我加了个 any(clean_row) 过滤掉全空的行，这样后面检索不会浪费资源

3.4 Word 处理

Word 文档用 python-docx 处理。需要主要的是表格里的单元格可能有多段文字，直接取 cell.text 会把所有段落拼一起，但换行符会丢失，最后写成这样：

cell_text = ' '.join([p.text for p in cell.paragraphs])

这样段落之间用空格连接，语义不会断，另外 Word 里的表格也可能有空行，同样要过滤。

3.5 CSV 处理

CSV的编码方式五花八门，UTF-8加上BOM、GBK、GB2312，直接用utf-8打开会有一堆报错，所以我用了chardet自动检测编码

def read_csv_file(file_path, encoding=None):
    if encoding is None:
        with open(file_path, 'rb') as f:
            raw_data = f.read()
            encoding = chardet.detect(raw_data)['encoding'] or 'utf-8'
    try:
        df = pd.read_csv(file_path, encoding=encoding)
    except:
        # 如果失败了，尝试其他常见编码
        for enc in ['utf-8', 'gbk', 'gb2312', 'latin-1', 'cp1252', 'big5']:
            try:
                return read_csv_file(file_path, encoding=enc)
            except:
                continue

有些 CSV 文件前几行是空行，或者全是空列，用 pandas 读进来之后需要清理：

df = df.dropna(how='all').dropna(axis=1, how='all')

3.6 文本提取与转换

每个文件处理完，最终都变成前面说的统一字典。这里面我加了很多元数据，比如文件大小、页数、行数等，后面检索时可以过滤用

result['metadata'].update({
    'file_path': file_path,
    'file_name': Path(file_path).name,
    'file_size': Path(file_path).stat().st_size
})

3.7 文本切分

原始文档可能很长，直接转向量会丢失细节，而且大模型上下文有限。所以需要把长文本切成小块。我写了两种切分方式：

按章节标题切分：根据关键词识别标题，然后把内容归到对应的章节。比如看到“基础信息”、“指标数据”这种短行，就认为是标题，后面的内容属于它

for line in lines:
    for section, keywords in default_keywords.items():
        if any(kw in line.lower() for kw in keywords) and len(line) < 50:
            current_section = section
            break
    sections[current_section] += line + '\n'

这里判断 len(line) < 50 是为了防止把正文里的关键词也当成标题。

按规则切分：如果文档没有明显的章节标题，就用关键词匹配段落。比如段落里出现“公司”、“注册”就归到“基础信息”；出现“增长率”、“百分比”就归到“指标数据”。

两种方法互相帮助，总能找到一种方式把文本结构化。

3.8 数据清洗

切分完的文本里有很多噪音：多余的空格、空行、特殊符号，还有各种格式的日期，我写了个clean_text_data函数统一清洗

# 去除多余空格和空行
text = re.sub(r'\s+', ' ', text)
text = re.sub(r'\n\s*\n', '\n\n', text)

# 只保留中英文、数字和基本标点
text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s\.,;:!?\-\(\)\[\]%￥元万元亿]', '', text)

# 日期标准化：2020年3月15日 → 2020-3-15
date_patterns = [
    (r'(\d{4})年(\d{1,2})月(\d{1,2})日', r'\1-\2-\3'),
    (r'(\d{4})\.(\d{1,2})\.(\d{1,2})', r'\1-\2-\3'),

]
for pattern, repl in date_patterns:
    text = re.sub(pattern, repl, text)

3.9 数值提取

很多问题涉及数值，比如“注册资本是多少”、“增长率是多少”。所以我专门写了个extract_numeric_data函数，把文本里各种形式的数字都抠出来

# 所有数字
numbers = re.findall(r'-?\d+\.?\d*', text)

# 百分比
percentages = re.findall(r'(\d+\.?\d*)%', text)

# 金额，统一换算成元
amount_patterns = [
    (r'(\d+\.?\d*)(?=\s*元)', 1),
    (r'(\d+\.?\d*)(?=\s*万元)', 10000),
    (r'(\d+\.?\d*)(?=\s*亿元)', 100000000)
]
for pattern, multiplier in amount_patterns:
    amounts = re.findall(pattern, text)
    result['amounts'].extend([float(a) * multiplier for a in amounts])

3.10 缺失值处理

提取出来的数值列表可能有缺失，比如有些问题没提到具体指标，我提供了几种处理策略：

remove：直接删掉缺失值

zero：补 0

mean：补均值

median：补中位数

这样可以保证后面做数据统计时不出错

3.11 综合流程

最后用 clean_and_extract_pipeline 把清洗、数值提取、缺失值处理、统计信息全部串起来。输入是切分好的章节字典，输出是包含清洗后文本和各种统计的字典

result = {
    'cleaned_text': cleaned_text,
    'statistics': section_stats,
    'summary': {
        'numeric_stats': {...},
        'text_stats': {...}
    }
}

运行结构：