文本分类：情感分析与主题分类

# 文本分类任务形式化定义
from typing import List, Dict, Tuple

class TextClassificationTask:
    """文本分类任务的抽象定义"""

    def __init__(self, name: str, labels: List[str], task_type: str):
        self.name = name
        self.labels = labels
        self.task_type = task_type  # "binary" | "multiclass" | "multilabel"
        self.label2id = {label: i for i, label in enumerate(labels)}
        self.id2label = {i: label for label, i in self.label2id.items()}

    @property
    def num_classes(self) -> int:
        return len(self.labels)

    def encode_labels(self, raw_labels: List[str]) -> List[int]:
        return [self.label2id[l] for l in raw_labels]

    def decode_labels(self, ids: List[int]) -> List[str]:
        return [self.id2label[i] for i in ids]

# 定义一个情感分析任务
sentiment_task = TextClassificationTask(
    name="情感分析",
    labels=["正面", "负面", "中性"],
    task_type="multiclass"
)
print(f"任务: {sentiment_task.name}")
print(f"类别数: {sentiment_task.num_classes}")
print(f"标签映射: {sentiment_task.label2id}")

# 文本分类的典型评估指标
from sklearn.metrics import accuracy_score, precision_recall_fscore_support
from sklearn.metrics import classification_report, confusion_matrix

def evaluate_classifier(y_true, y_pred, labels):
    """全面评估文本分类模型"""
    # 基础指标
    accuracy = accuracy_score(y_true, y_pred)
    precision, recall, f1, support = precision_recall_fscore_support(
        y_true, y_pred, labels=labels, average=None
    )

    print(f"{'类别':<8} {'精确率':>8} {'召回率':>8} {'F1':>8} {'样本数':>8}")
    print("-" * 45)
    for i, label in enumerate(labels):
        print(f"{label:<8} {precision[i]:>8.3f} {recall[i]:>8.3f} {f1[i]:>8.3f} {support[i]:>8}")
    print("-" * 45)
    print(f"整体准确率: {accuracy:.4f}")
    print(f"
分类报告:")
    print(classification_report(y_true, y_pred, target_names=labels))

# 示例：模型在测试集上的预测
y_true = ["正面", "正面", "负面", "中性", "正面", "负面"]
y_pred = ["正面", "中性", "负面", "中性", "正面", "正面"]
evaluate_classifier(y_true, y_pred, ["正面", "负面", "中性"])

# TF-IDF 特征提取 + 朴素贝叶斯分类
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split

# 示例数据集
texts = [
    "这部电影太精彩了，演员演技炸裂",
    "剧情拖沓，浪费时间，差评",
    "人工智能正在改变世界",
    "量子计算有望突破摩尔定律瓶颈",
    "服务太差，再也不会来了",
    "深度学习在医疗诊断中取得突破",
]
labels = ["正面", "负面", "科技", "科技", "负面", "科技"]

# 构建 Pipeline：TF-IDF + 朴素贝叶斯
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(
        max_features=5000,
        ngram_range=(1, 2),    # 使用 unigram 和 bigram
        min_df=1,
        sublinear_tf=True      # 对 TF 做对数平滑
    )),
    ("clf", MultinomialNB(alpha=0.1))  # 拉普拉斯平滑
])

# 训练与评估
X_train, X_test, y_train, y_test = train_test_split(
    texts, labels, test_size=0.3, random_state=42
)
pipeline.fit(X_train, y_train)
preds = pipeline.predict(X_test)
print(f"准确率: {sum(p == t for p, t in zip(preds, y_test)) / len(y_test):.2f}")

# TF-IDF + SVM 文本分类（通常效果更好）
from sklearn.svm import LinearSVC
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import GridSearchCV

# 更大的数据集示例
train_texts = [
    "产品非常好，质量超出预期",
    "快递太慢了，包装也破损",
    "新款 iPhone 搭载了 M4 芯片",
    "特斯拉发布新一代自动驾驶系统",
    "客服态度恶劣，要求退款",
    "这个手机拍照效果绝了",
    "谷歌发布 Gemini 2.0 模型",
    "物流速度很快，第二天就到了",
    "AMD 新一代处理器性能强劲",
    "食物已经变质，吃出异物",
]
train_labels = [
    "正面", "负面", "科技", "科技", "负面",
    "正面", "科技", "正面", "科技", "负面"
]

# Pipeline + 网格搜索最优参数
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(max_features=10000, ngram_range=(1, 2))),
    ("clf", LinearSVC(dual=True, class_weight="balanced"))
])

param_grid = {
    "tfidf__max_features": [5000, 10000],
    "tfidf__ngram_range": [(1, 1), (1, 2)],
    "clf__C": [0.1, 1.0, 10.0],
}

grid = GridSearchCV(pipeline, param_grid, cv=3, scoring="f1_macro")
grid.fit(train_texts, train_labels)
print(f"最佳参数: {grid.best_params_}")
print(f"最佳 Macro-F1: {grid.best_score_:.4f}")

# 查看 TF-IDF 最重要的特征
tfidf = grid.best_estimator_.named_steps["tfidf"]
feature_names = tfidf.get_feature_names_out()
print(f"词汇表大小: {len(feature_names)}")

# TextCNN 模型实现（PyTorch）
import torch
import torch.nn as nn
import torch.nn.functional as F

class TextCNN(nn.Module):
    """Kim (2014) TextCNN 文本分类模型"""

    def __init__(self, vocab_size, embed_dim, num_classes,
                 kernel_sizes=(3, 4, 5), num_filters=100, dropout=0.5):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)

        # 多尺度卷积核：捕获不同长度的 n-gram 模式
        self.convs = nn.ModuleList([
            nn.Conv1d(in_channels=embed_dim,
                      out_channels=num_filters,
                      kernel_size=k)
            for k in kernel_sizes
        ])

        self.dropout = nn.Dropout(dropout)
        self.fc = nn.Linear(num_filters * len(kernel_sizes), num_classes)

    def forward(self, x):
        # x: (batch, seq_len) -> 词索引序列
        x = self.embedding(x)           # (batch, seq_len, embed_dim)
        x = x.permute(0, 2, 1)          # (batch, embed_dim, seq_len)

        # 多个卷积核并行处理
        conv_outs = []
        for conv in self.convs:
            c = F.relu(conv(x))         # 卷积 + ReLU
            c = F.max_pool1d(c, c.size(2)).squeeze(2)  # 1-max pooling
            conv_outs.append(c)

        x = torch.cat(conv_outs, dim=1) # 拼接所有卷积核的输出
        x = self.dropout(x)
        return self.fc(x)               # (batch, num_classes)

# 实例化模型
model = TextCNN(vocab_size=30000, embed_dim=300, num_classes=3)
print(f"参数量: {sum(p.numel() for p in model.parameters()):,}")

# BiLSTM + Attention 文本分类模型
class AttentionLayer(nn.Module):
    """自注意力层：为每个时间步计算注意力权重"""

    def __init__(self, hidden_dim):
        super().__init__()
        self.attention = nn.Linear(hidden_dim, 1)

    def forward(self, lstm_out, mask=None):
        # lstm_out: (batch, seq_len, hidden_dim)
        scores = self.attention(lstm_out).squeeze(2)  # (batch, seq_len)
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        weights = F.softmax(scores, dim=1)            # 归一化
        # 加权求和
        context = torch.bmm(weights.unsqueeze(1), lstm_out).squeeze(1)
        return context, weights

class BiLSTMAttention(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_classes, dropout=0.3):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, batch_first=True,
                            bidirectional=True, num_layers=2)
        self.attention = AttentionLayer(hidden_dim * 2)
        self.dropout = nn.Dropout(dropout)
        self.fc = nn.Linear(hidden_dim * 2, num_classes)

    def forward(self, x, mask=None):
        x = self.embedding(x)
        lstm_out, _ = self.lstm(x)
        context, attn_weights = self.attention(lstm_out, mask)
        return self.fc(self.dropout(context)), attn_weights

# HuggingFace Transformers：BERT 微调完整流程
from transformers import (
    AutoTokenizer, AutoModelForSequenceClassification,
    TrainingArguments, Trainer
)
from datasets import Dataset
import torch

# 1. 加载预训练模型和分词器
model_name = "bert-base-chinese"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(
    model_name, num_labels=3  # 正面/负面/中性
)

# 2. 准备数据
texts = ["这部电影太精彩了", "剧情很差，浪费时间", "中规中矩，没什么亮点"]
labels = [0, 1, 2]  # 0=正面, 1=负面, 2=中性
dataset = Dataset.from_dict({"text": texts, "label": labels})

# 3. 分词处理
def tokenize_fn(batch):
    return tokenizer(
        batch["text"],
        padding="max_length",
        truncation=True,
        max_length=128,
        return_tensors=None  # 返回 dict 而非 tensor
    )

dataset = dataset.map(tokenize_fn, batched=True)
dataset = dataset.rename_column("label", "labels")
dataset.set_format(type="torch", columns=[
    "input_ids", "attention_mask", "labels"
])

# 4. 训练配置
training_args = TrainingArguments(
    output_dir="./bert-sentiment",
    num_train_epochs=3,
    per_device_train_batch_size=16,
    learning_rate=2e-5,
    weight_decay=0.01,
    evaluation_strategy="no",
    save_strategy="epoch",
    logging_steps=10,
)

trainer = Trainer(model=model, args=training_args, train_dataset=dataset)
trainer.train()  # 开始微调

# BERT 分类推理 + 可解释性分析
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import numpy as np

# 加载微调好的模型
model = AutoModelForSequenceClassification.from_pretrained(
    "./bert-sentiment/checkpoint-epoch-3"
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
model.eval()

def predict_with_confidence(text):
    """预测并返回置信度"""
    inputs = tokenizer(text, return_tensors="pt", truncation=True,
                       max_length=128, padding="max_length")
    with torch.no_grad():
        outputs = model(**inputs)
        logits = outputs.logits
        probs = torch.softmax(logits, dim=1)[0]
        pred_id = torch.argmax(probs).item()

    label2name = {0: "正面", 1: "负面", 2: "中性"}
    confidence = probs[pred_id].item()
    return label2name[pred_id], confidence, probs.tolist()

# 测试
test_sentences = [
    "这款手机性价比超高，拍照清晰",
    "质量太差了，用了三天就坏了",
    "一般般吧，没什么特别的",
]
for sent in test_sentences:
    label, conf, all_probs = predict_with_confidence(sent)
    print(f"文本: {sent}")
    print(f"预测: {label} (置信度: {conf:.2%})")
    print(f"概率分布: 正面={all_probs[0]:.3f}, 负面={all_probs[1]:.3f}, 中性={all_probs[2]:.3f}")
    print()

# 方法一：类别权重 + Focal Loss
import torch
import torch.nn as nn
from sklearn.utils.class_weight import compute_class_weight
import numpy as np

# 假设训练数据中类别分布极度不平衡
train_labels = np.array([0, 0, 0, 0, 0, 0, 0, 0, 1, 1])  # 80% vs 20%

# 计算类别权重（sklearn 自动平衡）
classes = np.unique(train_labels)
weights = compute_class_weight("balanced", classes=classes, y=train_labels)
class_weights = torch.FloatTensor(weights)
print(f"类别权重: {class_weights}")  # 少数类获得更高权重

# Focal Loss 实现
class FocalLoss(nn.Module):
    """Focal Loss: 降低易分样本的权重，聚焦难分样本"""
    def __init__(self, alpha=1.0, gamma=2.0):
        super().__init__()
        self.alpha = alpha
        self.gamma = gamma
        self.ce = nn.CrossEntropyLoss()

    def forward(self, logits, targets):
        ce_loss = self.ce(logits, targets)
        pt = torch.exp(-ce_loss)
        focal_loss = self.alpha * (1 - pt) ** self.gamma * ce_loss
        return focal_loss

# 对比：标准 CE vs Focal Loss
# gamma=2 时，pt=0.9（易分样本）的权重为 (1-0.9)^2 = 0.01
# pt=0.3（难分样本）的权重为 (1-0.3)^2 = 0.49
# Focal Loss 让模型关注那些难以分类的样本

# 方法二：SMOTE 过采样 + 阈值调优
from imblearn.over_sampling import SMOTE
from imblearn.pipeline import Pipeline as ImbPipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
import numpy as np

# 模拟不平衡数据
texts_majority = ["产品很好", "非常满意", "推荐购买"] * 100
texts_minority = ["质量有问题", "售后服务差"] * 20
all_texts = texts_majority + texts_minority
all_labels = [0] * 300 + [1] * 40

print(f"类别分布: 0类={all_labels.count(0)}, 1类={all_labels.count(1)}")
print(f"不平衡比: {all_labels.count(0) / all_labels.count(1):.1f}:1")

# SMOTE 不能直接处理文本，需要先用 TF-IDF 转为数值特征
vectorizer = TfidfVectorizer(max_features=5000)
X = vectorizer.fit_transform(all_texts)

# SMOTE 过采样
smote = SMOTE(random_state=42, k_neighbors=3)
X_res, y_res = smote.fit_resample(X, all_labels)
print(f"过采样后: 0类={sum(1 for y in y_res if y==0)}, 1类={sum(1 for y in y_res if y==1)}")

# 阈值调优：找到最优分类阈值
from sklearn.metrics import f1_score
model = LogisticRegression(class_weight="balanced")
model.fit(X_res, y_res)
probs = model.predict_proba(X)[:, 1]

best_threshold, best_f1 = 0.5, 0
for threshold in np.arange(0.1, 0.9, 0.01):
    preds = (probs >= threshold).astype(int)
    f1 = f1_score(all_labels, preds)
    if f1 > best_f1:
        best_f1, best_threshold = f1, threshold

print(f"最优阈值: {best_threshold:.2f}, F1: {best_f1:.4f}")

# sklearn 多标签分类：Binary Relevance vs Classifier Chains
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.multiclass import OneVsRestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.multioutput import ClassifierChain
from sklearn.metrics import hamming_loss, f1_score
import numpy as np

# 多标签数据集
texts = [
    "AI 技术在医疗领域取得重大突破",
    "苹果公司发布新一代 iPhone 手机",
    "央行宣布降息 0.25 个百分点",
    "世界杯决赛法国对阵阿根廷",
    "量子计算机实现 1000 量子比特突破",
    "美联储加息 75 个基点应对通胀",
]
# 多标签：每篇文章可同时属于多个类别
all_labels_set = ["科技", "商业", "金融", "体育", "医疗"]
labels = [
    [1, 0, 0, 0, 1],  # 科技 + 医疗
    [1, 1, 0, 0, 0],  # 科技 + 商业
    [0, 1, 1, 0, 0],  # 商业 + 金融
    [0, 0, 0, 1, 0],  # 体育
    [1, 0, 0, 0, 0],  # 科技
    [0, 1, 1, 0, 0],  # 商业 + 金融
]
labels = np.array(labels)

# Binary Relevance: 每个标签独立分类
br_model = OneVsRestClassifier(
    LogisticRegression(max_iter=1000)
)
X = TfidfVectorizer().fit_transform(texts)
br_model.fit(X, labels)
br_preds = br_model.predict(X)

# Classifier Chains: 利用标签相关性
cc_model = ClassifierChain(
    LogisticRegression(max_iter=1000),
    order="random", random_state=42
)
cc_model.fit(X, labels)
cc_preds = cc_model.predict(X)

print("=== Binary Relevance ===")
print(f"Hamming Loss: {hamming_loss(labels, br_preds):.4f}")
print(f"Micro-F1: {f1_score(labels, br_preds, average='micro'):.4f}")
print(f"Macro-F1: {f1_score(labels, br_preds, average='macro'):.4f}")
print("\n=== Classifier Chains ===")
print(f"Hamming Loss: {hamming_loss(labels, cc_preds):.4f}")
print(f"Micro-F1: {f1_score(labels, cc_preds, average='micro'):.4f}")
print(f"Macro-F1: {f1_score(labels, cc_preds, average='macro'):.4f}")

# 深度学习多标签分类：PyTorch + Sigmoid 输出
import torch
import torch.nn as nn

class MultiLabelTextClassifier(nn.Module):
    """多标签文本分类器：每标签独立的 Sigmoid 输出"""

    def __init__(self, vocab_size, embed_dim, num_labels,
                 hidden_dim=256, dropout=0.3):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, batch_first=True,
                            bidirectional=True)
        self.dropout = nn.Dropout(dropout)
        # 输出层：num_labels 个独立的 Sigmoid
        self.classifier = nn.Linear(hidden_dim * 2, num_labels)

    def forward(self, input_ids, attention_mask=None):
        x = self.embedding(input_ids)
        lstm_out, (h_n, _) = self.lstm(x)
        # 取最后一个时间步的双向隐藏状态拼接
        # 或使用 pooling：这里用 mean pooling
        if attention_mask is not None:
            mask = attention_mask.unsqueeze(-1).float()
            x = (lstm_out * mask).sum(dim=1) / mask.sum(dim=1)
        else:
            x = lstm_out.mean(dim=1)

        x = self.dropout(x)
        logits = self.classifier(x)   # (batch, num_labels)
        # 注意：训练时 BCEWithLogitsLoss 内部包含 Sigmoid
        # 推理时手动 Sigmoid 得到概率
        return logits

# 训练配置
num_labels = 5
criterion = nn.BCEWithLogitsLoss()  # 多标签 BCE（内置 Sigmoid）

# 推理
model.eval()
with torch.no_grad():
    logits = model(input_ids, attention_mask)
    probs = torch.sigmoid(logits)
    # 方法一：固定阈值 0.5
    predictions = (probs > 0.5).float()
    # 方法二：每标签自适应阈值
    # 需要在验证集上统计每标签的最优阈值

# 完整项目：sklearn 新闻分类 Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import classification_report
import joblib
import json

# 1. 加载数据
news_data = {
    "texts": [
        "央行今日开展 1000 亿元逆回购操作",
        "中国队在亚洲杯半决赛中战胜日本队",
        "SpaceX 星舰第四次试飞成功着陆",
        "华为发布鸿蒙 HarmonyOS 5.0 系统",
        "A 股三大指数集体高开，券商板块领涨",
        "NBA 总决赛湖人 4:2 击败凯尔特人",
        "新型 mRNA 疫苗对变异病毒有效率达 85%",
        "教育部推出双减政策实施细则",
        "特斯拉中国工厂产能突破 100 万辆",
        "世界杯预选赛中国队 3:0 获胜",
    ],
    "labels": ["财经", "体育", "科技", "科技", "财经", "体育", "科技", "社会", "财经", "体育"],
}

# 2. 划分数据集
X_train, X_test, y_train, y_test = train_test_split(
    news_data["texts"], news_data["labels"],
    test_size=0.2, random_state=42, stratify=news_data["labels"]
)

# 3. 构建分类 Pipeline
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(
        max_features=50000,
        ngram_range=(1, 2),
        min_df=2,
        sublinear_tf=True
    )),
    ("clf", LogisticRegression(
        C=1.0, max_iter=1000,
        class_weight="balanced",
        solver="lbfgs"
    ))
])

# 4. 训练
pipeline.fit(X_train, y_train)

# 5. 评估
y_pred = pipeline.predict(X_test)
print(classification_report(y_test, y_pred, zero_division=0))

# 6. 交叉验证
cv_scores = cross_val_score(pipeline, X_train, y_train, cv=3, scoring="f1_macro")
print(f"Cross-Validation Macro-F1: {cv_scores.mean():.4f} (+/- {cv_scores.std():.4f})")

# 7. 保存模型
joblib.dump(pipeline, "news_classifier.pkl")
print("模型已保存到 news_classifier.pkl")

# 完整项目：HuggingFace 新闻分类（含部署接口）
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
import json

class NewsClassifier:
    """生产级新闻分类器"""

    def __init__(self, model_path_or_name="hfl/chinese-macbert-base"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path_or_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_path_or_name, num_labels=14
        )
        self.categories = [
            "财经", "房产", "教育", "科技", "军事",
            "汽车", "体育", "娱乐", "时尚", "社会",
            "游戏", "家居", "彩票", "星座",
        ]
        self.model.eval()

    def predict(self, text, top_k=3):
        """预测新闻类别，返回 top_k 个结果"""
        inputs = self.tokenizer(
            text, return_tensors="pt", truncation=True,
            max_length=512, padding="max_length"
        )
        with torch.no_grad():
            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=1)[0]
            top_probs, top_ids = torch.topk(probs, top_k)

        results = []
        for prob, idx in zip(top_probs, top_ids):
            results.append({
                "category": self.categories[idx.item()],
                "confidence": round(prob.item(), 4),
            })
        return results

    def batch_predict(self, texts, batch_size=16):
        """批量预测"""
        all_results = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            for text in batch:
                all_results.append(self.predict(text))
        return all_results

# 使用示例
classifier = NewsClassifier()
text = "苹果发布会：iPhone 16 搭载 A18 芯片，支持 Apple Intelligence"
results = classifier.predict(text)
print(f"文本: {text}")
for r in results:
    print(f"  {r['category']}: {r['confidence']:.1%}")