17_文本预处理全流程：分词到lemmatization

原始文本 → 文本清洗 → 分词(Tokenization) → 去除停用词 → 标准化(词干提取/词形还原) → 特征提取

import re

def remove_html_tags(text):
    # 使用正则表达式移除HTML标签
    clean = re.compile('<.*?>')
    return re.sub(clean, '', text)

# 2025年spaCy新增的HTML处理功能
import spacy
nlp = spacy.load("en_core_web_sm")
doc = nlp.make_doc("<p>This is a <strong>test</strong> paragraph.</p>")
clean_text = "".join([token.text for token in doc if not token.is_space and not re.match(r'^<.*>$', token.text)])

def remove_special_characters(text):
    # 保留字母、数字和基本标点
    pattern = r'[^a-zA-Z0-9\s.!?,]'
    return re.sub(pattern, '', text)

def normalize_whitespace(text):
    # 将多个空格替换为单个空格
    return re.sub('\s+', ' ', text).strip()

def to_lowercase(text):
    return text.lower()

# NLTK基于规则的分词示例
import nltk
tokens = nltk.word_tokenize("Hello world! This is a test.")
print(tokens)  # ['Hello', 'world', '!', 'This', 'is', 'a', 'test', '.']

# 简单的基于频率的分词逻辑示例
from collections import Counter
def simple_statistical_tokenization(text, vocab_freq):
    # 简化版，实际实现更复杂
    tokens = []
    i = 0
    while i < len(text):
        # 尝试最长匹配
        for j in range(min(i+10, len(text)), i, -1):
            if text[i:j] in vocab_freq and vocab_freq[text[i:j]] > 0:
                tokens.append(text[i:j])
                i = j
                break
        else:
            # 未找到匹配，按单个字符切分
            tokens.append(text[i])
            i += 1
    return tokens

# 使用spaCy的神经网络分词器
import spacy
nlp = spacy.load("zh_core_web_sm")  # 中文模型
doc = nlp("我爱自然语言处理技术")
tokens = [token.text for token in doc]
print(tokens)  # ['我', '爱', '自然语言处理', '技术']

# NLTK停用词处理
from nltk.corpus import stopwords
import nltk

# 下载停用词库（如果尚未下载）
nltk.download('stopwords')

# 获取英语停用词表
english_stopwords = set(stopwords.words('english'))

def remove_stopwords(tokens):
    return [token for token in tokens if token.lower() not in english_stopwords]

# spaCy停用词处理
import spacy
nlp = spacy.load("en_core_web_sm")
doc = nlp("This is a sample text with some common words.")
tokens_no_stopwords = [token.text for token in doc if not token.is_stop]

# NLTK词干提取示例
from nltk.stem import PorterStemmer, LancasterStemmer, SnowballStemmer

# Porter词干提取器（最常用）
porter = PorterStemmer()
print(porter.stem("running"))  # "run"
print(porter.stem("runs"))     # "run"
print(porter.stem("ran"))      # "ran" （注意：Porter无法处理不规则变化）

# Lancaster词干提取器（更激进）
lancaster = LancasterStemmer()
print(lancaster.stem("running"))  # "run"
print(lancaster.stem("runs"))     # "run"
print(lancaster.stem("ran"))      # "ran"

# Snowball词干提取器（多语言支持）
snowball = SnowballStemmer("english")
print(snowball.stem("running"))  # "run"

# NLTK词形还原示例
from nltk.stem import WordNetLemmatizer
from nltk import pos_tag, word_tokenize
nltk.download('wordnet')
nltk.download('averaged_perceptron_tagger')

lemmatizer = WordNetLemmatizer()

# 基本词形还原（默认按名词处理）
print(lemmatizer.lemmatize("running"))  # "running" （默认按名词处理）
print(lemmatizer.lemmatize("running", pos='v'))  # "run" （指定动词）

# 带词性标注的词形还原
def lemmatize_with_pos(text):
    tagged_words = pos_tag(word_tokenize(text))
    # 简化的POS标签映射
    pos_map = {'JJ':'a', 'JJR':'a', 'JJS':'a',
               'VB':'v', 'VBD':'v', 'VBG':'v', 'VBN':'v', 'VBP':'v', 'VBZ':'v',
               'NN':'n', 'NNS':'n', 'NNP':'n', 'NNPS':'n',
               'RB':'r', 'RBR':'r', 'RBS':'r'}
    lemmatized_words = []
    for word, tag in tagged_words:
        pos = pos_map.get(tag[:2], 'n')  # 默认按名词处理
        lemmatized_words.append(lemmatizer.lemmatize(word.lower(), pos=pos))
    return lemmatized_words

# spaCy词形还原示例
import spacy
nlp = spacy.load("en_core_web_sm")
doc = nlp("I am running in the park and have run there before.")
for token in doc:
    print(f"{token.text} → {token.lemma_}")

# 使用Hugging Face的跨语言分词器
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("facebook/mbart-large-50")

# 处理多语言文本
texts = ["Hello world!", "你好世界！", "Bonjour monde!", "Привет мир!"]
for text in texts:
    tokens = tokenizer.tokenize(text)
    print(f"{text} → {tokens}")

# 文本规范化示例
import re

def normalize_text(text):
    # 小写化
    text = text.lower()
    # 标准化数字（简单示例）
    text = re.sub(r'\b(\d+)\s+(\d+)\b', r'\1\2', text)  # 合并数字间空格
    # 标准化日期格式（简单示例）
    text = re.sub(r'(\d{1,2})/(\d{1,2})/(\d{2,4})', r'\3-\2-\1', text)  # MM/DD/YYYY → YYYY-MM-DD
    return text

# 使用TextBlob进行拼写校正（简单示例）
from textblob import TextBlob

def correct_spelling(text):
    blob = TextBlob(text)
    return str(blob.correct())

# 2025年更先进的拼写校正方法通常结合BERT或GPT等预训练模型

# 使用SimHash进行文本去重
from datasketch import MinHash, MinHashLSH

def deduplicate_texts(texts, threshold=0.8):
    # 创建LSH索引
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    minhashes = {}
    unique_texts = []
    
    for i, text in enumerate(texts):
        # 创建文本的MinHash
        m = MinHash(num_perm=128)
        for word in text.split():
            m.update(word.encode('utf-8'))
        
        # 检查是否重复
        similar = lsh.query(m)
        if not similar:
            lsh.insert(f"text_{i}", m)
            minhashes[f"text_{i}"] = m
            unique_texts.append(text)
    
    return unique_texts

# 使用spaCy进行实体识别与替换
import spacy
nlp = spacy.load("en_core_web_sm")

def anonymize_text(text):
    doc = nlp(text)
    result = text
    # 从后向前替换，避免索引偏移
    for ent in reversed(doc.ents):
        if ent.label_ in ['PERSON', 'EMAIL', 'PHONE']:
            result = result[:ent.start_char] + f"[{ent.label_}]" + result[ent.end_char:]
    return result

import re
import spacy
from nltk.corpus import stopwords
import nltk

# 下载必要的资源
nltk.download('stopwords')

# 加载spaCy模型
nlp = spacy.load("en_core_web_sm")

# 构建预处理流水线
def text_preprocessing_pipeline(text):
    # 1. 文本清洗
    # 移除HTML标签
    text = re.sub(r'<.*?>', '', text)
    # 移除非字母数字字符（保留基本标点）
    text = re.sub(r'[^a-zA-Z0-9\s.!?,]', '', text)
    # 标准化空白字符
    text = re.sub(r'\s+', ' ', text).strip()
    
    # 2. 使用spaCy进行高级处理
    doc = nlp(text)
    
    # 3. 分词并进行后续处理
    tokens = []
    for token in doc:
        # 跳过停用词
        if token.is_stop:
            continue
        # 跳过纯数字
        if token.is_digit:
            continue
        # 跳过过长或过短的token
        if len(token.text) < 2 or len(token.text) > 20:
            continue
        # 添加词形还原后的token
        tokens.append(token.lemma_.lower())
    
    # 4. 重新组合为文本
    processed_text = ' '.join(tokens)
    
    return processed_text, tokens

# 测试流水线
text = "This is a <strong>sample</strong> text with some 123 numbers and common words! It's testing the preprocessing pipeline."
processed_text, tokens = text_preprocessing_pipeline(text)
print("原始文本:", text)
print("处理后文本:", processed_text)
print("处理后词元:", tokens)

import re
import spacy

# 加载中文spaCy模型
nlp = spacy.load("zh_core_web_sm")

# 中文停用词列表（简化版）
chinese_stopwords = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'}

def chinese_preprocessing_pipeline(text):
    # 1. 文本清洗
    # 移除HTML标签
    text = re.sub(r'<.*?>', '', text)
    # 移除非中文字符（保留基本标点）
    text = re.sub(r'[^\u4e00-\u9fa5\s.!?,]', '', text)
    # 标准化空白字符
    text = re.sub(r'\s+', ' ', text).strip()
    
    # 2. 使用spaCy进行中文处理
    doc = nlp(text)
    
    # 3. 分词并进行后续处理
    tokens = []
    for token in doc:
        # 跳过停用词
        if token.text in chinese_stopwords:
            continue
        # 跳过纯数字
        if token.is_digit:
            continue
        # 跳过过长或过短的token
        if len(token.text) < 1 or len(token.text) > 10:
            continue
        # 添加token（中文通常不需要词形还原）
        tokens.append(token.text)
    
    # 4. 重新组合为文本
    processed_text = ' '.join(tokens)
    
    return processed_text, tokens

# 测试中文流水线
chinese_text = "这是一个<strong>中文</strong>示例文本，包含一些数字123和常用词语！测试中文预处理流水线。"
processed_text, tokens = chinese_preprocessing_pipeline(chinese_text)
print("原始中文文本:", chinese_text)
print("处理后中文文本:", processed_text)
print("处理后中文词元:", tokens)

import re
import spacy
from langdetect import detect

# 加载多种语言模型（按需加载）
language_models = {
    'en': spacy.load("en_core_web_sm"),
    'zh': spacy.load("zh_core_web_sm"),
    'fr': spacy.load("fr_core_news_sm"),
    'de': spacy.load("de_core_news_sm")
}

# 简单的停用词字典
stopwords_dict = {
    'en': {'the', 'is', 'and', 'to', 'of', 'a', 'in', 'that', 'have', 'I'},
    'zh': {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人'}
    # 可以添加更多语言的停用词
}

def multilingual_preprocessing_pipeline(text):
    try:
        # 检测语言
        lang = detect(text)
        print(f"检测到语言: {lang}")
        
        # 使用相应的语言模型
        if lang in language_models:
            nlp = language_models[lang]
        else:
            # 使用英语作为默认模型
            nlp = language_models['en']
            print(f"未支持的语言 {lang}，使用英语模型作为默认值")
        
        # 文本清洗（通用部分）
        text = re.sub(r'<.*?>', '', text)  # 移除HTML标签
        text = re.sub(r'\s+', ' ', text).strip()  # 标准化空白字符
        
        # 使用spaCy处理
        doc = nlp(text)
        
        # 分词并处理
        tokens = []
        for token in doc:
            # 跳过停用词（如果语言有停用词表）
            if lang in stopwords_dict and token.text.lower() in stopwords_dict[lang]:
                continue
            # 跳过纯数字
            if token.is_digit:
                continue
            # 跳过过长token
            if len(token.text) > 20:
                continue
            # 根据语言选择处理方式
            if lang == 'zh':
                # 中文通常不需要词形还原
                tokens.append(token.text)
            else:
                # 其他语言进行词形还原
                tokens.append(token.lemma_.lower())
        
        return ' '.join(tokens), tokens, lang
    except Exception as e:
        print(f"预处理错误: {e}")
        return text, text.split(), 'unknown'

# 测试多语言流水线
multilingual_texts = [
    "This is an English sample text.",
    "这是一个中文示例文本。",
    "Ceci est un exemple de texte français.",
    "Dies ist ein Beispieltext auf Deutsch."
]

for i, text in enumerate(multilingual_texts):
    print(f"\n文本 {i+1}:")
    processed_text, tokens, lang = multilingual_preprocessing_pipeline(text)
    print(f"处理后: {processed_text}")
    print(f"词元: {tokens}")

# 批量预处理示例
import spacy
from concurrent.futures import ProcessPoolExecutor
import time

# 加载模型
nlp = spacy.load("en_core_web_sm")

# 单个文本处理函数
def process_single_text(text):
    doc = nlp(text)
    return [token.lemma_.lower() for token in doc if not token.is_stop and not token.is_punct]

# 批量处理函数
def batch_process(texts, batch_size=100):
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        # 使用spaCy的pipe方法进行批量处理
        docs = list(nlp.pipe(batch))
        for doc in docs:
            results.append([token.lemma_.lower() for token in doc if not token.is_stop and not token.is_punct])
    return results

# 并行处理函数
def parallel_process(texts, max_workers=4):
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(process_single_text, texts))
    return results

# 性能比较
def compare_performance(texts):
    # 单文本处理
    start_time = time.time()
    single_results = [process_single_text(text) for text in texts]
    single_time = time.time() - start_time
    print(f"单文本处理时间: {single_time:.4f}秒")
    
    # 批量处理
    start_time = time.time()
    batch_results = batch_process(texts)
    batch_time = time.time() - start_time
    print(f"批量处理时间: {batch_time:.4f}秒")
    print(f"批量处理速度提升: {single_time/batch_time:.2f}倍")
    
    # 并行处理（注意：在小数据集上可能看不到优势）
    if len(texts) > 100:  # 只在大数据集上测试并行
        start_time = time.time()
        parallel_results = parallel_process(texts)
        parallel_time = time.time() - start_time
        print(f"并行处理时间: {parallel_time:.4f}秒")
        print(f"并行处理速度提升: {single_time/parallel_time:.2f}倍")

# 测试性能
# 创建测试数据
test_texts = ["This is a sample text for performance testing." * 10 for _ in range(500)]
compare_performance(test_texts)

# 简单的预处理质量评估函数
def evaluate_preprocessing_quality(original_texts, processed_texts):
    results = {
        'text_count': len(original_texts),
        'avg_length_ratio': 0,
        'avg_token_count': 0,
        'empty_text_count': 0
    }
    
    token_counts = []
    for i, (original, processed) in enumerate(zip(original_texts, processed_texts)):
        # 计算长度比率
        if len(original) > 0:
            length_ratio = len(processed) / len(original)
            results['avg_length_ratio'] += length_ratio
        
        # 计算token数量
        tokens = processed.split()
        token_counts.append(len(tokens))
        
        # 检查空文本
        if len(processed.strip()) == 0:
            results['empty_text_count'] += 1
    
    # 计算平均值
    if results['text_count'] > 0:
        results['avg_length_ratio'] /= results['text_count']
        results['avg_token_count'] = sum(token_counts) / len(token_counts)
    
    # 打印评估结果
    print(f"预处理质量评估:")
    print(f"文本总数: {results['text_count']}")
    print(f"平均长度比率: {results['avg_length_ratio']:.2f}")
    print(f"平均token数量: {results['avg_token_count']:.2f}")
    print(f"空文本数量: {results['empty_text_count']} ({results['empty_text_count']/results['text_count']*100:.2f}%)")
    
    return results