别再只会用PS加Logo了！聊聊图片、文本、数据库里那些‘看不见’的数字水印（附Python代码示例）

数字水印技术早已超越了简单的版权标识，它正在成为数据安全领域的隐形守护者。想象一下，你可以在不改变图片观感的情况下嵌入溯源信息，或是在普通文本中隐藏只有特定程序才能识别的身份标记——这种"隐形墨水"般的特性，让数字水印在知识产权保护、数据追踪等场景中展现出独特价值。本文将带你用Python实现三种主流数字水印技术，从图像LSB到文本零宽字符，再到数据库字段的隐形标记，每个方案都附带可直接运行的代码示例。

1. 图像水印：用Pillow玩转LSB隐写术

最低有效位(LSB)算法是图像水印的入门技术，其核心思想是利用人眼对像素细微变化不敏感的特性。一个24位RGB像素由红绿蓝三个通道组成，每个通道8位（取值0-255），修改最低位的0/1只会造成±1的颜色值变化——这种差异肉眼根本无法察觉。

from PIL import Image

def lsb_embed(image_path, watermark_text, output_path):
    img = Image.open(image_path)
    pixels = img.load()
    
    # 将水印文本转换为二进制
    binary_text = ''.join(format(ord(c), '08b') for c in watermark_text)
    binary_text += '00000000'  # 结束标记
    
    if len(binary_text) > img.width * img.height * 3:
        raise ValueError("水印信息超出图像容量")
    
    index = 0
    for x in range(img.width):
        for y in range(img.height):
            r, g, b = pixels[x, y]
            
            # 修改RGB通道的最低有效位
            if index < len(binary_text):
                r = (r & 0xFE) | int(binary_text[index])
                index += 1
            if index < len(binary_text):
                g = (g & 0xFE) | int(binary_text[index])
                index += 1
            if index < len(binary_text):
                b = (b & 0xFE) | int(binary_text[index])
                index += 1
            
            pixels[x, y] = (r, g, b)
    
    img.save(output_path)

# 使用示例
lsb_embed('original.jpg', 'SECRET2023', 'watermarked.jpg')

提取水印时需要反向操作：

def lsb_extract(image_path):
    img = Image.open(image_path)
    pixels = img.load()
    
    binary_text = ''
    for x in range(img.width):
        for y in range(img.height):
            r, g, b = pixels[x, y]
            binary_text += str(r & 1)
            binary_text += str(g & 1)
            binary_text += str(b & 1)
    
    # 按8位一组解码
    watermark = ''
    for i in range(0, len(binary_text), 8):
        byte = binary_text[i:i+8]
        if byte == '00000000':
            break
        watermark += chr(int(byte, 2))
    
    return watermark

LSB技术的优缺点对比 ：

提示：LSB适合需要快速实现且对鲁棒性要求不高的场景，如内部文档标记。对于重要版权保护，建议结合后文的DCT/DWT变换域技术。

2. 文本水印：零宽字符的隐形魔法

零宽字符(Zero-Width Characters)是Unicode中的特殊存在，它们不占显示宽度却可以携带信息。常见的零宽字符包括：

• U+200B (零宽度空格)
• U+200C (零宽度非连接符)
• U+200D (零宽度连接符)
• U+FEFF (零宽度不换行空格)

def embed_zwc(text, watermark):
    zwc_map = {
        '0': '\u200b',
        '1': '\u200c',
        ' ': '\u200d'
    }
    
    # 将水印转为二进制并用空格分隔
    binary = ' '.join(format(ord(c), '08b') for c in watermark)
    zwc_watermark = ''.join(zwc_map.get(c, '') for c in binary)
    
    # 在原文第三个字符后插入
    return text[:3] + zwc_watermark + text[3:]

def extract_zwc(text):
    zwc_reverse = {
        '\u200b': '0',
        '\u200c': '1',
        '\u200d': ' '
    }
    
    binary = []
    for char in text:
        if char in zwc_reverse:
            binary.append(zwc_reverse[char])
    
    binary_str = ''.join(binary)
    watermark = ''
    for byte in binary_str.split(' '):
        if byte:
            watermark += chr(int(byte, 2))
    
    return watermark

# 使用示例
marked_text = embed_zwc("重要合同内容", "AUTHOR:张三")
print(f"含水印文本长度：{len(marked_text)}")  # 肉眼看起来与原文本相同
print(extract_zwc(marked_text))  # 输出: AUTHOR:张三

零宽字符水印的独特优势在于：

• 跨平台兼容 ：能在PDF、Word、网页等各种文本载体中存活
• 抗格式修改 ：即使改变字体/颜色/排版，水印依然存在
• 隐蔽性强 ：常规检查根本无法发现

注意：某些社交平台会过滤零宽字符，使用前需测试目标系统的兼容性。对于关键应用，建议混合使用多种零宽字符提升鲁棒性。

3. 数据库水印：给数据打上隐形身份证

数据库水印需要在不影响数据使用的前提下，为每条记录植入溯源标记。以数值型数据为例，我们采用改进的LSB算法——只对特定位置的记录进行修改，避免全表变动。

import hashlib
import sqlite3

class DatabaseWatermarker:
    def __init__(self, db_path, secret_key):
        self.conn = sqlite3.connect(db_path)
        self.key = secret_key
        
    def _get_marked_positions(self, total_records, watermark):
        # 用密钥+水印生成确定性的记录位置
        hash_str = hashlib.sha256((self.key + watermark).encode()).hexdigest()
        positions = set()
        
        for i in range(0, len(hash_str), 4):
            pos = int(hash_str[i:i+4], 16) % total_records
            positions.add(pos)
            
        return sorted(positions)
    
    def embed(self, table_name, column_name, watermark):
        cursor = self.conn.cursor()
        cursor.execute(f"SELECT rowid, {column_name} FROM {table_name}")
        records = cursor.fetchall()
        
        marked_positions = self._get_marked_positions(len(records), watermark)
        binary_mark = ''.join(format(ord(c), '08b') for c in watermark)
        
        for i, pos in enumerate(marked_positions):
            if i >= len(binary_mark):
                break
                
            rowid, value = records[pos]
            new_value = int(value) | (int(binary_mark[i]) << 1)  # 使用次低位
            cursor.execute(
                f"UPDATE {table_name} SET {column_name}=? WHERE rowid=?",
                (new_value, rowid)
            )
        
        self.conn.commit()
    
    def extract(self, table_name, column_name, watermark):
        cursor = self.conn.cursor()
        cursor.execute(f"SELECT rowid, {column_name} FROM {table_name}")
        records = cursor.fetchall()
        
        marked_positions = self._get_marked_positions(len(records), watermark)
        binary_str = []
        
        for i, pos in enumerate(marked_positions):
            if i >= len(watermark) * 8:
                break
                
            value = records[pos][1]
            bit = (value >> 1) & 1
            binary_str.append(str(bit))
        
        watermark = ''
        for i in range(0, len(binary_str), 8):
            byte = binary_str[i:i+8]
            if len(byte) == 8:
                watermark += chr(int(''.join(byte), 2))
        
        return watermark

# 使用示例
watermarker = DatabaseWatermarker('sales.db', 'COMPANY_SECRET')
watermarker.embed('transactions', 'amount', 'DEP001')

# 数据泄露后提取水印
leaked_db = DatabaseWatermarker('leaked.db', 'COMPANY_SECRET')
print(leaked_db.extract('transactions', 'amount', 'DEP001'))  # 输出: DEP001

数据库水印设计要点 ：

1. 选择性修改 ：只改动部分记录的关键位，保持数据统计特性
2. 密钥绑定 ：水印位置由密钥决定，不知道密钥无法提取
3. 容错设计 ：即使部分数据被修改，仍能提取完整水印
4. 多重验证 ：在不同字段植入相同水印，提高检测成功率

4. 水印技术实战：保护Python源代码

将上述技术组合使用，我们可以为Python源代码文件创建多层保护：

import ast
import zlib

def protect_source(code, owner_id):
    # 第一层：在注释中添加零宽字符水印
    zwc_mark = embed_zwc("", f"OWNER:{owner_id}")
    protected = f"# Copyright {zwc_mark}\n{code}"
    
    # 第二层：在AST节点中添加隐形属性
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
            node.name += '\u200b'  # 在名称后添加零宽字符
    
    # 第三层：生成校验哈希
    checksum = zlib.adler32(code.encode()) & 0xffffffff
    protected += f"\n__checksum__ = {checksum}\n"
    
    return ast.unparse(tree)

# 验证函数
def verify_source(code, owner_id):
    # 提取零宽字符水印
    first_line = code.split('\n')[0]
    extracted = extract_zwc(first_line)
    
    if f"OWNER:{owner_id}" not in extracted:
        return False
    
    # 验证校验和
    checksum_line = code.strip().split('\n')[-1]
    if not checksum_line.startswith('__checksum__ = '):
        return False
    
    expected = int(checksum_line.split('=')[1].strip())
    actual = zlib.adler32(code.split('__checksum__')[0].encode()) & 0xffffffff
    
    return expected == actual

这种组合方案实现了：

• 可见版权声明 （常规注释）
• 隐形身份标记 （零宽字符）
• 代码完整性校验 （ Adler-32哈希）
• 结构特征标记 （AST节点修改）

实际项目中，可以根据需要选择不同层级的保护。对于商业软件，还可以将水印信息编译到字节码中，实现更高级别的保护。