import re
import hashlib
class BasicChineseTextCleaner:
    def __init__(self):
        # 特殊符号正则（匹配HTML标签、特殊转义符、冗余空格）
        self.special_char_pattern = re.compile(r'<.*?>|&[a-z]+;|\s+')
        # 中英文标点映射（统一为中文标点）
        self.punct_map = {
            '.': '。', ',': '，', '?': '？', '!': '！',
            ':': '：', ';': '；', '(': '（', ')': '）'
        }
    def clean_special_chars(self, text):
        """去除特殊符号和冗余空格"""
        cleaned = self.special_char_pattern.sub(' ', text)
        return cleaned.strip()
    def unify_punctuation(self, text):
        """统一中英文标点"""
        for en_punc, cn_punc in self.punct_map.items():
            text = text.replace(en_punc, cn_punc)
        return text
    def filter_short_text(self, text, min_len=10):
        """过滤过短文本（默认至少10个字符）"""
        return text if len(text) >= min_len else None
    def get_text_hash(self, text):
        """生成文本哈希值（用于完全去重）"""
        return hashlib.md5(text.encode('utf-8')).hexdigest()
    def deduplicate_texts(self, texts):
        """基于哈希去重"""
        hash_set = set()
        unique_texts = []
        for text in texts:
            if text is None:
                continue
            text_hash = self.get_text_hash(text)
            if text_hash not in hash_set:
                hash_set.add(text_hash)
                unique_texts.append(text)
        return unique_texts
    def clean(self, raw_texts):
        """全流程基础清洗"""
        cleaned_texts = []
        for text in raw_texts:
            # 步骤1：去特殊符号
            text = self.clean_special_chars(text)
            if not text:
                continue
            # 步骤2：统一标点
            text = self.unify_punctuation(text)
            # 步骤3：过滤短文本
            text = self.filter_short_text(text)
            if text:
                cleaned_texts.append(text)
        # 步骤4：去重
        cleaned_texts = self.deduplicate_texts(cleaned_texts)
        return cleaned_texts
# 测试示例
if __name__ == "__main__":
    # 原始噪声语料
    raw_texts = [
        "  <p>今天吃了苹果，味道超棒！</p>&nbsp;",  # 含HTML标签+转义符
        "今天吃了苹果.味道超棒!",  # 英文标点
        "好吃",  # 过短文本
        "  <p>今天吃了苹果，味道超棒！</p>&nbsp;",  # 重复文本
        "我今天买了华为手机，性能很好～"
    ]
    
    cleaner = BasicChineseTextCleaner()
    result = cleaner.clean(raw_texts)
    
    print("清洗后结果：")
    for i, text in enumerate(result):
        print(f"{i+1}. {text}")

import torch
import torch.nn as nn
from transformers import BertTokenizer, BertModel
from torch.utils.data import Dataset, DataLoader
from modelscope.hub.snapshot_download import snapshot_download
# 1. 定义数据集（模拟人工标注的质量语料）
class TextQualityDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_len=128):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_len = max_len
    def __len__(self):
        return len(self.texts)
    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]
        
        # 中文BERT分词
        encoding = self.tokenizer(
            text,
            truncation=True,
            padding='max_length',
            max_length=self.max_len,
            return_tensors='pt'
        )
        
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }
# 2. 定义质量评分模型（基于BERT微调）
class TextQualityClassifier(nn.Module):
    def __init__(self, bert_model_name='bert-base-chinese', num_classes=2):
        super().__init__()
        cache_dir = "D:\\modelscope\\hub"
        model_name = "google-bert/bert-base-chinese"
        local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
        self.bert = BertModel.from_pretrained(local_model_path)
        self.dropout = nn.Dropout(0.3)
        self.fc = nn.Linear(self.bert.config.hidden_size, num_classes)
    def forward(self, input_ids, attention_mask):
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        pooled_output = outputs.pooler_output  # [CLS] token输出
        x = self.dropout(pooled_output)
        logits = self.fc(x)
        return logits
# 3. 训练+预测流程
def train_and_predict():
    # 初始化工具
    cache_dir = "D:\\modelscope\\hub"
    model_name = "google-bert/bert-base-chinese"
    local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
    tokenizer = BertTokenizer.from_pretrained(local_model_path)
    model = TextQualityClassifier(bert_model_name=local_model_path)
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    
    # 模拟标注数据（1=高质量，0=低质量）
    train_texts = [
        "人工智能技术的发展推动了智能制造的升级，相关算法在工业场景中落地效果显著。",  # 高质量
        "啊啊啊这啥呀不知道说啥随便写写",  # 低质量
        "2025年中国数字经济规模预计突破80万亿元，同比增长9.5%。",  # 高质量
        "凑字数凑字数凑字数，没啥内容的垃圾文本",  # 低质量
        "唐诗宋词是中国传统文化的瑰宝，其中蕴含的美学价值至今仍被广泛研究。"  # 高质量
    ]
    train_labels = [1, 0, 1, 0, 1]
    
    # 构建数据集
    dataset = TextQualityDataset(train_texts, train_labels, tokenizer)
    dataloader = DataLoader(dataset, batch_size=2, shuffle=True)
    
    # 训练配置
    optimizer = torch.optim.Adam(model.parameters(), lr=2e-5)
    loss_fn = nn.CrossEntropyLoss()
    model.train()
    
    # 简易训练（5轮）
    for epoch in range(5):
        total_loss = 0
        for batch in dataloader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)
            
            outputs = model(input_ids, attention_mask)
            loss = loss_fn(outputs, labels)
            
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        print(f"Epoch {epoch+1}, Loss: {total_loss/len(dataloader):.4f}")
    
    # 测试新语料质量评分
    test_texts = [
        "大模型训练数据治理需要兼顾噪声过滤和语义一致性。",
        "随便打几个字，这是低质量文本"
    ]
    model.eval()
    with torch.no_grad():
        for text in test_texts:
            encoding = tokenizer(
                text, truncation=True, padding='max_length', 
                max_length=128, return_tensors='pt'
            )
            input_ids = encoding['input_ids'].to(device)
            attention_mask = encoding['attention_mask'].to(device)
            
            outputs = model(input_ids, attention_mask)
            prob = torch.softmax(outputs, dim=1)
            quality_score = prob[0][1].item()  # 高质量概率
            print(f"文本：{text}\n质量评分（0-1）：{quality_score:.4f}\n")
if __name__ == "__main__":
    train_and_predict()

import torch
from transformers import AutoModel, AutoTokenizer
import jieba
from datetime import datetime
from modelscope.hub.snapshot_download import snapshot_download
class AdvancedChineseTextGovernor:
    def __init__(self):
        # 加载中文词义消歧模型（基于SimBERT）
        cache_dir = "D:\\modelscope\\hub"
        model_name = "Jerry0/text2vec-base-chinese"
        local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
        self.tokenizer = AutoTokenizer.from_pretrained(local_model_path)
        self.semantic_model = AutoModel.from_pretrained(local_model_path)
        # 中文多义词典（示例：苹果、银行）
        self.polysemous_words = {
            "苹果": ["水果", "科技公司"],
            "银行": ["金融机构", "河边的堤岸"]
        }
        # 时效性关键词库（示例）
        self.timely_keywords = ["2025政策", "2025技术", "2025经济"]
    def get_text_embedding(self, text):
        """获取文本语义向量（用于词义消歧）"""
        inputs = self.tokenizer(
            text, return_tensors='pt', padding=True, 
            truncation=True, max_length=128
        )
        with torch.no_grad():
            outputs = self.semantic_model(** inputs)
        # 均值池化获取句向量
        emb = outputs.last_hidden_state.mean(dim=1)
        return emb
    def disambiguate_polysemous_word(self, text, target_word):
        """多义词义消歧"""
        if target_word not in self.polysemous_words:
            return "未知语义"
        
        # 生成目标词不同语义的参考向量
        ref_embeddings = {}
        for meaning in self.polysemous_words[target_word]:
            ref_text = f"{target_word} {meaning}"
            ref_embeddings[meaning] = self.get_text_embedding(ref_text)
        
        # 计算文本与各语义的相似度
        text_emb = self.get_text_embedding(text)
        max_sim = 0
        best_meaning = ""
        for meaning, ref_emb in ref_embeddings.items():
            sim = torch.cosine_similarity(text_emb, ref_emb).item()
            if sim > max_sim:
                max_sim = sim
                best_meaning = meaning
        
        return best_meaning
    def filter_timely_text(self, texts_with_timestamp, cutoff_date="2024-01-01"):
        """过滤过时文本（保留2024年后的时效性内容）"""
        cutoff = datetime.strptime(cutoff_date, "%Y-%m-%d")
        timely_texts = []
        for text, timestamp in texts_with_timestamp:
            text_date = datetime.strptime(timestamp, "%Y-%m-%d")
            # 同时筛选含时效性关键词的文本
            if text_date >= cutoff and any(kw in text for kw in self.timely_keywords):
                timely_texts.append(text)
        return timely_texts
# 测试示例
if __name__ == "__main__":
    governor = AdvancedChineseTextGovernor()
    
    # 1. 词义消歧测试
    test_texts = [
        "苹果发布了新款iPhone 16",
        "我买了一斤苹果，很甜"
    ]
    print("=== 词义消歧结果 ===")
    for text in test_texts:
        meaning = governor.disambiguate_polysemous_word(text, "苹果")
        print(f"文本：{text}\n'苹果'语义：{meaning}\n")
    
    # 2. 时效性过滤测试
    texts_with_ts = [
        ("2025政策：大模型数据治理新规发布", "2025-03-15"),
        ("2023年的旧政策：人工智能发展规划", "2023-05-20"),
        ("2025技术：中文大模型语义一致性治理突破", "2025-01-10")
    ]
    timely_texts = governor.filter_timely_text(texts_with_ts)
    print("=== 时效性过滤结果 ===")
    for text in timely_texts:
        print(text)