import re import string class TextCleaner: def __init__(self): self.patterns = { 'url': r'https?://\S+|www\.\S+', 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'html': r'<.*?>', 'special_chars': r'[^a-zA-Z0-9\s]', 'extra_spaces': r'\s+' } def clean(self, text): text = text.lower() text = re.sub(self.patterns['url'], '', text) text = re.sub(self.patterns['email'], '', text) text = re.sub(self.patterns['html'], '', text) text = re.sub(self.patterns['special_chars'], '', text) text = re.sub(self.patterns['extra_spaces'], ' ', text) return text.strip() class ChineseTextCleaner: def __init__(self): self.patterns = { 'url': r'https?://\S+|www\.\S+', 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'html': r'<.*?>', 'punctuation': r'[，。！？、；：""''（）【】《》]', 'extra_spaces': r'\s+' } def clean(self, text): text = re.sub(self.patterns['url'], '', text) text = re.sub(self.patterns['email'], '', text) text = re.sub(self.patterns['html'], '', text) text = re.sub(self.patterns['punctuation'], '', text) text = re.sub(self.patterns['extra_spaces'], '', text) return text.strip()
2.2 分词处理
import jieba class Tokenizer: def __init__(self, language='english'): self.language = language if language == 'chinese': self.tokenizer = jieba elif language == 'english': from nltk.tokenize import word_tokenize self.tokenizer = word_tokenize def tokenize(self, text): if self.language == 'chinese': return self.tokenizer.lcut(text) else: return self.tokenizer(text) def tokenize_batch(self, texts): return [self.tokenize(text) for text in texts] class StopwordFilter: def __init__(self, language='english'): if language == 'english': from nltk.corpus import stopwords self.stopwords = set(stopwords.words('english')) elif language == 'chinese': self.stopwords = self._load_chinese_stopwords() def _load_chinese_stopwords(self): stopwords = set() common_stopwords = [ '的', '是', '在', '和', '有', '我', '他', '她', '它', '这', '那', '个', '都', '就', '也', '很', '到', '说', '要', '去', '会', '着', '没有', '看', '好', '自己', '又' ] stopwords.update(common_stopwords) return stopwords def filter(self, tokens): return [token for token in tokens if token not in self.stopwords] def filter_batch(self, tokenized_texts): return [self.filter(tokens) for tokens in tokenized_texts]
2.3 向量化
import torch import torch.nn as nn from sklearn.feature_extraction.text import TfidfVectorizer class TFIDFVectorizer: def __init__(self, max_features=5000): self.vectorizer = TfidfVectorizer(max_features=max_features) def fit(self, texts): self.vectorizer.fit(texts) def transform(self, texts): return torch.tensor(self.vectorizer.transform(texts).toarray(), dtype=torch.float32) def fit_transform(self, texts): return torch.tensor(self.vectorizer.fit_transform(texts).toarray(), dtype=torch.float32) class WordEmbeddingVectorizer: def __init__(self, embedding_dim=100): self.embedding_dim = embedding_dim self.word_to_idx = {} self.embedding = None def fit(self, tokenized_texts): vocab = set() for tokens in tokenized_texts: vocab.update(tokens) self.word_to_idx = {word: i + 1 for i, word in enumerate(vocab)} self.word_to_idx['<UNK>'] = 0 self.embedding = nn.Embedding(len(self.word_to_idx), self.embedding_dim) def transform(self, tokenized_texts, max_len=50): sequences = [] for tokens in tokenized_texts: sequence = [] for token in tokens[:max_len]: sequence.append(self.word_to_idx.get(token, 0)) sequence += [0] * (max_len - len(sequence)) sequences.append(sequence) return torch.tensor(sequences, dtype=torch.long) class BERTVectorizer: def __init__(self, model_name='bert-base-uncased'): from transformers import BertModel, BertTokenizer self.model = BertModel.from_pretrained(model_name) self.tokenizer = BertTokenizer.from_pretrained(model_name) def encode(self, texts): inputs = self.tokenizer( texts, padding=True, truncation=True, max_length=512, return_tensors='pt' ) with torch.no_grad(): outputs = self.model(**inputs) return outputs.last_hidden_state[:, 0, :]
2.4 特征工程
class TextFeatureExtractor: def __init__(self): self.features = [] def add_length_feature(self, texts): lengths = [len(text) for text in texts] self.features.append(torch.tensor(lengths, dtype=torch.float32).unsqueeze(1)) def add_word_count_feature(self, tokenized_texts): word_counts = [len(tokens) for tokens in tokenized_texts] self.features.append(torch.tensor(word_counts, dtype=torch.float32).unsqueeze(1)) def add_punctuation_feature(self, texts): punctuation_ratios = [] for text in texts: punctuation_count = sum(1 for char in text if char in string.punctuation) ratio = punctuation_count / len(text) if len(text) > 0 else 0 punctuation_ratios.append(ratio) self.features.append(torch.tensor(punctuation_ratios, dtype=torch.float32).unsqueeze(1)) def get_features(self): if not self.features: return None return torch.cat(self.features, dim=1) class FeaturePipeline: def __init__(self, steps): self.steps = steps def fit_transform(self, texts): features = texts for step in self.steps: features = step.fit_transform(features) return features def transform(self, texts): features = texts for step in self.steps: features = step.transform(features) return features
class NLPPreprocessor: def __init__(self, tokenizer, vectorizer, cleaner=None): self.tokenizer = tokenizer self.vectorizer = vectorizer self.cleaner = cleaner def process(self, texts): if self.cleaner: texts = [self.cleaner.clean(text) for text in texts] tokenized = self.tokenizer.tokenize_batch(texts) if hasattr(self.vectorizer, 'fit_transform'): features = self.vectorizer.fit_transform(tokenized) else: features = self.vectorizer.encode(texts) return features def transform(self, texts): if self.cleaner: texts = [self.cleaner.clean(text) for text in texts] tokenized = self.tokenizer.tokenize_batch(texts) if hasattr(self.vectorizer, 'transform'): features = self.vectorizer.transform(tokenized) else: features = self.vectorizer.encode(texts) return features
