AI 音频内容治理框架：从检测到水印、从平台政策到法律合规的完整指南

import librosa
import numpy as np
import torch
import torch.nn as nn
from pathlib import Path

class AIAudioDetector:
    """基于声学特征和深度学习分类器的 AI 音频检测器"""
    
    def __init__(self, model_path: str = "models/audio_detector_v3.pt"):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.classifier = self._load_classifier(model_path)
        
    def _load_classifier(self, path: str) -> nn.Module:
        """加载预训练的 AI 音频分类器"""
        model = nn.Sequential(
            nn.Conv1d(in_channels=128, out_channels=256, kernel_size=3),
            nn.ReLU(),
            nn.MaxPool1d(2),
            nn.Conv1d(256, 512, kernel_size=3),
            nn.ReLU(),
            nn.AdaptiveAvgPool1d(1),
            nn.Flatten(),
            nn.Linear(512, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 1),
            nn.Sigmoid()
        )
        model.load_state_dict(torch.load(path, map_location=self.device))
        model.to(self.device)
        model.eval()
        return model
    
    def extract_features(self, audio_path: str) -> np.ndarray:
        """提取音频的 Mel 频谱特征"""
        y, sr = librosa.load(audio_path, sr=22050)
        
        # 提取 128 维 Mel 频谱图
        mel_spec = librosa.feature.melspectrogram(
            y=y, sr=sr, n_mels=128, 
            hop_length=512, n_fft=2048
        )
        
        # 转换为对数刻度
        log_mel = librosa.power_to_db(mel_spec)
        
        # 归一化到 [0, 1]
        log_mel = (log_mel - log_mel.min()) / (log_mel.max() - log_mel.min() + 1e-8)
        
        return log_mel.astype(np.float32)
    
    def analyze_spectral_artifacts(self, audio_path: str) -> dict:
        """分析频谱伪影——AI 生成的常见痕迹"""
        y, sr = librosa.load(audio_path, sr=22050)
        
        # 高频能量比率（AI 语音通常在高频区域能量异常）
        spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
        high_freq_ratio = np.mean(spectral_centroid > sr * 0.35)
        
        # 谐波-打击乐分离（真实语音的谐波结构更自然）
        harmonic, percussive = librosa.effects.hpss(y)
        harmonic_ratio = np.mean(np.abs(harmonic)) / (np.mean(np.abs(percussive)) + 1e-8)
        
        # 基频连续性（AI 语音的基频变化可能过于平滑）
        f0, voiced_flag, _ = librosa.pyin(
            y, fmin=librosa.note_to_hz('C2'),
            fmax=librosa.note_to_hz('C7'), sr=sr
        )
        f0_valid = f0[~np.isnan(f0)]
        f0_smoothness = np.std(np.diff(f0_valid)) if len(f0_valid) > 1 else 0
        
        return {
            "high_freq_ratio": float(high_freq_ratio),
            "harmonic_ratio": float(harmonic_ratio),
            "f0_smoothness": float(f0_smoothness),
            "duration_sec": float(len(y) / sr)
        }
    
    def detect(self, audio_path: str) -> dict:
        """执行完整的 AI 音频检测流程"""
        # 1. 提取特征
        features = self.extract_features(audio_path)
        
        # 2. 深度学习分类器预测
        features_tensor = torch.tensor(features).unsqueeze(0).to(self.device)
        with torch.no_grad():
            ai_probability = float(self.classifier(features_tensor).item())
        
        # 3. 声学特征分析
        artifacts = self.analyze_spectral_artifacts(audio_path)
        
        # 4. 综合判断
        # 结合分类器输出和声学特征进行综合判断
        score = ai_probability
        if artifacts["f0_smoothness"] < 0.5:
            score = score * 0.9  # 基频变化自然，降低 AI 概率
        if artifacts["high_freq_ratio"] > 0.7:
            score = score * 1.1  # 高频异常，提高 AI 概率
        
        verdict = "AI_GENERATED" if score > 0.7 else (
            "LIKELY_AI" if score > 0.5 else (
                "LIKELY_HUMAN" if score > 0.3 else "HUMAN"
            )
        )
        
        return {
            "ai_probability": round(score, 4),
            "verdict": verdict,
            "confidence": abs(score - 0.5) * 2,  # 距离 0.5 越远，置信度越高
            "artifacts": artifacts,
            "recommendation": self._get_recommendation(verdict, score)
        }
    
    def _get_recommendation(self, verdict: str, score: float) -> str:
        if verdict == "AI_GENERATED":
            return "建议标记为 AI 生成内容"
        elif verdict == "LIKELY_AI":
            return "建议进一步人工审核"
        elif verdict == "LIKELY_HUMAN":
            return "可能为真人录制，但建议保留检测记录"
        else:
            return "高度可信为真人录制"

import torch
import torchaudio
from pathlib import Path

class AudioSealWatermarkDetector:
    """检测 AI 音频中是否嵌入了 AudioSeal 水印"""
    
    def __init__(
        self, 
        detector_path: str = "models/audio_seal_detector.pth",
        sample_rate: int = 16000
    ):
        self.sample_rate = sample_rate
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.detector = self._load_detector(detector_path)
    
    def _load_detector(self, path: str):
        """加载 AudioSeal 水印检测器"""
        # AudioSeal 使用轻量级 CNN 检测器
        # 输入：音频波形，输出：每帧的水印存在概率
        from torch import nn
        
        class SimpleDetector(nn.Module):
            def __init__(self):
                super().__init__()
                self.encoder = nn.Sequential(
                    nn.Conv1d(1, 32, kernel_size=31, stride=1, padding=15),
                    nn.LeakyReLU(0.2),
                    nn.Conv1d(32, 64, kernel_size=15, stride=2, padding=7),
                    nn.LeakyReLU(0.2),
                    nn.Conv1d(64, 128, kernel_size=7, stride=2, padding=3),
                    nn.LeakyReLU(0.2),
                    nn.Conv1d(128, 1, kernel_size=3, stride=1, padding=1),
                    nn.Sigmoid()
                )
            
            def forward(self, x):
                return self.encoder(x)
        
        model = SimpleDetector()
        model.load_state_dict(torch.load(path, map_location=self.device))
        model.to(self.device)
        model.eval()
        return model
    
    def check_watermark(self, audio_path: str, threshold: float = 0.6) -> dict:
        """检查音频中是否存在 AudioSeal 水印
        
        Args:
            audio_path: 音频文件路径
            threshold: 水印检测阈值（默认 0.6）
            
        Returns:
            水印检测结果
        """
        # 加载音频并归一化
        waveform, sr = torchaudio.load(audio_path)
        if sr != self.sample_rate:
            waveform = torchaudio.functional.resample(
                waveform, orig_freq=sr, new_freq=self.sample_rate
            )
        
        # 确保单声道
        if waveform.shape[0] > 1:
            waveform = waveform.mean(dim=0, keepdim=True)
        
        # 截断或填充到固定长度（5 秒）
        target_len = self.sample_rate * 5
        if waveform.shape[1] < target_len:
            waveform = torch.nn.functional.pad(
                waveform, (0, target_len - waveform.shape[1])
            )
        else:
            waveform = waveform[:, :target_len]
        
        # 运行检测器
        waveform = waveform.to(self.device)
        with torch.no_grad():
            # 输出形状: (1, 1, num_frames)
            watermark_scores = self.detector(waveform)
        
        # 分析检测结果
        max_score = float(watermark_scores.max())
        mean_score = float(watermark_scores.mean())
        frame_count = watermark_scores.shape[2]
        positive_frames = int((watermark_scores > threshold).sum())
        positive_ratio = positive_frames / frame_count
        
        # 判定水印是否存在
        has_watermark = positive_ratio > 0.3  # 超过 30% 的帧检测到水印
        
        # 解码水印信息（简化版——实际应使用 AudioSeal 的解码器）
        watermark_info = None
        if has_watermark:
            # 这里应该调用 AudioSeal 的 message_decoder
            # 返回生成者 ID、时间戳等信息
            watermark_info = {
                "detected": True,
                "positive_ratio": round(positive_ratio, 4),
                "max_confidence": round(max_score, 4),
                "note": "需要 AudioSeal message_decoder 解码完整水印信息"
            }
        
        return {
            "has_watermark": has_watermark,
            "watermark_info": watermark_info,
            "detection_summary": {
                "total_frames": frame_count,
                "positive_frames": positive_frames,
                "positive_ratio": round(positive_ratio, 4),
                "mean_score": round(mean_score, 4),
                "max_score": round(max_score, 4)
            },
            "interpretation": (
                "检测到水印——此音频来自支持 AudioSeal 的 AI 生成工具"
                if has_watermark else
                "未检测到水印——可能来自不支持水印的工具，或水印已被移除"
            )
        }

import numpy as np
import soundfile as sf
from typing import Tuple
import hashlib
import struct

class SimpleAudioWatermarker:
    """简化的音频水印系统——用于教学和理解原理
    注意：生产环境应使用 AudioSeal 等专业方案"""
    
    def __init__(self, sample_rate: int = 44100):
        self.sample_rate = sample_rate
        # 水印嵌入强度——值越大水印越鲁棒但对音质影响越大
        self.watermark_strength = 0.001
    
    def _generate_watermark_signal(
        self, message: bytes, audio_length: int
    ) -> np.ndarray:
        """生成水印信号——使用扩展频谱技术
        
        Args:
            message: 要嵌入的消息字节
            audio_length: 音频样本总数
            
        Returns:
            水印信号数组（与音频等长）
        """
        # 将消息转换为比特流
        bits = []
        for byte in message:
            for i in range(7, -1, -1):
                bits.append((byte >> i) & 1)
        
        # 使用直接序列扩频（DSSS）
        # 每个比特用一段伪随机序列表示
        chip_rate = 1000  # 码片速率（每秒码片数）
        total_chips = int(audio_length / self.sample_rate * chip_rate)
        
        # 使用消息的哈希作为伪随机序列的种子
        seed = int(hashlib.sha256(message).hexdigest()[:8], 16)
        rng = np.random.RandomState(seed)
        
        # 生成扩频序列
        spreading_sequence = rng.choice([-1, 1], size=total_chips)
        
        # 将比特流映射到扩频序列
        chip_per_bit = total_chips // len(bits)
        watermark_chips = np.zeros(total_chips)
        for i, bit in enumerate(bits):
            start = i * chip_per_bit
            end = start + chip_per_bit
            if bit == 1:
                watermark_chips[start:end] = spreading_sequence[start:end]
            else:
                watermark_chips[start:end] = -spreading_sequence[start:end]
        
        # 上采样到音频采样率
        watermark_signal = np.repeat(
            watermark_chips, 
            self.sample_rate // chip_rate
        )
        
        # 截断或填充到音频长度
        if len(watermark_signal) > audio_length:
            watermark_signal = watermark_signal[:audio_length]
        else:
            watermark_signal = np.pad(
                watermark_signal, 
                (0, audio_length - len(watermark_signal))
            )
        
        return watermark_signal
    
    def embed_watermark(
        self, audio: np.ndarray, message: bytes
    ) -> Tuple[np.ndarray, dict]:
        """将水印嵌入音频
        
        Args:
            audio: 音频样本数组（归一化到 [-1, 1]）
            message: 要嵌入的消息
            
        Returns:
            (带水印的音频, 水印元数据)
        """
        watermark = self._generate_watermark_signal(message, len(audio))
        
        # 嵌入水印（加法嵌入）
        watermarked = audio + self.watermark_strength * watermark
        
        # 裁剪到合法范围
        watermarked = np.clip(watermarked, -1.0, 1.0)
        
        # 计算水印质量指标
        mse = np.mean((watermarked - audio) ** 2)
        psnr = 10 * np.log10(1.0 / (mse + 1e-10))
        snr = 10 * np.log10(
            np.mean(audio ** 2) / (mse + 1e-10)
        )
        
        return watermarked, {
            "message_length_bytes": len(message),
            "psnr_db": round(psnr, 2),
            "snr_db": round(snr, 2),
            "strength": self.watermark_strength,
            "quality_assessment": (
                "优秀——PSNR > 40dB，人耳不可察觉"
                if psnr > 40 else
                "良好——PSNR > 30dB，极端听力可能察觉"
                if psnr > 30 else
                "注意——PSNR < 30dB，可能影响音质"
            )
        }
    
    def extract_watermark(
        self, watermarked_audio: np.ndarray, original_message: bytes
    ) -> dict:
        """从音频中提取并验证水印
        
        Args:
            watermarked_audio: 带水印的音频
            original_message: 原始消息（用于验证）
            
        Returns:
            水印验证结果
        """
        # 生成参考水印信号
        ref_watermark = self._generate_watermark_signal(
            original_message, len(watermarked_audio)
        )
        
        # 计算相关性
        correlation = np.correlate(
            watermarked_audio, ref_watermark, mode='full'
        )
        max_corr = np.max(np.abs(correlation))
        
        # 归一化
        audio_energy = np.sqrt(np.mean(watermarked_audio ** 2))
        watermark_energy = np.sqrt(np.mean(ref_watermark ** 2))
        normalized_corr = max_corr / (
            audio_energy * watermark_energy * len(watermarked_audio) + 1e-10
        )
        
        # 检测阈值
        threshold = 0.0001
        detected = normalized_corr > threshold
        
        return {
            "watermark_detected": detected,
            "correlation_score": round(float(normalized_corr), 8),
            "threshold": threshold,
            "confidence": round(
                min(1.0, normalized_corr / (threshold * 3)), 4
            ),
            "message_verified": (
                "水印验证通过——内容与声明一致"
                if detected else
                "水印验证失败——内容可能被修改或水印被移除"
            )
        }

# === 使用示例 ===
if __name__ == "__main__":
    watermarker = SimpleAudioWatermarker(sample_rate=44100)
    
    # 模拟音频信号（实际应从文件加载）
    t = np.linspace(0, 5, 44100 * 5)
    original_audio = np.sin(2 * np.pi * 440 * t) * 0.5  # 440Hz 正弦波
    
    # 嵌入水印信息
    message = b"AUDIOGEN-2026-05-04-ELEVENLABS"
    watermarked, metadata = watermarker.embed_watermark(
        original_audio, message
    )
    
    print(f"水印质量: {metadata['quality_assessment']}")
    print(f"PSNR: {metadata['psnr_db']} dB")
    
    # 验证水印
    result = watermarker.extract_watermark(watermarked, message)
    print(f"水印检测: {result['message_verified']}"