AI水印技术对比：追踪AI生成内容的数字指纹

class VocabularyBiasWatermark:
    def __init__(self, secret_key):
        self.secret_key = secret_key
        self.bias_strength = 0.1
    
    def watermark_logits(self, logits, token_ids, position):
        """对logits施加水印偏向"""
        # 使用密钥和位置生成伪随机偏向
        rng = np.random.RandomState(seed=hash((self.secret_key, position)))
        bias = rng.randn(len(logits)) * self.bias_strength
        
        # 施加偏向
        watermarked_logits = logits + bias
        
        return watermarked_logits
    
    def detect_watermark(self, text, secret_key):
        """检测水印"""
        tokens = tokenize(text)
        bias_correlation = []
        
        for i, token in enumerate(tokens):
            # 重现伪随机序列
            rng = np.random.RandomState(seed=hash((secret_key, i)))
            expected_bias = rng.randn(vocab_size)
            
            # 检查实际token是否符合期望偏向
            correlation = np.corrcoef([expected_bias], [one_hot(token, vocab_size)])[0, 1]
            bias_correlation.append(correlation)
        
        # 统计检验
        avg_correlation = np.mean(bias_correlation)
        p_value = self.statistical_test(bias_correlation)
        
        return {
            "has_watermark": p_value < 0.01,
            "confidence": 1 - p_value,
            "correlation": avg_correlation
        }

class SyntaxWatermark:
    def __init__(self, watermark_rules):
        self.rules = watermark_rules  # 语法偏好规则
        
    def apply_watermark(self, generation_options):
        """应用水印到生成选项"""
        # 根据水印规则调整不同语法结构的采样概率
        for structure, preference in self.rules.items():
            generation_options[structure] *= preference
        
        return generation_options

class FrequencyDomainWatermark:
    def __init__(self, watermark_data):
        self.watermark = watermark_data  # 二进制水印数据
    
    def embed(self, image):
        """嵌入水印"""
        # 1. 变换到频域
        freq_domain = dct_transform(image)
        
        # 2. 选择中频分量嵌入（高频易丢失，低频易感知）
        mid_freq = freq_domain[mid_freq_range]
        
        # 3. 修改中频分量
        for i, bit in enumerate(self.watermark):
            if bit == 1:
                mid_freq[i] = abs(mid_freq[i]) * 1.01  # 微小修改
            else:
                mid_freq[i] = -abs(mid_freq[i]) * 1.01
        
        # 4. 逆变换回空域
        watermarked = inverse_dct(freq_domain)
        
        return watermarked
    
    def detect(self, image):
        """检测水印"""
        freq_domain = dct_transform(image)
        mid_freq = freq_domain[mid_freq_range]
        
        # 提取比特
        extracted = []
        for coeff in mid_freq:
            if coeff > 0:
                extracted.append(1)
            else:
                extracted.append(0)
        
        # 验证水印
        return self.verify_watermark(extracted)

class DiffusionWatermark:
    def __init__(self, watermark_key):
        self.key = watermark_key
    
    def watermark_latent(self, latent, strength=0.05):
        """在扩散过程的隐空间嵌入水印"""
        # 生成与密钥相关的扰动模式
        rng = np.random.RandomState(seed=hash(self.key))
        perturbation = rng.randn(*latent.shape) * strength
        
        # 将扰动添加到隐向量
        watermarked_latent = latent + perturbation
        
        return watermarked_latent
    
    def detect_from_image(self, image, diffusion_model):
        """从生成的图像检测水印"""
        # 1. 逆向扩散到隐空间
        latent = diffusion_model.invert(image)
        
        # 2. 提取扰动模式
        perturbation = latent - self.estimate_original_latent(latent)
        
        # 3. 与密钥生成的模式比对
        rng = np.random.RandomState(seed=hash(self.key))
        expected_perturbation = rng.randn(*latent.shape) * 0.05
        
        correlation = np.corrcoef(perturbation.flatten(), expected_perturbation.flatten())[0, 1]
        
        return {
            "has_watermark": correlation > 0.5,
            "confidence": abs(correlation)
        }

class VideoWatermark:
    def embed_temporal_watermark(self, video_frames, watermark_bits):
        """时域水印：通过帧间关系编码水印"""
        watermarked_frames = []
        
        for i, frame in enumerate(video_frames):
            if i > 0:
                # 根据水印比特修改帧间差异
                bit = watermark_bits[i % len(watermark_bits)]
                if bit == 1:
                    # 增强与上一帧的差异
                    frame = self.enhance_difference(frame, video_frames[i-1])
                else:
                    # 减弱与上一帧的差异
                    frame = self.reduce_difference(frame, video_frames[i-1])
            
            watermarked_frames.append(frame)
        
        return watermarked_frames

class WatermarkingStrategy:
    def __init__(self):
        self.strategy = {
            "high_stakes": "strong_watermark",  # 高风险内容，强水印
            "commercial": "commercial_watermark",  # 商业内容，包含版权信息
            "low_stakes": "light_watermark",  # 低风险内容，轻微水印
        }
    
    def select_watermark(self, content_type, user_tier, destination):
        """选择水印策略"""
        if destination == "public" and content_type == "image":
            return "strong_watermark"
        elif user_tier == "premium":
            return "premium_watermark"  # 包含用户ID
        else:
            return "default_watermark"