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

48.文本预处理：给文字数据洗个澡

news 2025/10/15 8:00:01

文本预处理：给文字数据洗个澡

🎯 前言：数据界的洗衣店

想象一下，你刚从泥巴地里踢完足球回家，满身污泥，这时候你妈妈会让你直接上餐桌吃饭吗？当然不会！你得先去洗个澡，换身干净衣服。

文本数据也是一样的道理。从互联网上爬取的文本、用户输入的评论、社交媒体的动态，这些"野生"的文本数据就像刚从泥巴地里踢球回来的你一样，充满了各种"污垢"：

多余的空格和换行符（就像衣服上的泥点）
大小写不统一（就像袜子一只高一只低）
标点符号混乱（就像头发乱糟糟的）
停用词满天飞（就像口袋里装满了废纸）

今天我们就来学习如何给文字数据"洗个澡"，让它们变得干干净净，整整齐齐，准备好被我们的AI模型"招待"！

📚 目录

🧠 核心概念：什么是文本预处理
🛠️ 基础清洗：去除脏数据
🔤 标准化处理：统一格式
📝 分词与词性标注
🚫 停用词过滤
🏷️ 词干提取与词形还原
💻 实战项目：构建文本预处理管道
🚀 进阶技巧：正则表达式的威力
🔧 常见问题与解决方案
📖 扩展阅读
🎬 下集预告
📝 总结与思考题

🧠 核心概念：什么是文本预处理

文本预处理就像是数据界的洗衣店，它的任务是把"脏兮兮"的原始文本变成"干干净净"的标准文本。这个过程包括：

1. 清洗（Cleaning）

去除无用的字符、符号和格式：

# 原始文本（脏衣服）
raw_text = "   Hello World!!! \n\n  这是一个测试文本...   "# 清洗后（干净衣服）
clean_text = "Hello World! 这是一个测试文本"

2. 标准化（Normalization）

统一文本格式，让所有文本都遵循相同的规范：

# 大小写不统一（乱穿衣服）
messy_text = "PyThOn is GREAT for NLP"# 标准化后（统一着装）
normalized_text = "python is great for nlp"

3. 分词（Tokenization）

把连续的文本切分成单个词语，就像把一整块布料裁成合适的布片：

# 原始句子
sentence = "自然语言处理很有趣"# 分词后
tokens = ["自然", "语言", "处理", "很", "有趣"]

4. 过滤（Filtering）

去除对分析没有帮助的词语，就像挑选衣服时扔掉过时的款式：

# 包含停用词
with_stopwords = ["我", "觉得", "这个", "电影", "很", "好看"]# 过滤停用词后
without_stopwords = ["电影", "好看"]

🛠️ 基础清洗：去除脏数据

首先，让我们来安装必要的工具包：

# 安装必要的包
pip install jieba nltk pandas numpy reimport re
import jieba
import pandas as pd
import numpy as np
from collections import Counter

1. 去除特殊字符和多余空格

def basic_clean(text):"""基础清洗函数：去除特殊字符和多余空格就像洗衣机的预洗功能"""# 去除HTML标签text = re.sub(r'<[^>]+>', '', text)# 去除URL链接text = re.sub(r'http\S+|www\S+|https\S+', '', text)# 去除邮箱地址text = re.sub(r'\S+@\S+', '', text)# 去除多余的空白字符text = re.sub(r'\s+', ' ', text)# 去除首尾空格text = text.strip()return text# 测试一下
dirty_text = """<p>这是一个测试文本！！！</p>访问我们的网站：https://example.com联系我们：contact@example.com有很多       空格和换行符
"""clean_text = basic_clean(dirty_text)
print(f"清洗前：{repr(dirty_text)}")
print(f"清洗后：{repr(clean_text)}")

2. 处理标点符号

def handle_punctuation(text, remove_all=False):"""处理标点符号remove_all=True：完全移除标点符号remove_all=False：标准化标点符号"""if remove_all:# 移除所有标点符号text = re.sub(r'[^\w\s]', '', text)else:# 标准化标点符号text = re.sub(r'[！]{2,}', '!', text)  # 多个感叹号变成一个text = re.sub(r'[？]{2,}', '?', text)  # 多个问号变成一个text = re.sub(r'[。]{2,}', '。', text)  # 多个句号变成一个text = re.sub(r'[…]{2,}', '…', text)  # 多个省略号变成一个return text# 测试
punctuation_text = "这真的太好了！！！难道不是吗？？？我觉得是的。。。"
print(f"原文：{punctuation_text}")
print(f"标准化：{handle_punctuation(punctuation_text)}")
print(f"移除所有：{handle_punctuation(punctuation_text, remove_all=True)}")

3. 处理数字

def handle_numbers(text, method='keep'):"""处理数字的不同方法keep: 保持原样remove: 移除所有数字replace: 替换为特殊标记"""if method == 'remove':text = re.sub(r'\d+', '', text)elif method == 'replace':text = re.sub(r'\d+', '<NUM>', text)# keep的话就不做任何处理return text# 测试
number_text = "我有3个苹果和5个橙子，总共8个水果"
print(f"原文：{number_text}")
print(f"保持数字：{handle_numbers(number_text, 'keep')}")
print(f"移除数字：{handle_numbers(number_text, 'remove')}")
print(f"替换数字：{handle_numbers(number_text, 'replace')}")

🔤 标准化处理：统一格式

1. 大小写转换

def normalize_case(text, method='lower'):"""大小写标准化就像给所有人统一发校服"""if method == 'lower':return text.lower()elif method == 'upper':return text.upper()elif method == 'title':return text.title()else:return text# 测试
mixed_case = "PyThOn Machine Learning is AWESOME"
print(f"原文：{mixed_case}")
print(f"小写：{normalize_case(mixed_case, 'lower')}")
print(f"大写：{normalize_case(mixed_case, 'upper')}")
print(f"标题：{normalize_case(mixed_case, 'title')}")

2. 全角半角转换

def normalize_width(text):"""全角半角字符转换把所有字符都穿上"标准尺码"的衣服"""# 全角转半角result = ""for char in text:code = ord(char)if code == 0x3000:  # 全角空格result += chr(0x0020)elif 0xFF01 <= code <= 0xFF5E:  # 全角字符result += chr(code - 0xFEE0)else:result += charreturn result# 测试
fullwidth_text = "Ｈｅｌｌｏ，Ｗｏｒｌｄ！"
print(f"原文：{fullwidth_text}")
print(f"转换后：{normalize_width(fullwidth_text)}")

3. 繁简体转换

# 需要安装：pip install opencc-python-reimplemented
import openccdef convert_chinese(text, direction='t2s'):"""繁简体转换direction: 't2s' 繁体转简体，'s2t' 简体转繁体"""if direction == 't2s':converter = opencc.OpenCC('t2s')else:converter = opencc.OpenCC('s2t')return converter.convert(text)# 测试（如果安装了opencc）
try:traditional_text = "機器學習很有趣"simplified_text = convert_chinese(traditional_text, 't2s')print(f"繁体：{traditional_text}")print(f"简体：{simplified_text}")
except:print("需要安装opencc包来进行繁简体转换")

📝 分词与词性标注

1. 中文分词

import jieba
import jieba.posseg as psegdef chinese_tokenize(text, use_paddle=False):"""中文分词：把连续的汉字切分成词语就像把一整块蛋糕切成一片片"""if use_paddle:# 使用paddle模式（更准确但需要安装paddlepaddle）jieba.enable_paddle()# 精确模式分词words = list(jieba.cut(text, cut_all=False))return words# 测试
chinese_text = "自然语言处理是人工智能的重要分支"
words = chinese_tokenize(chinese_text)
print(f"原文：{chinese_text}")
print(f"分词结果：{words}")

2. 词性标注

def pos_tagging(text):"""词性标注：给每个词语标注词性就像给每个人贴上职业标签"""words_with_pos = list(pseg.cut(text))result = []for word, pos in words_with_pos:result.append((word, pos))return result# 测试
pos_result = pos_tagging(chinese_text)
print(f"词性标注结果：")
for word, pos in pos_result:print(f"{word}/{pos}", end="  ")
print()

3. 自定义词典

def add_custom_words():"""添加自定义词典就像教分词工具认识新的"专业术语""""# 添加自定义词汇jieba.add_word("人工智能", freq=1000, tag='n')jieba.add_word("机器学习", freq=1000, tag='n')jieba.add_word("深度学习", freq=1000, tag='n')# 也可以从文件加载# jieba.load_userdict("user_dict.txt")# 测试自定义词典的效果
add_custom_words()
custom_text = "深度学习是机器学习的一个分支"
print(f"添加自定义词典前：{list(jieba.cut(custom_text))}")

🚫 停用词过滤

停用词就像是文本中的"废话"，比如"的"、“是”、"在"这些词虽然在语法上必要，但对文本分析往往没有太大帮助。

def create_stopwords_list():"""创建停用词列表就像制作"废话"黑名单"""# 常见的中文停用词chinese_stopwords = ['的', '是', '在', '了', '和', '有', '这', '那', '就', '都', '而', '及', '与', '或', '但', '却', '还', '也', '又', '再','我', '你', '他', '她', '它', '们', '我们', '你们', '他们','这个', '那个', '什么', '怎么', '为什么', '哪里', '哪些']# 常见的英文停用词english_stopwords = ['the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to','for', 'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be','been', 'being', 'have', 'has', 'had', 'do', 'does', 'did','will', 'would', 'should', 'could', 'can', 'may', 'might','i', 'you', 'he', 'she', 'it', 'we', 'they', 'this', 'that']return set(chinese_stopwords + english_stopwords)def remove_stopwords(words, stopwords_set):"""移除停用词就像从垃圾中挑选有用的东西"""return [word for word in words if word not in stopwords_set and len(word) > 1]# 测试
stopwords = create_stopwords_list()
test_words = ['我', '觉得', '这个', '电影', '很', '好看', '的', '剧情']
filtered_words = remove_stopwords(test_words, stopwords)print(f"原始词语：{test_words}")
print(f"过滤后：{filtered_words}")

🏷️ 词干提取与词形还原

1. 英文词干提取

# 需要安装：pip install nltk
import nltk
from nltk.stem import PorterStemmer, SnowballStemmer# 下载必要的数据
nltk.download('punkt')def stem_words(words, language='english'):"""词干提取：把单词还原成词根形式就像把所有的花朵都摘成花瓣"""if language == 'english':stemmer = PorterStemmer()else:stemmer = SnowballStemmer(language)return [stemmer.stem(word) for word in words]# 测试
english_words = ['running', 'runs', 'ran', 'easily', 'fairly', 'quickly']
stemmed_words = stem_words(english_words)print(f"原词：{english_words}")
print(f"词干：{stemmed_words}")

2. 英文词形还原

from nltk.stem import WordNetLemmatizer
from nltk.corpus import wordnetnltk.download('wordnet')
nltk.download('averaged_perceptron_tagger')def lemmatize_words(words):"""词形还原：把单词还原成词典形式比词干提取更准确，就像把各种变形的玩具还原成原始形状"""lemmatizer = WordNetLemmatizer()def get_wordnet_pos(word):"""获取词性标注"""tag = nltk.pos_tag([word])[0][1][0].upper()tag_dict = {"J": wordnet.ADJ,"N": wordnet.NOUN,"V": wordnet.VERB,"R": wordnet.ADV}return tag_dict.get(tag, wordnet.NOUN)return [lemmatizer.lemmatize(word, get_wordnet_pos(word)) for word in words]# 测试
test_words = ['running', 'runs', 'ran', 'better', 'worse', 'mice', 'children']
lemmatized = lemmatize_words(test_words)print(f"原词：{test_words}")
print(f"还原后：{lemmatized}")

💻 实战项目：构建文本预处理管道

现在让我们把所有的技术组合起来，构建一个完整的文本预处理管道：

class TextPreprocessor:"""文本预处理器：一站式文本清洗服务就像一个全自动洗衣机"""def __init__(self, language='chinese'):self.language = languageself.stopwords = self._load_stopwords()def _load_stopwords(self):"""加载停用词"""return create_stopwords_list()def clean_text(self, text):"""基础清洗"""text = basic_clean(text)text = handle_punctuation(text)text = normalize_case(text, 'lower')text = normalize_width(text)return textdef tokenize(self, text):"""分词"""if self.language == 'chinese':return chinese_tokenize(text)else:return text.split()def filter_words(self, words):"""过滤停用词和短词"""return remove_stopwords(words, self.stopwords)def preprocess(self, text):"""完整的预处理流程"""# 1. 清洗text = self.clean_text(text)# 2. 分词words = self.tokenize(text)# 3. 过滤words = self.filter_words(words)return wordsdef batch_preprocess(self, texts):"""批量处理"""return [self.preprocess(text) for text in texts]# 测试完整的预处理管道
preprocessor = TextPreprocessor(language='chinese')# 测试文本
test_texts = ["   这是一个测试文本！！！   自然语言处理很有趣。  ","机器学习和深度学习是人工智能的重要分支！","我觉得这个电影很好看，剧情很棒！"
]print("原始文本：")
for i, text in enumerate(test_texts):print(f"{i+1}. {repr(text)}")print("\n预处理后：")
processed_texts = preprocessor.batch_preprocess(test_texts)
for i, words in enumerate(processed_texts):print(f"{i+1}. {words}")

🚀 进阶技巧：正则表达式的威力

正则表达式就像是文本处理的"瑞士军刀"，能够精确地匹配和处理各种复杂的文本模式。

1. 常用正则表达式模式

import redef advanced_text_cleaning(text):"""使用正则表达式进行高级文本清洗就像用激光手术刀精确切除文本中的"肿瘤""""# 1. 清理电话号码text = re.sub(r'1[3-9]\d{9}', '<PHONE>', text)# 2. 清理身份证号text = re.sub(r'\d{17}[\dXx]', '<ID>', text)# 3. 清理QQ号text = re.sub(r'[1-9]\d{4,10}', '<QQ>', text)# 4. 清理时间表达式text = re.sub(r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日号]?', '<DATE>', text)text = re.sub(r'\d{1,2}[点时:]\d{1,2}[分]?', '<TIME>', text)# 5. 清理价格表达式text = re.sub(r'[￥$]\d+(\.\d+)?', '<PRICE>', text)# 6. 清理重复字符text = re.sub(r'(.)\1{2,}', r'\1\1', text)  # 连续3个以上相同字符变成2个return text# 测试
messy_text = """
联系电话：13812345678
身份证号：123456789012345678
QQ号：123456789
今天是2024年1月15日下午3点30分
价格：￥199.99
哈哈哈哈哈哈哈哈哈哈
"""cleaned = advanced_text_cleaning(messy_text)
print(f"原文：{messy_text}")
print(f"清洗后：{cleaned}")

2. 中文文本特有的处理

def chinese_text_processing(text):"""中文文本的特殊处理就像为中文量身定制的"洗衣程序""""# 1. 去除中文中的英文和数字（保留中文）def extract_chinese_only(text):return re.sub(r'[^\u4e00-\u9fff]', '', text)# 2. 提取中英文混合内容def extract_chinese_english(text):return re.sub(r'[^\u4e00-\u9fff\w\s]', '', text)# 3. 标准化中文标点符号def normalize_chinese_punctuation(text):# 英文标点转中文标点text = text.replace(',', '，')text = text.replace('.', '。')text = text.replace('?', '？')text = text.replace('!', '！')text = text.replace(':', '：')text = text.replace(';', '；')text = text.replace('(', '（')text = text.replace(')', '）')return text# 4. 处理中文数字def normalize_chinese_numbers(text):# 中文数字转阿拉伯数字chinese_num_map = {'零': '0', '一': '1', '二': '2', '三': '3', '四': '4','五': '5', '六': '6', '七': '7', '八': '8', '九': '9','十': '10', '百': '100', '千': '1000', '万': '10000'}for cn_num, ar_num in chinese_num_map.items():text = text.replace(cn_num, ar_num)return text# 应用所有处理result = {'original': text,'chinese_only': extract_chinese_only(text),'chinese_english': extract_chinese_english(text),'normalized_punctuation': normalize_chinese_punctuation(text),'normalized_numbers': normalize_chinese_numbers(text)}return result# 测试
chinese_text = "这是一个测试文本，包含English和数字123！还有一些标点符号."
results = chinese_text_processing(chinese_text)for key, value in results.items():print(f"{key}: {value}")

3. 文本质量评估

def text_quality_assessment(text):"""评估文本质量就像给洗好的衣服打分"""# 1. 长度检查length_score = min(len(text) / 100, 1.0) if len(text) > 0 else 0# 2. 中文字符比例chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))chinese_ratio = chinese_chars / len(text) if len(text) > 0 else 0# 3. 标点符号比例punctuation_chars = len(re.findall(r'[^\w\s]', text))punctuation_ratio = punctuation_chars / len(text) if len(text) > 0 else 0# 4. 数字比例digit_chars = len(re.findall(r'\d', text))digit_ratio = digit_chars / len(text) if len(text) > 0 else 0# 5. 特殊字符比例special_chars = len(re.findall(r'[^\u4e00-\u9fff\w\s]', text))special_ratio = special_chars / len(text) if len(text) > 0 else 0# 综合评分quality_score = (length_score * 0.2 +chinese_ratio * 0.3 +(1 - punctuation_ratio) * 0.2 +(1 - digit_ratio) * 0.15 +(1 - special_ratio) * 0.15)return {'length': len(text),'chinese_ratio': chinese_ratio,'punctuation_ratio': punctuation_ratio,'digit_ratio': digit_ratio,'special_ratio': special_ratio,'quality_score': quality_score}# 测试
test_texts = ["这是一个高质量的中文文本示例。","Text with 123 numbers and @@@ symbols!!!","短文本","这是一个包含适量标点符号、数字123和英文words的混合文本。"
]print("文本质量评估结果：")
for i, text in enumerate(test_texts):quality = text_quality_assessment(text)print(f"\n文本{i+1}: {text}")print(f"质量得分: {quality['quality_score']:.2f}")print(f"详细信息: {quality}")

🔧 常见问题与解决方案

1. 中文分词常见问题

问题1：专业术语分词不准确

# 问题示例
text = "深度学习和机器学习是人工智能的重要分支"
words = list(jieba.cut(text))
print(f"默认分词: {words}")
# 可能得到：['深度', '学习', '和', '机器', '学习', '是', '人工智能', '的', '重要', '分支']# 解决方案：使用自定义词典
jieba.add_word("深度学习")
jieba.add_word("机器学习")
words = list(jieba.cut(text))
print(f"优化后分词: {words}")
# 得到：['深度学习', '和', '机器学习', '是', '人工智能', '的', '重要', '分支']

问题2：新词识别困难

# 解决方案：动态调整词典
def update_dictionary_with_new_words(texts):"""根据文本内容动态更新词典就像根据新潮流更新衣服款式"""# 使用统计方法找出可能的新词from collections import defaultdict# 简单的双字词统计word_count = defaultdict(int)for text in texts:words = list(jieba.cut(text, cut_all=True))for word in words:if len(word) >= 2:word_count[word] += 1# 添加高频词到词典for word, count in word_count.items():if count >= 3:  # 出现3次以上的词jieba.add_word(word)return word_count# 测试
new_texts = ["元宇宙是未来的趋势","NFT和元宇宙密切相关","Web3.0将改变互联网"
]word_stats = update_dictionary_with_new_words(new_texts)
print(f"发现的新词: {dict(word_stats)}")

2. 停用词过滤问题

问题：停用词表不够全面或过于严格

def adaptive_stopwords_filter(texts, min_df=2, max_df_ratio=0.8):"""自适应停用词过滤根据文本集合动态调整停用词表"""from collections import Counter# 统计词频all_words = []for text in texts:words = list(jieba.cut(text))all_words.extend(words)word_freq = Counter(all_words)total_docs = len(texts)# 找出过于频繁或过于稀少的词adaptive_stopwords = set()for word, freq in word_freq.items():# 文档频率doc_freq = sum(1 for text in texts if word in text)df_ratio = doc_freq / total_docs# 过于频繁的词（在80%以上文档中出现）if df_ratio > max_df_ratio:adaptive_stopwords.add(word)# 过于稀少的词（总共只出现1次）if freq < min_df:adaptive_stopwords.add(word)return adaptive_stopwords# 测试
test_corpus = ["这是第一篇文档，讨论机器学习","这是第二篇文档，也讨论机器学习","这是第三篇文档，讨论深度学习","这是第四篇文档，讨论人工智能"
]adaptive_stops = adaptive_stopwords_filter(test_corpus)
print(f"自适应停用词: {adaptive_stops}")

3. 编码和特殊字符问题

问题：文本编码不一致

def handle_encoding_issues(text):"""处理编码问题就像给不同"口音"的文本找到统一的"标准普通话""""# 尝试检测和转换编码import chardetif isinstance(text, bytes):# 检测编码detected = chardet.detect(text)encoding = detected['encoding']# 转换为UTF-8try:text = text.decode(encoding)except:text = text.decode('utf-8', errors='ignore')# 处理特殊Unicode字符# 标准化Unicodeimport unicodedatatext = unicodedata.normalize('NFKC', text)# 移除控制字符text = ''.join(char for char in text if unicodedata.category(char)[0] != 'C')return text# 测试
problematic_text = "这是一个包含特殊字符\u200b\u200c的文本"
clean_text = handle_encoding_issues(problematic_text)
print(f"原文: {repr(problematic_text)}")
print(f"清洗后: {repr(clean_text)}")

4. 性能优化问题

问题：大量文本处理速度慢

def batch_processing_optimization(texts, batch_size=1000):"""批量处理优化就像洗衣店的"工业洗衣机"模式"""import multiprocessing as mpfrom functools import partialdef process_batch(batch_texts):"""处理一批文本"""preprocessor = TextPreprocessor()return preprocessor.batch_preprocess(batch_texts)# 分批处理batches = [texts[i:i+batch_size] for i in range(0, len(texts), batch_size)]# 多进程处理with mp.Pool(processes=mp.cpu_count()) as pool:results = pool.map(process_batch, batches)# 合并结果final_results = []for batch_result in results:final_results.extend(batch_result)return final_results# 使用示例
large_texts = ["这是测试文本" + str(i) for i in range(10000)]
print(f"处理{len(large_texts)}个文本...")
# processed_texts = batch_processing_optimization(large_texts)
print("批量处理完成！")

📊 实际应用案例

1. 情感分析预处理

def sentiment_analysis_preprocessing(text):"""情感分析专用的文本预处理重点保留情感相关的词汇和表达"""# 1. 基础清洗但保留感叹号和问号text = re.sub(r'<[^>]+>', '', text)text = re.sub(r'http\S+', '', text)text = re.sub(r'\s+', ' ', text)# 2. 保留情感符号emotion_symbols = ['😊', '😢', '😡', '❤️', '👍', '👎', '😍', '😭']for symbol in emotion_symbols:text = text.replace(symbol, f' {symbol} ')# 3. 处理否定词（重要！）negation_words = ['不', '没', '无', '非', '未', '别', '勿']words = list(jieba.cut(text))# 给否定词后的词添加标记processed_words = []negation_flag = Falsefor word in words:if word in negation_words:negation_flag = Trueprocessed_words.append(word)elif negation_flag and word not in ['的', '了', '是', '在']:processed_words.append(f"NOT_{word}")negation_flag = Falseelse:processed_words.append(word)negation_flag = Falsereturn processed_words# 测试
sentiment_text = "这个电影不好看，我不喜欢这个剧情😢"
processed = sentiment_analysis_preprocessing(sentiment_text)
print(f"情感分析预处理结果: {processed}")

2. 搜索引擎预处理

def search_engine_preprocessing(query, document):"""搜索引擎专用预处理重点是提高匹配度和相关性"""def preprocess_for_search(text):# 1. 标准化text = text.lower()text = normalize_width(text)# 2. 同义词替换synonyms = {'手机': ['手机', '电话', '移动电话', '智能机'],'电脑': ['电脑', '计算机', '电子计算机', 'pc'],'好': ['好', '棒', '优秀', '不错', '很好']}words = list(jieba.cut(text))expanded_words = []for word in words:expanded_words.append(word)for key, values in synonyms.items():if word in values:expanded_words.extend(values)return list(set(expanded_words))  # 去重query_words = preprocess_for_search(query)doc_words = preprocess_for_search(document)# 计算匹配度common_words = set(query_words) & set(doc_words)match_score = len(common_words) / len(set(query_words)) if query_words else 0return {'query_words': query_words,'doc_words': doc_words,'common_words': list(common_words),'match_score': match_score}# 测试
query = "苹果手机好不好"
document = "这款苹果智能机性能很棒，非常优秀"
result = search_engine_preprocessing(query, document)
print(f"搜索匹配结果: {result}")

💡 最佳实践建议

1. 预处理流程设计原则

# 好的预处理流程设计
class BestPracticePreprocessor:"""最佳实践的文本预处理器遵循"可配置、可扩展、可监控"的原则"""def __init__(self, config):self.config = configself.stats = {'total_processed': 0,'average_length': 0,'error_count': 0}def preprocess_with_monitoring(self, text):"""带监控的预处理就像洗衣机的"智能监控"功能"""try:original_length = len(text)# 预处理步骤if self.config.get('clean_html', True):text = self._clean_html(text)if self.config.get('normalize_case', True):text = text.lower()if self.config.get('tokenize', True):text = self._tokenize(text)# 更新统计self.stats['total_processed'] += 1self.stats['average_length'] = (self.stats['average_length'] + original_length) / self.stats['total_processed']return textexcept Exception as e:self.stats['error_count'] += 1print(f"预处理错误: {e}")return text  # 返回原文本def _clean_html(self, text):return re.sub(r'<[^>]+>', '', text)def _tokenize(self, text):return list(jieba.cut(text))def get_stats(self):return self.stats# 使用示例
config = {'clean_html': True,'normalize_case': True,'tokenize': True
}preprocessor = BestPracticePreprocessor(config)
result = preprocessor.preprocess_with_monitoring("这是一个<b>测试</b>文本")
print(f"预处理结果: {result}")
print(f"处理统计: {preprocessor.get_stats()}")

2. 错误处理和容错机制

def robust_text_preprocessing(text):"""健壮的文本预处理就像给洗衣机装上"防护罩""""if not text or not isinstance(text, str):return []try:# 步骤1: 基础清洗text = basic_clean(text)# 步骤2: 分词（带容错）try:words = list(jieba.cut(text))except Exception as e:print(f"分词错误: {e}")words = text.split()  # 降级为简单分词# 步骤3: 过滤（带容错）try:words = [word for word in words if len(word) > 0]except Exception as e:print(f"过滤错误: {e}")words = [text]  # 降级为原文本return wordsexcept Exception as e:print(f"整体预处理错误: {e}")return [text] if text else []# 测试容错机制
test_cases = ["正常文本","",None,123,"包含特殊字符的文本\x00\x01","非常长的文本" * 1000
]for test_text in test_cases:result = robust_text_preprocessing(test_text)print(f"输入: {repr(test_text)[:50]}... -> 输出: {len(result)}个词")