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

用 Python 构建高质量的中文 Wikipedia 语料库：从原始 XML 到干净段落

news 2025/8/6 7:50:31

📌 一、模块作用

该脚本的核心作用是对从维基百科导出的数据（JSONL 格式）进行深度清洗、筛选和结构化处理，最终生成适用于下游自然语言处理（NLP）任务（如模型预训练、知识库构建、RAG 数据源）的干净语料。它专注于解决原始维基数据中包含大量噪声（如 HTML 标签、维基标记、模板信息）的问题。

适用场景：

  • 构建特定领域知识库，如本例中筛选“AI”和“教育”相关主题。
  • 为语言模型（LLM）准备高质量的预训练或微调数据集。
  • 在 RAG (Retrieval-Augmented Generation) 应用中，将非结构化的维基文章处理成结构化的知识片段。

🔢 二、输入输出说明

该脚本主要通过命令行参数接收输入，并输出一个处理后的文件。

  • 输入

    • dir_path (str): 存放原始维基百科 .jsonl 文件的目录路径。
    • num_workers (int): 用于并行处理文件的线程数，默认为 CPU 核心数。
    • max_length (int): 每个文本段（segment）包含的最大句子数量，用于文本切分。
    • stride (int): 文本切分时的滑动窗口步长，决定段落间的重叠度。

  • 输出

    • 一个名为 education.jsonl 的文件，其中每一行都是一个 JSON 对象，包含从原始文章中提取并处理好的一个文本段落。
    {"text": "处理后的一段文本..."}

🔧 三、核心逻辑

整个脚本的执行流程可以分为三个核心步骤：数据加载与筛选、文本深度清洗 和 文本分段。

  1. 数据加载与筛选 (load_corpus)

    • 文件遍历: 递归扫描输入目录 dir_path 下的所有文件。
    • 关键词过滤: 定义一个关键词列表（如 ['大模型', '机器学习', '深度学习']）。在读取每个 .jsonl 文件时，只保留 text 字段包含这些关键词的条目。
    • 并行处理: 利用 concurrent.futures.ThreadPoolExecutor 实现多线程并行读取文件，极大地提升了在多文件场景下的 I/O 效率。
    • 数量控制: 设置了一个阈值（例如 90），当收集到的语料条目达到该数量时即停止，便于快速生成小规模测试集。

  2. 文本深度清洗 (basic_process)
    这是整个脚本技术含量最高的部分。它使用大量正则表达式来清除维基百科特有的标记语言和格式噪声。

    • HTML 解码: 使用 html.unescape 将 & 等 HTML 实体转换回原始字符。
    • 页面类型排除: 过滤掉无用页面，如“消歧义页 (disambiguation)”、”列表页 (List of…)“ 和 ”重定向页 (REDIRECT)“。
    • 模板与标记移除: 核心清洗步骤，通过一系列精细的正则表达式，移除维基语法标记，例如：
      • 引用标记: {{cite ...}}
      • 内部链接: [[...]]
      • 数学公式/化学式: <math>...</math>, <chem>...</chem>
      • 表格样式与属性: |item_style=..., |width=... 等几十种规则。
      • 残留的 HTML 标签: <br>, <ref>...</ref>
    • 元数据清理: 清理如 File:..., Source:... 等图片和来源信息。

  3. 文本分段 (create_segments)
    为了让长文本适用于有长度限制的模型，需要将清洗后的文本切分成合适的段落。

    • 句子分割: 使用 spaCy 库的句子分割功能 (doc.sents)。相比简单的按句号切分，spaCy 能更准确地处理复杂的句子边界（如缩写 Mr.）。
    • 滑动窗口分段: 采用滑动窗口机制，按照 max_length (窗口大小) 和 stride (步长) 将句子组合成段落。例如，max_length=3, stride=1 会生成 [sent1, sent2, sent3], [sent2, sent3, sent4] … 这样的重叠段落，有助于在 RAG 检索时保留更完整的上下文信息。

💻 四、代码实现

以下是脚本中几个关键函数的简化示例代码。

1. 文本清洗 basic_process

import html
import redef basic_process(title, text):"""对单篇维基文章进行深度清洗"""try:# 解码HTML实体title = html.unescape(title)text = html.unescape(text)# 过滤掉消歧义页和重定向页if '(disambiguation)' in title.lower() or text.strip().upper().startswith("REDIRECT"):return None, None# 核心：使用正则表达式移除维基标记和噪声# 移除引用text = re.sub(r'\\{\\{cite .*?\\}\\}', ' ', text, flags=re.DOTALL)# 移除内部链接标记text = text.replace(r"[[", " ").replace(r"]]]", " ")# 移除数学公式text = re.sub(r'<math.*?</math>', '', text, flags=re.DOTALL)# 移除各种样式属性（示例）text = re.sub(r'\\| ?style= ?.*? ', ' ', text)# 移除换行符和制表符text = re.sub(r'[\\n\\t]', ' ', text)# 移除所有剩余的HTML标签text = re.sub(r'<.*?>', '', text)# 移除多余空格text = re.sub(r'\\s+', ' ', text).strip()return title, textexcept Exception as e:print(f"Error processing title '{title}': {e}")return None, None

2. 文本分段 create_segments

import spacy# 需要提前加载spacy模型: python -m spacy download en_core_web_sm
nlp = spacy.load("en_core_web_sm")def create_segments(doc_text, max_length, stride):"""使用spaCy和滑动窗口进行文本分段"""try:doc = nlp(doc_text.strip())sentences = [sent.text.strip() for sent in doc.sents]segments = []# 滑动窗口for i in range(0, len(sentences), stride):segment = " ".join(sentences[i:i + max_length])segments.append(segment)if i + max_length >= len(sentences):breakreturn segmentsexcept Exception as e:print(f"Error creating segments: {e}")return []

🧪 五、测试建议

为确保该预处理脚本的稳定性和准确性，建议进行以下测试：

  • 单元测试 (pytest):
    • 针对 basic_process 函数，创建一组测试用例，覆盖各种维基标记、HTML 实体和边缘情况（如空文本、只有标记的文本）。
    • 针对 create_segments 函数，测试不同 max_length 和 stride 组合下的分段结果是否符合预期，以及对长句、短句的处理是否正确。
  • 集成测试:
    • 使用一个包含少量（例如 5-10 个）.jsonl 文件的小型测试数据集，完整运行脚本。
    • 检查输出的 education.jsonl 文件是否存在，格式是否为合法的 JSONL。
    • 人工抽查几条输出的文本段落，确认清洗效果和分段逻辑是否正确。
  • 边界用例:
    • 测试输入目录为空、目录不存在的情况。
    • 测试 .jsonl 文件内容不合法（如非 JSON 格式）时的异常处理。
    • 测试文本内容极长或极短时的处理情况。

完整代码

import argparse
from tqdm import tqdm
import re
import html
import spacy
import os
import json
import subprocess
from pathlib import Path
import shutil
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Pool
import os
import jsondef load_corpus(dir_path):"""该函数从给定目录路径读取 .jsonl 文件，提取其文本字段中包含指定关键词的 JSON 条目。一旦语料库中的条目数量达到 设定的值 ，就停止添加。此外，它还利用多线程来加速文件处理。"""# 定义要在文本文件中查找的关键词列表keywords = ['智能教育','大模型','机器学习','深度学习','算法','自然语言处理']def iter_files(path):"""遍历位于根路径下的所有文件。"""if os.path.isfile(path):# 如果路径是文件，直接返回该文件yield pathelif os.path.isdir(path):# 如果路径是目录，遍历该目录中的每个文件for dirpath, _, filenames in os.walk(path):for f in filenames:yield os.path.join(dirpath, f)else:# 如果路径既不是目录也不是文件，抛出错误raise RuntimeError('Path %s is invalid' % path)def read_jsonl_file(file_path):"""读取 .jsonl 文件的行，并将相关数据添加到语料库中。"""with open(file_path, 'r', encoding='utf-8') as f:for line in f:json_data = json.loads(line)# 检查 JSON 数据的文本字段中是否包含任何关键词if any(keyword in json_data['text'] for keyword in keywords):corpus.append(json_data)# 如果语料库大小达到 90，停止添加if len(corpus) == 90:break# 从目录中收集所有文件路径all_files = [file for file in iter_files(dir_path)]# 初始化语料库列表corpus = []# 使用 ThreadPoolExecutor 并行读取文件with ThreadPoolExecutor(max_workers=args.num_workers) as executor:# 提交任务以读取每个文件for file_path in all_files:executor.submit(read_jsonl_file, file_path)# 返回填充好的语料库return corpusdef create_segments(doc_text, max_length, stride):"""数据预处理，在处理需要特定输入长度限制的NLP模型时，如某些类型的文本分类或语言理解模型。通过控制 max_length 和 stride 参数，可以灵活调整每个文本段的长度和重叠程度，以适应不同的处理需求。"""# 去除文档文本的首尾空白字符doc_text = doc_text.strip()# 使用NLP工具SpaCy解析文本，得到文档对象doc = nlp(doc_text)# 从文档对象中提取句子，并去除每个句子的首尾空格sentences = [sent.text.strip() for sent in doc.sents]# 初始化一个空列表，用于存储所有生成的文本段落segments = []# 使用步长 stride 遍历句子列表的索引for i in range(0, len(sentences), stride):# 将从当前索引开始的 max_length 个句子连接成一个字符串segment = " ".join(sentences[i:i + max_length])# 将构建的段落添加到列表中segments.append(segment)# 如果当前索引加上最大长度超过了句子列表的长度，停止循环if i + max_length >= len(sentences):break# 返回包含所有段落的列表return segmentsdef basic_process(title, text):# 使用html.unescape函数来解码HTML实体，恢复文本中的特殊字符title = html.unescape(title)text = html.unescape(text)# 删除文本首尾的空白字符text = text.strip()# # 如果标题含有特定的消歧义标记，则不处理这类页面if '(disambiguation)' in title.lower():return None, Noneif '(disambiguation page)' in title.lower():return None, None# 排除以列表、索引或大纲开头的页面，这些页面大多只包含链接if re.match(r'(List of .+)|(Index of .+)|(Outline of .+)',title):return None, None# 排除重定向页面if text.startswith("REDIRECT") or text.startswith("redirect"):return None, None# 如果文本以 ". References." 结尾，则删除该部分if text.endswith(". References."):text = text[:-len(" References.")].strip()# 删除文本中的特定格式标记，如引用标记text = re.sub('\{\{cite .*?\}\}', ' ', text, flags=re.DOTALL)# 替换或删除不必要的格式和标签text = text.replace(r"TABLETOREPLACE", " ")text = text.replace(r"'''", " ")text = text.replace(r"[[", " ")text = text.replace(r"]]", " ")text = text.replace(r"{{", " ")text = text.replace(r"}}", " ")text = text.replace("<br>", " ")text = text.replace("&quot;", "\"")text = text.replace("&amp;", "&")text = text.replace("& amp;", "&")text = text.replace("nbsp;", " ")text = text.replace("formatnum:", "")# 删除特定HTML标签内的文本，如<math>, <chem>, <score>text = re.sub('<math.*?</math>', '', text, flags=re.DOTALL)text = re.sub('<chem.*?</chem>', '', text, flags=re.DOTALL)text = re.sub('<score.*?</score>', '', text, flags=re.DOTALL)# 使用正则表达式删除样式相关的属性，例如：item_style, col_style等text = re.sub('\| ?item[0-9]?_?style= ?.*? ', ' ', text)text = re.sub('\| ?col[0-9]?_?style= ?.*? ', ' ', text)text = re.sub('\| ?row[0-9]?_?style= ?.*? ', ' ', text)text = re.sub('\| ?style= ?.*? ', ' ', text)text = re.sub('\| ?bodystyle= ?.*? ', ' ', text)text = re.sub('\| ?frame_?style= ?.*? ', ' ', text)text = re.sub('\| ?data_?style= ?.*? ', ' ', text)text = re.sub('\| ?label_?style= ?.*? ', ' ', text)text = re.sub('\| ?headerstyle= ?.*? ', ' ', text)text = re.sub('\| ?list_?style= ?.*? ', ' ', text)text = re.sub('\| ?title_?style= ?.*? ', ' ', text)text = re.sub('\| ?ul_?style= ?.*? ', ' ', text)text = re.sub('\| ?li_?style= ?.*? ', ' ', text)text = re.sub('\| ?border-style= ?.*? ', ' ', text)text = re.sub('\|? ?style=\".*?\"', '', text)text = re.sub('\|? ?rowspan=\".*?\"', '', text)text = re.sub('\|? ?colspan=\".*?\"', '', text)text = re.sub('\|? ?scope=\".*?\"', '', text)text = re.sub('\|? ?align=\".*?\"', '', text)text = re.sub('\|? ?valign=\".*?\"', '', text)text = re.sub('\|? ?lang=\".*?\"', '', text)text = re.sub('\|? ?bgcolor=\".*?\"', '', text)text = re.sub('\|? ?bg=\#[a-z]+', '', text)text = re.sub('\|? ?width=\".*?\"', '', text)text = re.sub('\|? ?height=[0-9]+', '', text)text = re.sub('\|? ?width=[0-9]+', '', text)text = re.sub('\|? ?rowspan=[0-9]+', '', text)text = re.sub('\|? ?colspan=[0-9]+', '', text)text = re.sub(r'[\n\t]', ' ', text)text = re.sub('<.*?/>', '', text)text = re.sub('\|? ?align=[a-z]+', '', text)text = re.sub('\|? ?valign=[a-z]+', '', text)text = re.sub('\|? ?scope=[a-z]+', '', text)text = re.sub('&lt;ref&gt;.*?&lt;/ref&gt;', ' ', text)text = re.sub('&lt;.*?&gt;', ' ', text)text = re.sub('File:[A-Za-z0-9 ]+\.[a-z]{3,4}(\|[0-9]+px)?', '', text)text = re.sub('Source: \[.*?\]', '', text)# 清理可能因XML导出错误而残留的格式标签# 使用正则表达式匹配并替换各种样式相关的属性text = text.replace("Country flag|", "country:")text = text.replace("flag|", "country:")text = text.replace("flagicon|", "country:")text = text.replace("flagcountry|", "country:")text = text.replace("Flagu|", "country:")text = text.replace("display=inline", "")text = text.replace("display=it", "")text = text.replace("abbr=on", "")text = text.replace("disp=table", "")title = title.replace("\n", " ").replace("\t", " ")return title, textdef split_list(lst, n):"""将一个列表分割成 n 个大致相等的部分。"""# k 是每个部分的基本大小，m 是 len(lst) 除以 n 的余数k, m = divmod(len(lst), n)# 使用列表推导来生成子列表# 计算每个部分的起始和结束索引return [lst[i * k + min(i, m):(i + 1) * k + min(i + 1, m)] for i in range(n)]def single_worker(docs):"""处理一个文档列表，对每个文档应用清洗和格式化操作。"""# 初始化结果列表，用于存储处理后的文档results = []# 遍历文档列表，使用 tqdm 库显示进度条for item in tqdm(docs):# 应用基础处理函数 basic_process 来处理每个文档的标题和文本title, text = basic_process(item[0], item[1])# 如果处理后的标题是 None（可能因为文档不符合处理要求），则跳过当前循环if title is None:continue# 格式化标题，加上双引号title = f"\"{title}\""# 将处理和格式化后的标题和文本以元组形式添加到结果列表中results.append((title, text))return resultsif __name__ == '__main__':parser = argparse.ArgumentParser(description='Generate clean wiki corpus file for indexing.')parser.add_argument('--dump_path', type=str)parser.add_argument('--seg_size', default=None, type=int)parser.add_argument('--stride', default=None, type=int)parser.add_argument('--num_workers', default=4, type=int)parser.add_argument('--save_path', type=str, default='clean_corpus.jsonl')args = parser.parse_args()# 设置临时目录用于存储WikiExtractor的输出temp_dir = os.path.join(Path(args.save_path).parent, 'temp')# 创建临时目录os.makedirs(temp_dir)# 使用wikiextractor从维基百科转储中提取文本，输出为JSON格式，过滤消歧义页面subprocess.run(['python', '-m','wikiextractor.WikiExtractor','--json', '--filter_disambig_pages', '--quiet','-o', temp_dir,'--process', str(args.num_workers),args.dump_path])# 载入处理后的语料库corpus = load_corpus(temp_dir)# 加载Spacy中文模型nlp = spacy.load("zh_core_web_lg")# 初始化一个字典来存储文档，以避免页面重复documents = {}# 使用tqdm显示进度条for item in tqdm(corpus):title = item['title']text = item['text']# # 检查标题是否已存在于字典中，以合并同一个标题下的不同部分if title in documents:documents[title] += " " + textelse:documents[title] = text# 开始预处理文本print("Start pre-processing...")documents = list(documents.items())# 使用Python的多进程库，创建进程池进行并行处理with Pool(processes=args.num_workers) as p:result_list = list(tqdm(p.imap(single_worker, split_list(documents,args.num_workers))))result_list = sum(result_list, [])all_title = [item[0] for item in result_list]all_text = [item[1] for item in result_list]print("Start chunking...")idx = 0clean_corpus = []# 使用spaCy的pipe方法进行高效的文本处理，指定进程数和批处理大小for doc in tqdm(nlp.pipe(all_text, n_process=args.num_workers, batch_size=10), total=len(all_text)):# 获取当前文档的标题title = all_title[idx]# 索引递增，指向下一个标题idx += 1# 初始化段落列表segments = []# 初始化单词计数器word_count = 0# 初始化段落的token列表segment_tokens = []# 遍历文档中的每个tokenfor token in doc:# token（包括空格）添加到段落令牌列表segment_tokens.append(token.text_with_ws)# 如果令牌不是空格也不是标点if not token.is_space and not token.is_punct:# 单词计数加一word_count+=1# 如果单词计数达到100，则重置计数器，生成一个新段落if word_count == 100:word_count = 0segments.append(''.join([token for token in segment_tokens]))segment_tokens = []# 检查最后是否还有剩余的单词没有形成完整段落if word_count != 0:for token in doc:segment_tokens.append(token.text_with_ws)if not token.is_space and not token.is_punct:word_count+=1if word_count == 100:word_count = 0segments.append(''.join([token for token in segment_tokens]))break# 检查最后一组token是否已添加到segmentsif word_count != 0:segments.append(''.join([token for token in segment_tokens]))for segment in segments:text = segment.replace("\n", " ").replace("\t", " ")# 将处理后的标题和文本以字典形式添加到清洗后的语料库列表clean_corpus.append({"title": title, "text": text})# 删除临时目录及其内容#shutil.rmtree(temp_dir)print("Start saving corpus...")# 检查保存路径的目录是否存在，如果不存在则创建os.makedirs(os.path.dirname(args.save_path), exist_ok=True)# 打开指定的文件路径进行写入，设置编码为utf-8with open(args.save_path, "w", encoding='utf-8') as f:# 遍历清洗后的语料库列表，每个元素都是一个包含标题和文本的字典for idx, item in enumerate(clean_corpus):# 将字典转换为JSON字符串格式，确保使用UTF-8编码来处理Unicode字符json_string = json.dumps({'id': idx,         # 添加唯一ID'title': item['title'],      # 文章标题'contents': item['text']     # 文章内容}, ensure_ascii=False)  # 确保不将Unicode字符编码为ASCII# 将JSON字符串写入文件，并在每个条目后添加换行符f.write(json_string + '\n')print("Finish!")
http://www.lryc.cn/news/610638.html

相关文章：

  • 【taro react】 ---- useModel 数据双向绑定 hook 实现
  • 【乐企板式文件生成工程】关于乐企板式文件（PDF/OFD/XML）生成工程介绍
  • Taro Hooks 完整分类详解
  • wps创建编辑excel customHeight 属性不是标准 Excel Open XML导致比对异常
  • 云计算一阶段Ⅱ——11. Linux 防火墙管理
  • 《Node.js与 Elasticsearch的全文搜索架构解析》
  • Sentinel全面实战指南
  • 剑指offer第2版：字符串
  • Day34 GPU训练及类的call方法
  • Android audio之 AudioDeviceInventory
  • PCBA电子产品复制全攻略：从入门到精通
  • 【音视频】WebRTC 一对一通话-信令服
  • 强化学习_Paper_1991_Reinforcement learning is direct adaptive optimal control
  • 自然语言处理×第三卷：文本数据分析——她不再只是贴着你听，而开始学会分析你语言的结构
  • python+MySQL组合实现生成销售财务报告
  • 游戏画面总是卡顿怎么办 告别延迟畅玩游戏
  • 电脑搜索不到公司无线网络
  • 基于ARM+FPGA多通道超声信号采集与传输系统设计
  • NuGet03-私有仓库搭建
  • mac前端环境安装
  • 【ARM】CMSIS6 介绍
  • Mac上pnpm的安装与使用
  • AIDL学习
  • 《算法导论》第 2 章 - 算法基础
  • 朴素贝叶斯（Naive Bayes）算法详解
  • pipeline方法关系抽取--课堂笔记
  • 神坛上的transformer
  • VUE2 学习笔记18 路由守卫
  • 无人机 × 巡检 × AI识别：一套可复制的超低延迟低空视频感知系统搭建实践
  • 人月神话：软件工程的永恒智慧