import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import matplotlib.pyplot as plt

def compare_model_vs_data_centric(X, y, model_class, noise_level=0.1):
    """
    模型中心 vs 数据中心方法性能对比

    Args:
        X: 特征矩阵
        y: 标签
        model_class: 基础模型类
        noise_level: 标签噪声比例
    """
    # 添加标签噪声
    noisy_y = y.copy()
    noise_idx = np.random.choice(len(y), int(len(y) * noise_level), replace=False)
    n_classes = len(np.unique(y))
    for idx in noise_idx:
        wrong_labels = [l for l in range(n_classes) if l != y[idx]]
        noisy_y[idx] = np.random.choice(wrong_labels)

    X_train, X_test, y_train_clean, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    _, _, y_train_noisy, _ = train_test_split(
        X, noisy_y, test_size=0.2, random_state=42
    )

    # --- 模型中心方法 ---
    # 噪声数据 + 基础模型
    base_model = model_class()
    base_model.fit(X_train, y_train_noisy)
    base_acc = accuracy_score(y_test, base_model.predict(X_test))

    # 噪声数据 + 更复杂的模型(集成)
    from sklearn.ensemble import GradientBoostingClassifier
    complex_model = GradientBoostingClassifier(n_estimators=200)
    complex_model.fit(X_train, y_train_noisy)
    complex_acc = accuracy_score(y_test, complex_model.predict(X_test))

    # --- 数据中心方法 ---
    # 干净数据 + 基础模型
    clean_model = model_class()
    clean_model.fit(X_train, y_train_clean)
    clean_acc = accuracy_score(y_test, clean_model.predict(X_test))

    print("=" * 50)
    print("模型中心 vs 数据中心对比")
    print("=" * 50)
    print(f"基础模型 + 噪声数据: {base_acc:.3f}")
    print(f"复杂模型 + 噪声数据: {complex_acc:.3f}")
    print(f"基础模型 + 干净数据: {clean_acc:.3f}")
    print(f"\n模型改进带来的效果: +{(complex_acc - base_acc):.3f}")
    print(f"数据改进带来的效果: +{(clean_acc - base_acc):.3f}")

    return {
        'base_model_noisy_data': base_acc,
        'complex_model_noisy_data': complex_acc,
        'base_model_clean_data': clean_acc
    }

import cleanlab
from cleanlab.filter import find_label_issues
from cleanlab.classification import CleanLearning
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_predict

def detect_label_errors_cleanlab(X, y, model=None):
    """
    用 Cleanlab 检测标签错误

    Args:
        X: 特征矩阵
        y: 标签数组
        model: 分类模型(默认: LogisticRegression)

    Returns:
        label_issues: 标签错误的索引及相关信息
    """
    if model is None:
        model = LogisticRegression(max_iter=1000)

    # 通过交叉验证预测类别概率
    pred_probs = cross_val_predict(
        model, X, y,
        cv=5,
        method='predict_proba'
    )

    # 查找标签错误
    label_issues = find_label_issues(
        labels=y,
        pred_probs=pred_probs,
        return_indices_ranked_by='self_confidence'
    )

    print(f"总样本数: {len(y)}")
    print(f"发现的标签错误: {len(label_issues)}")
    print(f"错误率: {len(label_issues)/len(y):.2%}")

    return label_issues


def cleanlab_full_pipeline(X_train, y_train_noisy, X_test, y_test):
    """
    Cleanlab 完整流水线：
    1. 检测标签错误
    2. 修正或剔除错误
    3. 用清洗后的数据重新训练模型
    """
    from cleanlab.classification import CleanLearning

    base_model = LogisticRegression(max_iter=1000)

    # CleanLearning：在训练过程中自动处理标签错误
    cl = CleanLearning(base_model, seed=42)
    cl.fit(X_train, y_train_noisy)

    # 评估
    y_pred = cl.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)
    print(f"CleanLearning 准确率: {accuracy:.3f}")

    # 输出标签问题信息
    label_issues_df = cl.get_label_issues()
    print(f"\n标签问题信息:")
    print(label_issues_df.head(10))

    return cl, label_issues_df


def confident_learning_manual(pred_probs, labels):
    """
    Confident Learning 的手动实现
    - 计算各类别的阈值
    - 构建 Confident Joint 矩阵
    """
    n_classes = pred_probs.shape[1]
    n_samples = len(labels)

    # 各类别阈值：该类别样本的平均预测概率
    thresholds = np.zeros(n_classes)
    for c in range(n_classes):
        class_mask = labels == c
        if class_mask.sum() > 0:
            thresholds[c] = pred_probs[class_mask, c].mean()

    # Confident Joint 矩阵 C[s][y]
    # s: 预测类别，y: 给定的标签
    C = np.zeros((n_classes, n_classes), dtype=int)

    for i in range(n_samples):
        y_given = labels[i]
        # 概率最高的类别(在超过阈值的类别中)
        over_threshold = pred_probs[i] >= thresholds

        if over_threshold.sum() == 0:
            y_hat = pred_probs[i].argmax()
        else:
            y_hat = (pred_probs[i] * over_threshold).argmax()

        C[y_hat, y_given] += 1

    # 非对角线元素多的类别对，是潜在的错误候选
    off_diagonal = C.copy()
    np.fill_diagonal(off_diagonal, 0)

    print("Confident Joint 矩阵(行: 估计的真实类别，列: 给定的标签):")
    print(C)
    print(f"\n潜在错误样本数: {off_diagonal.sum()}")

    return C

from sklearn.ensemble import IsolationForest
from sklearn.neighbors import LocalOutlierFactor
import torch
import numpy as np

class DataQualityChecker:
    """
    综合数据质量检查工具
    """

    def __init__(self):
        self.outlier_detector = None
        self.quality_report = {}

    def check_class_distribution(self, labels):
        """检查类别不均衡"""
        from collections import Counter
        import pandas as pd

        counts = Counter(labels)
        total = len(labels)

        df = pd.DataFrame([
            {'class': c, 'count': n, 'percentage': 100 * n / total}
            for c, n in sorted(counts.items())
        ])

        imbalance_ratio = max(counts.values()) / min(counts.values())

        print("类别分布:")
        print(df.to_string(index=False))
        print(f"\n不均衡比例: {imbalance_ratio:.2f}x")

        if imbalance_ratio > 10:
            print("警告: 存在严重的类别不均衡！")
        elif imbalance_ratio > 3:
            print("注意: 存在类别不均衡")

        self.quality_report['class_imbalance_ratio'] = imbalance_ratio
        return df

    def detect_outliers(self, X, method='isolation_forest', contamination=0.1):
        """
        异常值检测

        Args:
            method: 'isolation_forest' 或 'lof'
            contamination: 预期异常值比例
        """
        if method == 'isolation_forest':
            detector = IsolationForest(
                contamination=contamination,
                random_state=42
            )
        elif method == 'lof':
            detector = LocalOutlierFactor(
                contamination=contamination,
                novelty=True
            )

        # 异常值检测(-1: 异常值, 1: 正常)
        predictions = detector.fit_predict(X)
        outlier_mask = predictions == -1
        outlier_indices = np.where(outlier_mask)[0]

        print(f"检测到的异常值: {outlier_mask.sum()} / {len(X)} ({outlier_mask.mean():.2%})")

        self.quality_report['n_outliers'] = outlier_mask.sum()
        return outlier_indices, outlier_mask

    def check_duplicates(self, X, y=None, threshold=0.99):
        """
        重复样本检测

        Args:
            threshold: 相似度阈值(1.0 = 完全相同)
        """
        from sklearn.metrics.pairwise import cosine_similarity

        # 计算相似度矩阵(大规模数据时进行采样)
        if len(X) > 10000:
            sample_idx = np.random.choice(len(X), 10000, replace=False)
            X_sample = X[sample_idx]
        else:
            X_sample = X
            sample_idx = np.arange(len(X))

        sim_matrix = cosine_similarity(X_sample)
        np.fill_diagonal(sim_matrix, 0)

        # 查找相似度在阈值以上的样本对
        duplicate_pairs = np.argwhere(sim_matrix >= threshold)
        # 去重(只保留 i < j)
        duplicate_pairs = duplicate_pairs[duplicate_pairs[:, 0] < duplicate_pairs[:, 1]]

        print(f"发现的重复样本对: {len(duplicate_pairs)}")

        if y is not None and len(duplicate_pairs) > 0:
            # 标签不同的重复样本(潜在错误)
            label_conflicts = 0
            for i, j in duplicate_pairs:
                if y[sample_idx[i]] != y[sample_idx[j]]:
                    label_conflicts += 1
            print(f"标签冲突的重复样本对: {label_conflicts}")

        return duplicate_pairs

    def compute_data_quality_score(self, X, y):
        """计算综合数据质量分数"""
        scores = {}

        # 缺失值比例
        if hasattr(X, 'isnull'):
            missing_rate = X.isnull().mean().mean()
        else:
            missing_rate = np.isnan(X).mean()
        scores['completeness'] = 1 - missing_rate

        # 类别均衡度
        from collections import Counter
        counts = Counter(y)
        n_classes = len(counts)
        ideal_count = len(y) / n_classes
        balance_score = sum(
            min(c, ideal_count) / ideal_count
            for c in counts.values()
        ) / n_classes
        scores['balance'] = balance_score

        # 最终分数
        overall_score = np.mean(list(scores.values()))
        scores['overall'] = overall_score

        print("数据质量分数:")
        for metric, score in scores.items():
            print(f"  {metric}: {score:.3f}")

        return scores

def compute_inter_rater_agreement(annotations):
    """
    计算标注者间一致度

    Args:
        annotations: 形状为 (n_samples, n_raters) 的数组

    Returns:
        cohen_kappa: Cohen's Kappa 分数
        fleiss_kappa: Fleiss' Kappa 分数(多标注者场景)
    """
    from sklearn.metrics import cohen_kappa_score
    import numpy as np

    n_samples, n_raters = annotations.shape

    # 标注者两两之间的 Cohen's Kappa
    kappa_scores = []
    for i in range(n_raters):
        for j in range(i+1, n_raters):
            kappa = cohen_kappa_score(
                annotations[:, i],
                annotations[:, j]
            )
            kappa_scores.append((i, j, kappa))
            print(f"标注者 {i} vs 标注者 {j}: kappa = {kappa:.3f}")

    mean_kappa = np.mean([k for _, _, k in kappa_scores])
    print(f"\n平均 Cohen's Kappa: {mean_kappa:.3f}")

    # Kappa 解读
    if mean_kappa < 0.2:
        interpretation = "轻微一致"
    elif mean_kappa < 0.4:
        interpretation = "一般一致"
    elif mean_kappa < 0.6:
        interpretation = "中等一致"
    elif mean_kappa < 0.8:
        interpretation = "高度一致"
    else:
        interpretation = "几乎完全一致"

    print(f"解读: {interpretation}")

    # 生成多数决标签
    from scipy import stats
    majority_labels = stats.mode(annotations, axis=1)[0].flatten()
    print(f"\n多数决标签生成完成")

    return mean_kappa, majority_labels


def create_labeling_guidelines(task_name, examples):
    """
    生成标注指南模板

    优质标注指南应具备的要素：
    1. 清晰的定义与边界
    2. 正例/负例示例
    3. 边界情形的处理方式
    4. 一致性检查清单
    """
    guidelines = {
        'task': task_name,
        'categories': {},
        'edge_cases': [],
        'consistency_rules': []
    }

    print(f"'{task_name}' 标注指南模板:")
    print("-" * 50)
    print("1. 为每个类别写出清晰的定义")
    print("2. 至少包含 5 个正例和 5 个负例")
    print("3. 明确边界情形(Boundary Cases)的处理方式")
    print("4. 定义标注者无法确定时的默认规则")
    print("5. 包含质量检查标准")

    return guidelines

def snorkel_programmatic_labeling_demo():
    """
    使用 Snorkel 进行程序化标注的示例
    pip install snorkel
    """
    from snorkel.labeling import labeling_function, PandasLFApplier
    from snorkel.labeling.model import LabelModel
    import pandas as pd
    import re

    # 情感分类示例
    POSITIVE = 1
    NEGATIVE = 0
    ABSTAIN = -1

    @labeling_function()
    def lf_positive_keywords(x):
        """基于正面关键词的标注函数"""
        positive_words = ['good', 'great', 'excellent', 'amazing', 'love', 'best']
        text = x.text.lower()
        if any(word in text for word in positive_words):
            return POSITIVE
        return ABSTAIN

    @labeling_function()
    def lf_negative_keywords(x):
        """基于负面关键词的标注函数"""
        negative_words = ['bad', 'terrible', 'awful', 'hate', 'worst', 'horrible']
        text = x.text.lower()
        if any(word in text for word in negative_words):
            return NEGATIVE
        return ABSTAIN

    @labeling_function()
    def lf_rating_high(x):
        """基于高评分的标注函数"""
        if hasattr(x, 'rating') and x.rating >= 4:
            return POSITIVE
        return ABSTAIN

    @labeling_function()
    def lf_rating_low(x):
        """基于低评分的标注函数"""
        if hasattr(x, 'rating') and x.rating <= 2:
            return NEGATIVE
        return ABSTAIN

    @labeling_function()
    def lf_negation_check(x):
        """检测否定表达(NOT 运算符)"""
        text = x.text.lower()
        if re.search(r"not (good|great|excellent)", text):
            return NEGATIVE
        if re.search(r"not (bad|terrible)", text):
            return POSITIVE
        return ABSTAIN

    # 标注函数列表
    lfs = [
        lf_positive_keywords,
        lf_negative_keywords,
        lf_rating_high,
        lf_rating_low,
        lf_negation_check,
    ]

    # 将标注函数应用到数据上
    # df_train 是没有标签的文本数据框
    # applier = PandasLFApplier(lfs=lfs)
    # L_train = applier.apply(df_train)

    # 用标签模型进行整合
    # label_model = LabelModel(cardinality=2, verbose=True)
    # label_model.fit(L_train=L_train, n_epochs=500, log_freq=100)

    # 生成概率化标签
    # probs_train = label_model.predict_proba(L=L_train)

    print("Snorkel 程序化标注流水线:")
    print("1. 领域专家编写标注函数(LF)")
    print("2. 将 LF 应用到无标签数据上")
    print("3. 用标签模型整合多个 LF(去除噪声)")
    print("4. 用生成的软标签训练下游模型")
    print(f"\n定义的标注函数数量: {len(lfs)}")

    return lfs

import numpy as np
from sklearn.base import BaseEstimator
import torch
import torch.nn as nn

class ActiveLearner:
    """
    主动学习实现
    支持多种样本选择策略
    """

    def __init__(self, model, strategy='uncertainty', n_initial=100):
        self.model = model
        self.strategy = strategy
        self.n_initial = n_initial

    def uncertainty_sampling(self, X_unlabeled, n_samples):
        """
        不确定性采样
        选择模型最没把握的样本
        """
        # 预测概率
        probs = self._get_probs(X_unlabeled)

        if self.strategy == 'least_confidence':
            # 最高概率值最低的样本
            uncertainty = 1 - probs.max(axis=1)

        elif self.strategy == 'margin':
            # 前两个类别概率之差
            sorted_probs = np.sort(probs, axis=1)[:, ::-1]
            uncertainty = 1 - (sorted_probs[:, 0] - sorted_probs[:, 1])

        elif self.strategy == 'entropy':
            # 熵: 预测分布的不确定性
            uncertainty = -np.sum(probs * np.log(probs + 1e-10), axis=1)

        else:
            uncertainty = 1 - probs.max(axis=1)

        # 返回最不确定的样本索引
        selected_indices = np.argsort(uncertainty)[-n_samples:]
        return selected_indices, uncertainty

    def diversity_sampling(self, X_unlabeled, X_labeled, n_samples):
        """
        基于多样性的采样(CoreSet 方法)
        选择与已标注数据差异最大的样本
        """
        from sklearn.metrics.pairwise import euclidean_distances

        # 与已标注数据的最小距离
        distances = euclidean_distances(X_unlabeled, X_labeled)
        min_distances = distances.min(axis=1)

        # CoreSet: 依次选择距离最远的点
        selected = []
        remaining = list(range(len(X_unlabeled)))
        current_labeled = X_labeled.copy()

        for _ in range(n_samples):
            dists = euclidean_distances(
                X_unlabeled[remaining],
                current_labeled
            ).min(axis=1)

            # 选择距离最远的点
            best_idx = remaining[np.argmax(dists)]
            selected.append(best_idx)
            remaining.remove(best_idx)
            current_labeled = np.vstack([current_labeled, X_unlabeled[best_idx]])

        return np.array(selected)

    def batch_mode_active_learning(self, X_pool, y_oracle, X_test, y_test,
                                    n_iterations=10, n_per_iter=50):
        """
        批量模式主动学习循环

        Args:
            X_pool: 无标签数据池
            y_oracle: 标注者(真实标签来源)
            n_per_iter: 每轮标注的样本数
        """
        # 初始已标注集合
        initial_indices = np.random.choice(
            len(X_pool), self.n_initial, replace=False
        )
        labeled_indices = list(initial_indices)
        unlabeled_indices = [
            i for i in range(len(X_pool)) if i not in labeled_indices
        ]

        accuracies = []
        n_labeled_list = []

        for iteration in range(n_iterations):
            X_labeled = X_pool[labeled_indices]
            y_labeled = y_oracle[labeled_indices]

            # 用当前已标注数据训练模型
            self.model.fit(X_labeled, y_labeled)

            # 评估测试准确率
            acc = accuracy_score(y_test, self.model.predict(X_test))
            accuracies.append(acc)
            n_labeled_list.append(len(labeled_indices))

            print(f"第 {iteration+1} 轮: 已标注数量={len(labeled_indices)}, 准确率={acc:.3f}")

            if len(unlabeled_indices) == 0:
                break

            # 选择下一轮要标注的样本
            X_unlabeled = X_pool[unlabeled_indices]
            selected, _ = self.uncertainty_sampling(X_unlabeled, n_per_iter)

            # 转换为实际索引
            actual_selected = [unlabeled_indices[i] for i in selected]

            # 加入已标注集合
            labeled_indices.extend(actual_selected)
            unlabeled_indices = [
                i for i in unlabeled_indices if i not in actual_selected
            ]

        return accuracies, n_labeled_list

    def _get_probs(self, X):
        """返回模型预测概率"""
        if hasattr(self.model, 'predict_proba'):
            return self.model.predict_proba(X)
        else:
            logits = self.model.predict(X)
            from scipy.special import softmax
            return softmax(logits, axis=1)


def compare_active_learning_strategies(X, y, model, n_initial=50, n_budget=500):
    """比较多种主动学习策略"""
    strategies = ['least_confidence', 'margin', 'entropy']
    results = {}

    for strategy in strategies:
        learner = ActiveLearner(model, strategy=strategy, n_initial=n_initial)
        accs, n_labeled = learner.batch_mode_active_learning(
            X, y,
            X_test=X[:100],
            y_test=y[:100],
            n_iterations=10,
            n_per_iter=50
        )
        results[strategy] = {'accuracies': accs, 'n_labeled': n_labeled}

    # 随机采样基线
    random_learner = ActiveLearner(model, strategy='random', n_initial=n_initial)
    # (随机策略的实现省略)

    return results

import albumentations as A
from albumentations.pytorch import ToTensorV2
import cv2
import numpy as np
from PIL import Image

def get_train_transforms(image_size=224):
    """
    训练用的强力增强流水线(Albumentations)
    """
    return A.Compose([
        # 几何变换
        A.RandomResizedCrop(
            height=image_size,
            width=image_size,
            scale=(0.7, 1.0),
            ratio=(0.75, 1.33)
        ),
        A.HorizontalFlip(p=0.5),
        A.ShiftScaleRotate(
            shift_limit=0.1,
            scale_limit=0.2,
            rotate_limit=30,
            p=0.5
        ),

        # 颜色变换
        A.ColorJitter(
            brightness=0.3,
            contrast=0.3,
            saturation=0.3,
            hue=0.1,
            p=0.8
        ),
        A.ToGray(p=0.1),
        A.RandomGamma(gamma_limit=(80, 120), p=0.3),

        # 噪声与模糊
        A.GaussNoise(var_limit=(10, 50), p=0.3),
        A.OneOf([
            A.MotionBlur(blur_limit=7),
            A.GaussianBlur(blur_limit=7),
            A.MedianBlur(blur_limit=7),
        ], p=0.3),

        # 遮挡 / 随机擦除
        A.CoarseDropout(
            max_holes=8,
            max_height=32,
            max_width=32,
            fill_value=0,
            p=0.3
        ),

        # 网格畸变
        A.OneOf([
            A.GridDistortion(p=1),
            A.ElasticTransform(p=1),
            A.OpticalDistortion(p=1),
        ], p=0.2),

        # 归一化与张量转换
        A.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        ),
        ToTensorV2(),
    ])


def get_val_transforms(image_size=224):
    """验证用变换(不做增强，仅归一化)"""
    return A.Compose([
        A.Resize(height=image_size, width=image_size),
        A.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        ),
        ToTensorV2(),
    ])


def mixup_augmentation(images, labels, alpha=0.4):
    """
    MixUp 增强: 混合两张图像及其标签
    Zhang et al., "mixup: Beyond Empirical Risk Minimization" (2018)
    """
    import torch

    batch_size = images.shape[0]
    lam = np.random.beta(alpha, alpha)

    # 随机打乱顺序
    perm = torch.randperm(batch_size)

    # 混合图像
    mixed_images = lam * images + (1 - lam) * images[perm]

    # 混合标签(软标签)
    labels_a = labels
    labels_b = labels[perm]

    return mixed_images, labels_a, labels_b, lam


def cutmix_augmentation(images, labels, alpha=1.0):
    """
    CutMix 增强: 把一张图像的一部分剪切粘贴到另一张图像上
    Yun et al., "CutMix: Training Strategy that Makes Use of
    Sample Mixing" (2019)
    """
    import torch

    batch_size, c, h, w = images.shape
    lam = np.random.beta(alpha, alpha)

    perm = torch.randperm(batch_size)

    # 生成随机方框
    cut_ratio = np.sqrt(1 - lam)
    cut_h = int(h * cut_ratio)
    cut_w = int(w * cut_ratio)

    cx = np.random.randint(w)
    cy = np.random.randint(h)

    bbx1 = np.clip(cx - cut_w // 2, 0, w)
    bby1 = np.clip(cy - cut_h // 2, 0, h)
    bbx2 = np.clip(cx + cut_w // 2, 0, w)
    bby2 = np.clip(cy + cut_h // 2, 0, h)

    # 用方框区域替换为另一张图像的对应部分
    mixed_images = images.clone()
    mixed_images[:, :, bby1:bby2, bbx1:bbx2] = images[perm, :, bby1:bby2, bbx1:bbx2]

    # 调整实际混合比例
    lam = 1 - ((bbx2 - bbx1) * (bby2 - bby1) / (w * h))

    labels_a = labels
    labels_b = labels[perm]

    return mixed_images, labels_a, labels_b, lam

class TextAugmenter:
    """
    文本数据增强技巧集合
    """

    def __init__(self, language='en'):
        self.language = language

    def eda_synonym_replacement(self, text, n=1):
        """
        EDA (Easy Data Augmentation): 同义词替换
        Wei and Zou, "EDA: Easy Data Augmentation Techniques
        for Boosting Performance on Text Classification Tasks" (2019)
        """
        import nltk
        from nltk.corpus import wordnet

        words = text.split()
        new_words = words.copy()

        # 排除停用词
        stop_words = set(['a', 'an', 'the', 'is', 'are', 'was', 'were',
                          'i', 'me', 'my', 'we', 'our', 'you', 'your'])

        # 查找可替换的单词
        replaceable = [
            (i, word) for i, word in enumerate(words)
            if word.lower() not in stop_words
        ]

        np.random.shuffle(replaceable)

        replaced = 0
        for idx, word in replaceable:
            if replaced >= n:
                break

            synsets = wordnet.synsets(word)
            if synsets:
                synonyms = [
                    lemma.name() for synset in synsets
                    for lemma in synset.lemmas()
                    if lemma.name() != word
                ]
                if synonyms:
                    new_words[idx] = np.random.choice(synonyms).replace('_', ' ')
                    replaced += 1

        return ' '.join(new_words)

    def eda_random_insertion(self, text, n=1):
        """EDA: 在随机位置插入同义词"""
        import nltk
        from nltk.corpus import wordnet

        words = text.split()
        new_words = words.copy()

        for _ in range(n):
            # 查找随机单词的同义词
            word = np.random.choice(words)
            synsets = wordnet.synsets(word)

            if synsets:
                synonyms = [
                    lemma.name() for synset in synsets
                    for lemma in synset.lemmas()
                ]
                if synonyms:
                    synonym = np.random.choice(synonyms).replace('_', ' ')
                    insert_pos = np.random.randint(0, len(new_words) + 1)
                    new_words.insert(insert_pos, synonym)

        return ' '.join(new_words)

    def eda_random_swap(self, text, n=1):
        """EDA: 随机交换单词"""
        words = text.split()
        if len(words) < 2:
            return text

        new_words = words.copy()
        for _ in range(n):
            i, j = np.random.choice(len(new_words), 2, replace=False)
            new_words[i], new_words[j] = new_words[j], new_words[i]

        return ' '.join(new_words)

    def eda_random_deletion(self, text, p=0.1):
        """EDA: 随机删除单词"""
        words = text.split()
        if len(words) == 1:
            return text

        new_words = [
            word for word in words
            if np.random.random() > p
        ]

        return ' '.join(new_words) if new_words else np.random.choice(words)

    def back_translation(self, text, src_lang='en', pivot_lang='fr'):
        """
        回译增强: 英语 -> 法语 -> 英语
        在保持语义的同时使表达多样化
        (实际实现中使用翻译 API 或模型)
        """
        # 示例: 使用 Helsinki-NLP/opus-mt 模型
        try:
            from transformers import pipeline

            # 英语 -> 中转语言
            translator_fwd = pipeline(
                f"translation_{src_lang}_to_{pivot_lang}",
                model=f"Helsinki-NLP/opus-mt-{src_lang}-{pivot_lang}"
            )

            # 中转语言 -> 英语
            translator_bwd = pipeline(
                f"translation_{pivot_lang}_to_{src_lang}",
                model=f"Helsinki-NLP/opus-mt-{pivot_lang}-{src_lang}"
            )

            # 执行翻译
            pivot_text = translator_fwd(text)[0]['translation_text']
            back_translated = translator_bwd(pivot_text)[0]['translation_text']

            return back_translated

        except Exception as e:
            print(f"翻译错误: {e}")
            return text

    def augment_dataset(self, texts, labels, n_aug=4):
        """整个数据集的增强"""
        augmented_texts = []
        augmented_labels = []

        for text, label in zip(texts, labels):
            augmented_texts.append(text)
            augmented_labels.append(label)

            for _ in range(n_aug):
                aug_type = np.random.choice(
                    ['synonym', 'insertion', 'swap', 'deletion']
                )

                if aug_type == 'synonym':
                    aug_text = self.eda_synonym_replacement(text)
                elif aug_type == 'insertion':
                    aug_text = self.eda_random_insertion(text)
                elif aug_type == 'swap':
                    aug_text = self.eda_random_swap(text)
                else:
                    aug_text = self.eda_random_deletion(text)

                augmented_texts.append(aug_text)
                augmented_labels.append(label)

        print(f"原始样本数: {len(texts)}")
        print(f"增强后样本数: {len(augmented_texts)}")

        return augmented_texts, augmented_labels

import torch
import torchvision.transforms as transforms

def get_randaugment_transforms(n=2, m=9, image_size=224):
    """
    RandAugment: 随机增强策略
    Cubuk et al., "RandAugment: Practical Automated Data Augmentation" (2019)

    Args:
        n: 要应用的增强操作数量
        m: 增强强度(0-30)
    """
    # 使用 PyTorch 内置的 RandAugment(torchvision >= 0.12)
    transform = transforms.Compose([
        transforms.RandomResizedCrop(image_size),
        transforms.RandAugment(num_ops=n, magnitude=m),
        transforms.ToTensor(),
        transforms.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        )
    ])

    return transform


def specaugment_for_audio(spectrogram, freq_mask_param=27, time_mask_param=70):
    """
    SpecAugment: 音频声谱图增强
    Park et al., "SpecAugment: A Simple Data Augmentation Method
    for Automatic Speech Recognition" (2019)

    Args:
        spectrogram: 输入声谱图 (freq, time)
        freq_mask_param: 频率遮罩的最大尺寸
        time_mask_param: 时间遮罩的最大尺寸
    """
    import torch
    import torchaudio.transforms as T

    freq_mask = T.FrequencyMasking(freq_mask_param=freq_mask_param)
    time_mask = T.TimeMasking(time_mask_param=time_mask_param)

    # 频率遮罩
    augmented = freq_mask(spectrogram)
    # 时间遮罩
    augmented = time_mask(augmented)

    return augmented

class SyntheticTextGenerator:
    """
    使用 LLM 生成合成训练数据
    """

    def __init__(self, llm_client, model_name='gpt-4'):
        self.llm = llm_client
        self.model_name = model_name

    def generate_classification_data(self, class_name, n_samples=100,
                                     domain='general', style='diverse'):
        """
        为分类任务生成合成数据

        Args:
            class_name: 要生成的类别名称
            n_samples: 要生成的样本数
            domain: 领域(medical, legal 等)
            style: 风格(formal, casual, diverse)
        """
        prompt = f"""Generate {n_samples} diverse text examples for the class '{class_name}'.
Domain: {domain}
Style: {style}

Requirements:
- Each example should be 1-3 sentences
- Vary the vocabulary, sentence structure, and perspective
- Include both simple and complex cases
- Format as a JSON list: ["example1", "example2", ...]

Generate realistic examples that would appear in real-world {domain} data."""

        # 调用 LLM(实际实现中使用 API)
        # response = self.llm.complete(prompt)
        # examples = json.loads(response)

        print(f"'{class_name}' 类别的合成数据生成提示词:")
        print(prompt[:300] + "...")
        print(f"\n计划生成 {n_samples} 个样本")

    def generate_edge_cases(self, class_examples, n_edge_cases=20):
        """生成边界情形的合成数据"""
        prompt = f"""Based on these training examples:
{chr(10).join(class_examples[:5])}

Generate {n_edge_cases} challenging edge cases that:
1. Are ambiguous between different categories
2. Contain misleading keywords
3. Have unusual sentence structures
4. Test the model's true understanding

Format as JSON list."""

        print("边界情形生成提示词准备完成")

    def augment_with_paraphrase(self, texts, n_paraphrases=3):
        """
        用 LLM 生成释义(paraphrase)
        """
        augmented = []

        for text in texts:
            prompt = f"""Paraphrase the following text {n_paraphrases} times.
Keep the same meaning but use different words and sentence structures.

Original: "{text}"

Format as JSON list of {n_paraphrases} paraphrases."""

            augmented.append({
                'original': text,
                'paraphrases': []  # 由 LLM 响应填充
            })

        return augmented


class SyntheticImageGenerator:
    """
    使用 Diffusion Model 生成合成图像
    """

    def __init__(self, model_name='stabilityai/stable-diffusion-2-1'):
        self.model_name = model_name

    def setup_pipeline(self):
        """
        初始化 Stable Diffusion 流水线
        pip install diffusers accelerate
        """
        try:
            from diffusers import StableDiffusionPipeline
            import torch

            self.pipe = StableDiffusionPipeline.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16
            )

            if torch.cuda.is_available():
                self.pipe = self.pipe.to('cuda')

            print(f"流水线初始化完成: {self.model_name}")
        except Exception as e:
            print(f"流水线初始化错误: {e}")

    def generate_class_images(self, class_name, n_images=50,
                               style_prompt="high quality, photorealistic"):
        """
        生成特定类别的合成图像

        Args:
            class_name: 要生成的类别
            n_images: 要生成的图像数
            style_prompt: 风格提示词
        """
        prompts = [
            f"A photo of {class_name}, {style_prompt}",
            f"{class_name} in natural environment, {style_prompt}",
            f"Close-up of {class_name}, detailed, {style_prompt}",
            f"{class_name} from different angle, {style_prompt}",
        ]

        print(f"'{class_name}' 合成图像生成计划:")
        print(f"要生成的图像数: {n_images}")
        print("使用的提示词示例:")
        for p in prompts[:2]:
            print(f"  - {p}")

        # 实际生成代码
        # images = []
        # for i in range(n_images):
        #     prompt = prompts[i % len(prompts)]
        #     image = self.pipe(prompt).images[0]
        #     images.append(image)
        # return images

    def evaluate_synthetic_quality(self, real_images, synthetic_images):
        """
        合成图像质量评估(FID 分数)
        """
        try:
            from torchmetrics.image.fid import FrechetInceptionDistance

            fid = FrechetInceptionDistance(feature=64)

            # 加入真实图像
            fid.update(real_images, real=True)
            # 加入合成图像
            fid.update(synthetic_images, real=False)

            fid_score = fid.compute()
            print(f"FID 分数: {fid_score:.2f}")
            print("(越低越好，0 为完美)")

            return fid_score
        except Exception as e:
            print(f"FID 计算错误: {e}")

class DataFlywheelPipeline:
    """
    数据飞轮实现流水线
    """

    def __init__(self, model, feedback_store):
        self.model = model
        self.feedback_store = feedback_store
        self.version = 0

    def collect_production_feedback(self, predictions, user_feedback):
        """
        收集生产环境中的用户反馈

        Args:
            predictions: 模型预测值
            user_feedback: 用户纠正/确认的数据
        """
        valuable_samples = []

        for pred, feedback in zip(predictions, user_feedback):
            if feedback['corrected']:
                # 用户纠正过的样本(错误案例)
                sample = {
                    'input': feedback['input'],
                    'model_prediction': pred,
                    'true_label': feedback['correction'],
                    'confidence': pred['confidence'],
                    'timestamp': feedback['timestamp'],
                    'value': 'high'  # 错误案例的价值更高
                }
                valuable_samples.append(sample)

            elif feedback['confirmed'] and pred['confidence'] < 0.7:
                # 置信度低但预测正确的样本
                sample = {
                    'input': feedback['input'],
                    'true_label': pred['label'],
                    'confidence': pred['confidence'],
                    'value': 'medium'
                }
                valuable_samples.append(sample)

        print(f"收集到的有价值样本: {len(valuable_samples)}")
        return valuable_samples

    def prioritize_labeling_queue(self, unlabeled_pool, budget):
        """
        确定标注优先级

        优先级判定标准：
        1. 模型不确定性(越高越优先)
        2. 类别稀有度(越稀有越优先)
        3. 数据多样性(与既有数据差异越大越优先)
        """
        priorities = []

        for sample in unlabeled_pool:
            score = 0

            # 不确定性分数
            uncertainty = 1 - max(sample['predicted_probs'])
            score += uncertainty * 0.5

            # 类别稀有度分数
            predicted_class = max(sample['predicted_probs'],
                                  key=sample['predicted_probs'].get)
            rarity = 1 / (sample['class_counts'].get(predicted_class, 1) + 1)
            score += rarity * 0.3

            # 多样性分数(简单近似)
            diversity = np.std(sample['predicted_probs'].values())
            score += diversity * 0.2

            priorities.append((sample, score))

        # 按优先级排序
        priorities.sort(key=lambda x: x[1], reverse=True)

        # 在预算内选择
        selected = [s for s, _ in priorities[:budget]]

        return selected

    def continuous_model_improvement(self, new_data, evaluation_set):
        """
        持续模型改进循环
        """
        metrics_history = []

        while True:  # 实际使用中需加入终止条件
            # 1. 收集新数据
            new_samples = self.collect_production_feedback(
                # 实际实现中使用实时数据流
                predictions=[],
                user_feedback=[]
            )

            if len(new_samples) < 100:  # 最小样本数
                continue

            # 2. 数据质量检查
            checker = DataQualityChecker()
            # quality_scores = checker.compute_data_quality_score(...)

            # 3. 模型重新训练
            # self.model.finetune(new_samples)

            # 4. 评估与 A/B 测试
            # metrics = evaluate_model(self.model, evaluation_set)
            # metrics_history.append(metrics)

            self.version += 1
            print(f"模型版本 {self.version} 训练完成")

            break  # 演示用

        return metrics_history

import hashlib
import json
import os
from pathlib import Path
from datetime import datetime

class ReproducibleDataPipeline:
    """
    可复现的数据流水线
    - 追踪所有处理步骤
    - 用数据哈希验证完整性
    - 支持版本管理
    """

    def __init__(self, pipeline_name, base_dir='data/processed'):
        self.pipeline_name = pipeline_name
        self.base_dir = Path(base_dir)
        self.steps = []
        self.metadata = {
            'pipeline': pipeline_name,
            'created_at': datetime.now().isoformat(),
            'steps': []
        }

    def add_step(self, step_name, func, *args, **kwargs):
        """添加处理步骤"""
        self.steps.append({
            'name': step_name,
            'func': func,
            'args': args,
            'kwargs': kwargs
        })

    def compute_hash(self, data):
        """计算数据哈希"""
        if isinstance(data, np.ndarray):
            return hashlib.md5(data.tobytes()).hexdigest()
        elif isinstance(data, (list, dict)):
            return hashlib.md5(
                json.dumps(data, sort_keys=True, default=str).encode()
            ).hexdigest()
        else:
            return hashlib.md5(str(data).encode()).hexdigest()

    def run(self, input_data):
        """执行流水线"""
        data = input_data

        for step in self.steps:
            print(f"正在执行: {step['name']}")

            # 处理前的哈希
            hash_before = self.compute_hash(data)

            # 执行处理
            data = step['func'](data, *step['args'], **step['kwargs'])

            # 处理后的哈希
            hash_after = self.compute_hash(data)

            # 记录元数据
            self.metadata['steps'].append({
                'name': step['name'],
                'hash_before': hash_before,
                'hash_after': hash_after,
                'timestamp': datetime.now().isoformat()
            })

            print(f"  完成: {hash_before[:8]} -> {hash_after[:8]}")

        # 保存元数据
        metadata_path = self.base_dir / f"{self.pipeline_name}_metadata.json"
        metadata_path.parent.mkdir(parents=True, exist_ok=True)
        with open(metadata_path, 'w') as f:
            json.dump(self.metadata, f, indent=2)

        print(f"\n流水线完成。元数据: {metadata_path}")
        return data


class DataVersionControl:
    """
    DVC 风格的数据版本管理
    (实际使用中建议采用 dvc.org)
    """

    def __init__(self, storage_path='data/.dvc'):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(parents=True, exist_ok=True)

    def add(self, data_path):
        """开始追踪数据文件"""
        data_path = Path(data_path)

        # 计算哈希
        with open(data_path, 'rb') as f:
            file_hash = hashlib.md5(f.read()).hexdigest()

        # 生成 .dvc 元文件
        dvc_file = data_path.with_suffix('.dvc')
        dvc_metadata = {
            'md5': file_hash,
            'size': os.path.getsize(data_path),
            'path': str(data_path.name),
            'version': datetime.now().isoformat()
        }

        with open(dvc_file, 'w') as f:
            json.dump(dvc_metadata, f, indent=2)

        print(f"开始追踪: {data_path}")
        print(f"  MD5: {file_hash}")
        print(f"  元文件: {dvc_file}")

        return file_hash

    def create_data_contract(self, schema):
        """
        定义数据契约(Data Contract)
        - 定义 schema
        - 质量标准
        - SLA 要求
        """
        contract = {
            'version': '1.0',
            'schema': schema,
            'quality_rules': {
                'completeness': {'min_threshold': 0.99},
                'accuracy': {'label_error_rate': {'max': 0.05}},
                'consistency': {'duplicate_rate': {'max': 0.01}},
            },
            'sla': {
                'update_frequency': 'daily',
                'max_staleness_hours': 24,
            }
        }

        return contract


def demonstrate_full_data_pipeline():
    """
    数据中心 AI 完整流水线演示
    """
    print("=" * 60)
    print("数据中心 AI 流水线演示")
    print("=" * 60)

    # 1. 数据质量检查
    print("\n第 1 步: 数据质量检查")
    print("  - 测量标签错误率")
    print("  - 检测异常值")
    print("  - 去除重复样本")
    print("  - 分析类别不均衡")

    # 2. 标签清洗
    print("\n第 2 步: 标签清洗")
    print("  - 用 Cleanlab 检测错误标签")
    print("  - 通过多数决/专家复核进行修正")
    print("  - 提升标注者间一致度")

    # 3. 数据增强
    print("\n第 3 步: 数据增强")
    print("  - 图像: Albumentations")
    print("  - 文本: EDA, 回译")
    print("  - 探索自动增强策略")

    # 4. 合成数据
    print("\n第 4 步: 合成数据生成")
    print("  - 用 LLM 合成文本")
    print("  - 用 Diffusion Model 合成图像")
    print("  - 质量过滤(FID、分类器置信度)")

    # 5. 主动学习
    print("\n第 5 步: 主动学习")
    print("  - 用不确定性采样确定标注优先级")
    print("  - 用 CoreSet 方法保证多样性")

    # 6. 流水线版本管理
    print("\n第 6 步: 版本管理与监控")
    print("  - 用 DVC 进行数据版本管理")
    print("  - 用数据契约维持质量标准")
    print("  - 用数据飞轮持续改进")

    print("\n结论: 在很多情况下，改进数据质量比改进模型更有效！")