情感分析实战:群聊对骂检测

# 多标签分类:sigmoid + BCEWithLogitsLoss
class MultiLabelClassifier(nn.Module):
    def __init__(self, input_dim, num_labels):
        super().__init__()
        self.fc = nn.Linear(input_dim, num_labels)

    def forward(self, x):
        return self.fc(x)  # 不加 softmax,由 loss 函数处理

# 每个标签独立判断是否命中
criterion = nn.BCEWithLogitsLoss()  # 内含 sigmoid
# 标签格式: [1, 1, 0, 0, 1] → 对骂✓ 人身攻击✓ 带节奏✗ 性别歧视✗ 地域歧视✓

def build_context_window(messages, idx, window=2):
    """构建第 idx 条消息的上下文窗口"""
    start = max(0, idx - window)
    end = min(len(messages), idx + window + 1)

    parts = []
    for i in range(start, end):
        speaker = messages[i]["user"]
        text = messages[i]["text"]
        if i == idx:
            parts.append(f"[CUR] {speaker}: {text}")  # 标记当前消息
        else:
            parts.append(f"{speaker}: {text}")

    return " [SEP] ".join(parts)

# 示例输出:
# "用户B: 怎么了? [SEP] 用户A: 一堆 bug 还好意思提测 [SEP] [CUR] 用户B: 你行你来"

def extract_dialog_features(messages, idx, window=2):
    """提取对话级别的统计特征"""
    start = max(0, idx - window)
    end = min(len(messages), idx + window + 1)
    context = messages[start:end]
    current = messages[idx]

    features = {}

    # 回复频率:当前用户在窗口内的消息数
    user = current["user"]
    features["user_msg_count"] = sum(
        1 for m in context if m["user"] == user
    )

    # 回复速度:与上一条消息的时间差(秒)
    if idx > 0:
        delta = current["timestamp"] - messages[idx - 1]["timestamp"]
        features["reply_speed"] = delta.total_seconds()
    else:
        features["reply_speed"] = 999

    # 对话密度:窗口内消息数 / 时间跨度
    if len(context) > 1:
        time_span = (context[-1]["timestamp"] - context[0]["timestamp"]).total_seconds()
        features["dialog_density"] = len(context) / max(time_span, 1)
    else:
        features["dialog_density"] = 0

    # @频率
    features["at_count"] = sum(
        m["text"].count("@") for m in context
    )

    return features

class HybridClassifier(nn.Module):
    def __init__(self, text_model, dialog_feat_dim, num_classes):
        super().__init__()
        self.text_model = text_model             # TextCNN 或 LSTM
        self.dialog_fc = nn.Linear(dialog_feat_dim, 16)
        self.classifier = nn.Linear(128 + 16, num_classes)  # 拼接后分类

    def forward(self, text_input, dialog_features):
        text_vec = self.text_model.get_features(text_input)  # 128 维
        dialog_vec = torch.relu(self.dialog_fc(dialog_features))  # 16 维
        combined = torch.cat([text_vec, dialog_vec], dim=1)
        return self.classifier(combined)

def post_process(msg, model_prob, context_probs):
    """后处理:结合规则和上下文做最终判断"""

    # 规则 1:脏话词表强制命中
    if hit_abuse_keyword(msg["text"]):
        return max(model_prob, 0.85)

    # 规则 2:上下文升级--前后消息也被判为对骂,置信度提升
    recent_abuse = sum(1 for p in context_probs if p > 0.6)
    if recent_abuse >= 2:
        return min(model_prob * 1.3, 1.0)  # 上下文加持,但不超过 1

    # 规则 3:自嘲降级--主语是"我"的负面表达
    if msg["text"].startswith(("我真是", "我好")) and model_prob < 0.8:
        return model_prob * 0.5

    return model_prob

def preprocess_real_message(text):
    """真实消息预处理:处理常见噪声"""

    # 1. 繁简转换
    text = opencc.convert(text, "t2s")

    # 2. 谐音/变体还原(维护映射表)
    replacements = {
        "沙比": "傻逼", "沙b": "傻逼", "sha bi": "傻逼",
        "nmsl": "你妈死了", "nmsle": "你妈死了",
        "nt": "脑瘫", "nc": "脑残",
    }
    for variant, canonical in replacements.items():
        text = text.lower().replace(variant, canonical)

    # 3. 过滤纯表情/图片消息(返回空,跳过分类)
    if text.startswith("[") and text.endswith("]"):
        return None

    # 4. 截断过长消息
    return text[:256]

# 按类别看 F1
print(classification_report(y_true, y_pred, target_names=labels))

# 看混淆矩阵:漏报和误报各多少
from sklearn.metrics import confusion_matrix
cm = confusion_matrix(y_true, y_pred)
print(f"漏报(FN): {cm[1][0]} 条")
print(f"误报(FP): {cm[0][1]} 条")

# 灰度上线的阈值配置
THRESHOLDS = {
    "shadow":    {"action": "log_only",    "min_prob": 0.0},
    "cautious":  {"action": "warn",        "min_prob": 0.95},
    "normal":    {"action": "mute_5min",   "min_prob": 0.85},
    "aggressive": {"action": "mute_30min", "min_prob": 0.70},
}