Stable Diffusion v1.5 企业合规实践：生成内容水印与版权元数据自动标注

1. 引言：当 AI 创意遇上企业合规

你的设计团队用 Stable Diffusion v1.5 快速生成了上百张营销海报，效率显著提升。但法务部门可能提出尖锐问题：'这些 AI 生成的图片，版权怎么算？万一被竞争对手盗用了，我们怎么证明是自家生成的？'

# watermark_embedder.py - 基于 DCT 的隐形水印嵌入与提取工具 import cv2 import numpy as np from PIL import Image import hashlib import json class InvisibleWatermarker: def __init__(self, company_id="YOUR_COMPANY_CODE"): self.company_id = company_id # 水印嵌入强度，值越小越隐形，但提取容错性越低 self.alpha = 0.02 def _generate_watermark_pattern(self, seed_info): """根据种子信息生成二值水印图案""" # 将公司 ID 和种子信息（如时间戳、用户 ID）组合并哈希 combined_str = f"{self.company_id}_{seed_info}" hash_hex = hashlib.md5(combined_str.encode()).hexdigest() # 将哈希值转换为 64 位二进制序列（8x8 矩阵） binary_str = bin(int(hash_hex[:16], 16))[2:].zfill(64) watermark = np.array([int(b) for b in binary_str]).reshape(8, 8) # 将 0/1 转换为 -1/1，便于嵌入 return watermark * 2 - 1 def embed(self, image_array, seed_info="default"): """将隐形水印嵌入到 RGB 图像的 Y 通道（亮度）""" if len(image_array.shape) == 3 and image_array.shape[2] == 3: # 转换为 YCrCb 颜色空间，在 Y 通道（亮度）嵌入水印 ycrcb = cv2.cvtColor(image_array, cv2.COLOR_RGB2YCrCb) y_channel = ycrcb[:, :, 0].astype(np.float32) else: # 灰度图直接处理 y_channel = image_array.astype(np.float32) height, width = y_channel.shape watermark = self._generate_watermark_pattern(seed_info) watermarked_y = y_channel.copy() for i in range(0, height - 7, 8): for j in range(0, width - 7, 8): block = y_channel[i:i+8, j:j+8] dct_block = cv2.dct(block) # 在中频区域嵌入水印（避开直流和最高频） # 这里选择 (2:6, 2:6) 的 4x4 区域 dct_block[2:6, 2:6] += self.alpha * watermark watermarked_block = cv2.idct(dct_block) watermarked_y[i:i+8, j:j+8] = watermarked_block if len(image_array.shape) == 3: ycrcb[:, :, 0] = np.clip(watermarked_y, 0, 255).astype(np.uint8) watermarked_rgb = cv2.cvtColor(ycrcb, cv2.COLOR_YCrCb2RGB) return watermarked_rgb else: return np.clip(watermarked_y, 0, 255).astype(np.uint8) def extract(self, image_array, original_seed_info): """从可能被处理过的图像中尝试提取水印""" # 提取逻辑是嵌入的逆过程，需要原始种子信息来生成对比图案 # 此处为简化示例，实际应用需更复杂的相关性检测 expected_watermark = self._generate_watermark_pattern(original_seed_info) # ... 实际的提取和比对算法 ... # 返回提取出的水印图案和与预期图案的相似度 return expected_watermark, 0.95 # 示例相似度 # 使用示例 if __name__ == "__main__": # 加载 SD 生成的图片 img = Image.open("sd_generated_image.png") img_array = np.array(img) watermarker = InvisibleWatermarker(company_id="ACME_CORP_2024") # seed_info 可以是用户 ID+ 时间戳，用于唯一标识此次生成 seed_info = "user123_20240320_142305" # 嵌入水印 watermarked_array = watermarker.embed(img_array, seed_info) watermarked_img = Image.fromarray(watermarked_array) watermarked_img.save("watermarked_image.png") print("水印嵌入完成。肉眼几乎无法察觉差异。")

# metadata_writer.py - 自动写入版权和生成信息到图片元数据 from PIL import Image, PngImagePlugin from datetime import datetime import json class MetadataWriter: def __init__(self, company_name="Your Company", license_url="https://example.com/license"): self.company_name = company_name self.license_url = license_url def add_metadata_to_image(self, image_path, output_path, generation_data): """ 将生成数据和版权信息写入图片元数据 :param generation_data: 字典，包含所有生成参数和上下文例如：{ "prompt": "a beautiful landscape", "negative_prompt": "blurry, lowres", "steps": 20, "guidance_scale": 7.5, "seed": 12345, "model": "stable-diffusion-v1-5-archive", "generator": "Comfy-Org/stable-diffusion-v1-5-archive", "generated_at": "2024-03-20T14:23:05Z", "generated_by": "user123", "workflow_id": "campaign_2024_spring_001" } """ img = Image.open(image_path) # 准备要写入的元数据 metadata = PngImagePlugin.PngInfo() # 1. 写入标准 EXIF/ITPC 字段（部分格式支持） # 对于 PNG，我们主要使用自定义的 tEXt 块 metadata.add_text("Software", f"SD v1.5 Archive - Enhanced by {self.company_name}") metadata.add_text("Copyright", f"Copyright (c) {datetime.now().year} {self.company_name}. All rights reserved.") metadata.add_text("License", self.license_url) # 2. 写入详细的生成参数（用于审计和复现） metadata.add_text("AI_Generation_Parameters", json.dumps(generation_data, ensure_ascii=False)) # 3. 写入一个简明的版权声明（供人工阅读） human_readable_note = f""" AI Generated Image - {self.company_name} Model: {generation_data.get('model', 'N/A')} Prompt: {generation_data.get('prompt', 'N/A')[:100]}... Generated: {generation_data.get('generated_at', 'N/A')} This image is generated by AI tools. Commercial use may require permission. """ metadata.add_text("Disclaimer", human_readable_note) # 保存带有新元数据的图片 img.save(output_path, pnginfo=metadata) print(f"元数据已写入：{output_path}") # 返回写入的元数据摘要，便于记录到数据库 return { "output_file": output_path, "generation_id": generation_data.get("workflow_id", "") + "_" + str(generation_data.get("seed", "")), "metadata_added": True } # 使用示例 if __name__ == "__main__": writer = MetadataWriter(company_name="Acme 创意工场", license_url="https://acme.com/ai-license") gen_data = { "prompt": "a red vintage car on a rainy street, cinematic lighting, ultra detailed, 35mm film", "negative_prompt": "lowres, blurry, extra fingers", "steps": 25, "guidance_scale": 7.5, "seed": 42424242, "model": "stable-diffusion-v1-5-archive", "generator": "Comfy-Org/stable-diffusion-v1-5-archive", "generated_at": datetime.utcnow().isoformat() + "Z", "generated_by": "designer_li", "workflow_id": "social_media_post_0320" } result = writer.add_metadata_to_image("watermarked_image.png", "final_image_with_metadata.png", gen_data) print("元数据摘要:", result)

# compliance_middleware.py - 一个简单的 Flask 中间件示例 from flask import Flask, request, jsonify, send_file import requests import io import json from datetime import datetime from watermark_embedder import InvisibleWatermarker from metadata_writer import MetadataWriter import logging import uuid import numpy as np app = Flask(__name__) # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 配置 SD_API_URL = "http://localhost:7860/sdapi/v1/txt2img" # 假设 SD 服务运行在本机 7860 端口 WATERMARKER = InvisibleWatermarker(company_id="ENTERPRISE_AI_2024") METADATA_WRITER = MetadataWriter(company_name="企业 AI 创意中心") # 内存中的审计日志（生产环境应使用数据库） audit_log = [] @app.route('/api/v1/generate', methods=['POST']) def generate_image(): """增强的生成接口，添加水印和元数据""" try: user_data = request.json user_id = user_data.get('user_id', 'anonymous') workflow_id = user_data.get('workflow_id', str(uuid.uuid4())[:8]) # 1. 记录审计日志（请求） generation_id = f"{workflow_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}" audit_entry = { "generation_id": generation_id, "user_id": user_id, "workflow_id": workflow_id, "request_time": datetime.now().isoformat(), "request_data": user_data.get('sd_params', {}) } audit_log.append(audit_entry) logger.info(f"收到生成请求：{generation_id} from {user_id}") # 2. 调用原始 SD API sd_payload = user_data.get('sd_params', {}) response = requests.post(SD_API_URL, json=sd_payload) if response.status_code != 200: return jsonify({"error": "SD 服务调用失败", "details": response.text}), 500 result = response.json() # 假设 SD API 返回 base64 编码的图片 import base64 image_data = base64.b64decode(result['images'][0].split(",", 1)[0]) # 3. 处理图片（水印 + 元数据） from PIL import Image img = Image.open(io.BytesIO(image_data)) img_array = np.array(img) # 嵌入隐形水印 seed_info_for_watermark = f"{user_id}_{generation_id}" watermarked_array = WATERMARKER.embed(img_array, seed_info_for_watermark) watermarked_img = Image.fromarray(watermarked_array) # 准备元数据 generation_metadata = { **sd_payload, "model": "stable-diffusion-v1-5-archive", "generated_at": datetime.utcnow().isoformat() + "Z", "generated_by": user_id, "workflow_id": workflow_id, "generation_id": generation_id, "watermark_seed": seed_info_for_watermark } # 将带水印的图片保存到内存，并添加元数据 final_img_buffer = io.BytesIO() METADATA_WRITER._add_metadata_directly(watermarked_img, final_img_buffer, generation_metadata) final_img_buffer.seek(0) # 4. 更新审计日志（响应） audit_entry["response_time"] = datetime.now().isoformat() audit_entry["status"] = "success" audit_entry["generation_id"] = generation_id logger.info(f"生成完成：{generation_id}") # 5. 返回最终图片 return send_file(final_img_buffer, mimetype='image/png', as_attachment=False, download_name=f"{generation_id}.png") except Exception as e: logger.error(f"处理请求时出错：{str(e)}", exc_info=True) return jsonify({"error": "内部服务器错误", "message": str(e)}), 500 # 辅助方法：直接为 PIL Image 对象添加元数据 def _add_metadata_directly(self, pil_image, output_buffer, generation_data): """为 MetadataWriter 类添加的方法，直接处理 PIL Image 对象""" metadata = PngImagePlugin.PngInfo() metadata.add_text("AI_Generation_Parameters", json.dumps(generation_data, ensure_ascii=False)) metadata.add_text("Copyright", f"Copyright (c) {datetime.now().year} {self.company_name}") metadata.add_text("License", self.license_url) pil_image.save(output_buffer, format='PNG', pnginfo=metadata) # 将辅助方法动态添加到类中（生产环境应整合到类定义里） METADATA_WRITER._add_metadata_directly = lambda img, buf, data: _add_metadata_directly(METADATA_WRITER, img, buf, data) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=False)

Stable Diffusion v1.5 企业合规实践：生成内容水印与版权元数据自动标注