数据是AI的燃料

# ========== 音频数据质检 Pipeline（只查不改）==========
# 把待质检音频放在 ./raw_audio 文件夹，修改路径即可运行
# 依赖安装：pip install pydub audioop-lts numpy
# 额外需要安装 ffmpeg 并加入系统 PATH：https://ffmpeg.org/download.html
# 核心边界：本脚本只输出质检报告，不修改任何原始音频文件
import os
import hashlib
from pathlib import Path
from collections import Counter, defaultdict
from pydub import AudioSegment
import numpy as np

# ---------- 配置参数（行业通用参考阈值）----------
INPUT_DIR   = "./raw_audio"       # 待质检音频目录
TARGET_SR   = 16000               # 期望采样率
TARGET_CH   = 1                   # 期望声道数
MIN_SEC     = 1.0                 # 最短合法时长（秒）
MAX_SEC     = 60.0                # 最长合法时长（秒）
MIN_DBFS    = -30.0               # 最低合法音量（低于视为过静音）
MAX_DBFS    = -0.5                # 最高合法音量（高于视为削波）
SNR_THRESH  = 10.0                # 信噪比阈值（低于视为高噪声）
DUP_THRESH  = 0.95                # 近重复相似度阈值
ACCEPT_EXT  = {".mp3", ".wav", ".flac", ".m4a", ".ogg", ".aac"}

def estimate_snr(seg: AudioSegment) -> float:
    """估算信噪比（dB）：信号能量 / 静音段能量"""
    samples = np.array(seg.get_array_of_samples()).astype(np.float32)
    if len(samples) == 0:
        return 0.0
    # 用滑动窗口找静音段（振幅 < 最大值 5% 的段）
    window   = len(samples) // 20
    max_amp  = np.max(np.abs(samples))
    if max_amp < 1e-6:
        return 0.0
    noise_chunks = []
    for i in range(0, len(samples) - window, window // 2):
        chunk = samples[i:i+window]
        if np.max(np.abs(chunk)) < max_amp * 0.05:
            noise_chunks.append(chunk)
    if not noise_chunks:
        return 100.0  # 没检测到静音段 → 认为 SNR 很高
    noise_power = np.mean(np.concatenate(noise_chunks) ** 2)
    signal_power = np.mean(samples ** 2) - noise_power
    if noise_power <= 0 or signal_power <= 0:
        return 0.0
    return 10 * np.log10(signal_power / noise_power)

def spectral_hash(seg: AudioSegment, hash_len: int = 64) -> str:
    """简易频谱哈希：对音频做 STFT → 取能量峰值的 0/1 序列"""
    samples = np.array(seg.set_frame_rate(2000).set_channels(1).get_array_of_samples()).astype(np.float32)
    if len(samples) == 0:
        return ""
    # 简单版：分帧 + 每帧 RMS 对比均值
    frame_size = len(samples) // hash_len
    if frame_size == 0:
        return ""
    rms_vals = [np.sqrt(np.mean(samples[i:i+frame_size]**2)) for i in range(0, len(samples)-frame_size, frame_size)]
    rms_arr  = np.array(rms_vals[:hash_len])
    avg      = np.mean(rms_arr)
    return "".join("1" if v >= avg else "0" for v in rms_arr)

def hamming_distance(h1: str, h2: str) -> int:
    return sum(c1 != c2 for c1, c2 in zip(h1, h2))

def quality_check_one(src_path: Path) -> dict:
    """单文件质检：只查不改，返回问题标签字典"""
    issues = {}
    try:
        seg = AudioSegment.from_file(str(src_path))
    except Exception as e:
        return {"status": "✗ 损坏", "issues": [f"无法解码: {e}"]}

    dur_sec   = len(seg) / 1000.0
    sr        = seg.frame_rate
    channels  = seg.channels
    dbfs      = seg.dBFS if seg.dBFS != float("-inf") else -100.0
    snr       = estimate_snr(seg)
    sp_hash   = spectral_hash(seg)

    # ① 格式/参数合规
    if sr != TARGET_SR:
        issues["参数不一致"] = f"采样率 {sr}Hz（期望 {TARGET_SR}Hz）"
    if channels != TARGET_CH:
        issues["参数不一致"] = issues.get("参数不一致", "") + f"；声道数 {channels}（期望 {TARGET_CH}）"

    # ② 时长异常
    if dur_sec < MIN_SEC:
        issues["时长过短"] = f"{dur_sec:.2f}s < {MIN_SEC}s"
    elif dur_sec > MAX_SEC:
        issues["时长过长"] = f"{dur_sec:.2f}s > {MAX_SEC}s"

    # ③ 音量/SNR
    if dbfs > MAX_DBFS:
        issues["削波风险"] = f"dBFS={dbfs:.1f} > {MAX_DBFS}"
    elif dbfs < MIN_DBFS:
        issues["过静音"] = f"dBFS={dbfs:.1f} < {MIN_DBFS}"
    if snr < SNR_THRESH:
        issues["高噪声"] = f"SNR={snr:.1f}dB < {SNR_THRESH}dB"

    return {
        "status": "✓ 合格" if not issues else "⚠ 有问题",
        "duration": f"{dur_sec:.2f}s",
        "sr": f"{sr}Hz",
        "ch": f"{channels}ch",
        "dbfs": f"{dbfs:.1f}",
        "snr": f"{snr:.1f}dB",
        "sp_hash": sp_hash,
        "issues": issues,
    }

def batch_qc(input_dir: str):
    """批量质检：打印报告，不修改文件"""
    in_path = Path(input_dir)
    if not in_path.exists():
        print(f"❌ 目录不存在: {input_dir}")
        return

    files = sorted([f for f in in_path.rglob("*")
                    if f.is_file() and f.suffix.lower() in ACCEPT_EXT])
    if not files:
        print(f"❌ 目录 {input_dir} 下没有可处理的音频文件")
        return

    print(f"🔍 开始质检，共 {len(files)} 个文件\n")
    fmt_counter = Counter()
    results     = {}
    fail_list   = []

    for f in files:
        fmt_counter[f.suffix.lower()] += 1
        r = quality_check_one(f)
        results[f.name] = r
        tag = "✓" if r["status"] == "✓ 合格" else "✗"
        print(f"  {tag} {f.name:40s} | {r.get('duration','-'):>8} | "
              f"{r.get('sr','-'):>8} | {r.get('dbfs','-'):>6} | "
              f"{r.get('snr','-'):>8} | {r['status']}")
        if r["status"] != "✓ 合格":
            fail_list.append((f.name, r))

    # ---------- 跨文件：近重复检测 ----------
    print(f"\n🔗 近重复检测（相似度阈值 > {DUP_THRESH}）：")
    dup_groups = []
    checked = set()
    names = list(results.keys())
    for i, a in enumerate(names):
        if a in checked or not results[a].get("sp_hash"):
            continue
        group = [a]
        for j in range(i+1, len(names)):
            b = names[j]
            if b in checked or not results[b].get("sp_hash"):
                continue
            h1, h2 = results[a]["sp_hash"], results[b]["sp_hash"]
            max_len = max(len(h1), len(h2))
            if max_len == 0:
                continue
            sim = 1 - hamming_distance(h1, h2) / max_len
            if sim >= DUP_THRESH:
                group.append(b)
                checked.add(b)
        if len(group) > 1:
            dup_groups.append(group)
            checked.add(a)
            print(f"  ⚠ 近重复组：{group}")

    # ---------- 汇总报告 ----------
    total   = len(files)
    ok_cnt  = sum(1 for r in results.values() if r["status"] == "✓ 合格")
    fail_cnt = total - ok_cnt
    print(f"\n{'='*60}")
    print(f"📊 质检汇总")
    print(f"{'='*60}")
    print(f"  总文件数：  {total}")
    print(f"  ✓ 合格：   {ok_cnt} ({ok_cnt/total*100:.1f}%)")
    print(f"  ⚠ 不合格： {fail_cnt} ({fail_cnt/total*100:.1f}%)")
    print(f"  格式分布：  {dict(fmt_counter)}")
    if dup_groups:
        print(f"  近重复组：  {len(dup_groups)} 组")
    if fail_list:
        print(f"\n📋 不合格文件明细：")
        for name, r in fail_list:
            print(f"    ✗ {name}")
            for k, v in r.get("issues", {}).items():
                print(f"       [{k}] {v}")
    print(f"\n✅ 质检完成（原始文件未被修改）")

# ===== 运行 =====
if __name__ == "__main__":
    batch_qc(INPUT_DIR)

# ========== 时间序列数据质检 Pipeline（只查不改）==========
# 把待质检 CSV 放在 ./raw_timeseries/ 文件夹，修改路径即可运行
# 依赖安装：pip install pandas numpy
# 核心边界：本脚本只输出质检报告，不修改任何原始数据
import pandas as pd
import numpy as np
from pathlib import Path
from collections import defaultdict

# ---------- 配置参数 ----------
INPUT_DIR   = "./raw_timeseries"     # 待质检 CSV 目录
TIME_COL    = "timestamp"            # 时间列名
VALUE_COLS  = ["value", "temperature", "power"]  # 数值列（按需改）
FREQ_HINT   = pd.Timedelta(minutes=5)  # 期望采样间隔（5 分钟）
OUTLIER_SIGMA = 3.0                  # 3σ 阈值
OUTLIER_IQR_K = 1.5                  # IQR 倍数
TREND_WIN   = 20                     # 滑动窗口大小

def load_ts(path):
    """读 CSV → 按时间列排序、设索引"""
    df = pd.read_csv(path, parse_dates=[TIME_COL])
    df = df.sort_values(TIME_COL).reset_index(drop=True)
    return df

# ==========================================================
# ① 缺失值检测：时间戳空洞 + 指标值 NaN
# ==========================================================
def check_missing(df):
    report = {}
    # 时间戳空洞（重采样后缺失的时间点）
    full_idx = pd.date_range(df[TIME_COL].min(), df[TIME_COL].max(), freq=FREQ_HINT)
    missing_ts = full_idx.difference(pd.DatetimeIndex(df[TIME_COL]))
    report['时间空洞'] = len(missing_ts)
    # 每列的 NaN
    for col in VALUE_COLS:
        if col in df.columns:
            report[f'{col} 缺失'] = int(df[col].isna().sum())
    return report

# ==========================================================
# ② 时间对齐：频率不一致 + 重复时间戳 + 漂移
# ==========================================================
def check_alignment(df):
    report = {}
    ts = pd.DatetimeIndex(df[TIME_COL])
    # 频率分布（相邻差值）
    diffs = ts.to_series().diff().dropna()
    freq_counts = diffs.value_counts().head(5)
    report['频率混入'] = len(freq_counts) - 1  # 多出来的非主频率
    # 重复时间戳
    report['重复时间戳'] = int(df.duplicated(subset=[TIME_COL]).sum())
    # 时间漂移（累计偏差）
    if len(diffs) > 10:
        drift = (diffs - FREQ_HINT).abs().sum()
        report['时间漂移'] = f"{drift.total_seconds()/60:.1f} 分钟"
    return report

# ==========================================================
# ③ 离群点：3σ + IQR + Z-score
# ==========================================================
def check_outlier(df):
    report = {}
    for col in VALUE_COLS:
        if col not in df.columns: continue
        vals = df[col].dropna()
        # 3σ
        mu, std = vals.mean(), vals.std()
        sigma_mask = (vals - mu).abs() > OUTLIER_SIGMA * std
        report[f'{col} 3σ'] = int(sigma_mask.sum())
        # IQR
        q1, q3 = vals.quantile(.25), vals.quantile(.75)
        iqr = q3 - q1
        iqr_mask = (vals < q1 - OUTLIER_IQR_K*iqr) | (vals > q3 + OUTLIER_IQR_K*iqr)
        report[f'{col} IQR'] = int(iqr_mask.sum())
        # Z-score
        z = (vals - mu) / std
        z_mask = z.abs() > 3.0
        report[f'{col} Z-score'] = int(z_mask.sum())
    return report

# ==========================================================
# ④ 趋势/周期异常：突变 + 周期破坏 + 振幅异常
# ==========================================================
def check_trend(df):
    report = {}
    for col in VALUE_COLS:
        if col not in df.columns: continue
        vals = df[col].dropna()
        # 突变跳跃（一阶差分绝对值超过滚动均值 5σ）
        diff = vals.diff().abs()
        roll = diff.rolling(TREND_WIN, min_periods=5)
        mu, std = roll.mean(), roll.std()
        jump_mask = diff > (mu + 5 * std)
        report[f'{col} 突变跳跃'] = int(jump_mask.sum())
        # 周期破坏（FFT 主峰偏移）
        if len(vals) > TREND_WIN * 2:
            fft = np.abs(np.fft.rfft(vals.values - vals.mean()))
            report[f'{col} 周期破坏'] = int((fft[1:] < fft.max() * 0.1).sum())
        # 振幅异常（滚动 std 突增）
        roll_std = vals.rolling(TREND_WIN).std()
        amp_mask = roll_std > roll_std.quantile(.99) * 3
        report[f'{col} 振幅异常'] = int(amp_mask.sum())
    return report

# ==========================================================
# 主流程
# ==========================================================
def batch_qc(input_dir):
    files = list(Path(input_dir).glob('*.csv'))
    if not files: print(f'⚠ {input_dir} 无 CSV'); return
    print("=" * 50)
    print(f"📈 时间序列质检报告 — {len(files)} 个文件")
    print("=" * 50)
    grand = defaultdict(int)
    for fp in files:
        df = load_ts(fp)
        print(f"
【{fp.name}】 {len(df)} 行")
        for name, fn in [('①缺失值', check_missing), ('②时间对齐', check_alignment),
                         ('③离群点', check_outlier), ('④趋势异常', check_trend)]:
            r = fn(df)
            print(f"  {name}: ", end='')
            items = [(k, v) for k, v in r.items() if v != 0]
            for k, v in items: grand[k] += v
            if items:
                print(' · '.join(f'{k}={v}' for k, v in items))
            else:
                print('✅ 通过')
    print(f"
📊 汇总: ", end='')
    print(' · '.join(f'{k}={v}' for k, v in grand.items()) or '✅ 全部通过')

if __name__ == "__main__":
    batch_qc(INPUT_DIR)