水印去除技术文档

概述

水印去除模块 (ocr_utils/watermark_utils.py) 提供了两层独立的水印去除能力，针对不同类型的文档和场景进行优化：

层级	处理对象	适用场景	特点
PDF 层级	文字型 PDF 的 XObject	银行流水等文字型 PDF	保留文字可搜索性，无损处理
图像层级	扫描件/渲染图像的像素	扫描件、图片	像素级处理，适用于 OCR 前预处理

处理流程

graph TB
    A[输入文档] --> B{是否为 PDF?}
    
    B -->|是| C[阶段一: PDF 层级去水印]
    B -->|否| F
    
    C --> D{启用 txt_pdf_watermark_removal?}
    D -->|是| E[扫描前 N 页检测水印 XObject]
    D -->|否| G
    E --> E1{发现水印?}
    E1 -->|是| E2[清除 XObject 内容流]
    E1 -->|否| G
    E2 --> G
    
    G[渲染为图像] --> H{PDF 类型?}
    H -->|文字型 txt| I[跳过阶段二]
    H -->|扫描件 ocr| J
    
    F[图像输入] --> J[阶段二: 图像级去水印]
    
    J --> K{启用 watermark_removal?}
    K -->|是| L[检测浅色斜向水印]
    K -->|否| N
    L --> M[阈值化去除水印]
    M --> N[方向校正]
    
    N --> O[Layout 检测]
    O --> P[OCR 识别]
    
    style C fill:#e1f5ff
    style E fill:#e1f5ff
    style E2 fill:#e1f5ff
    style J fill:#fff4e1
    style L fill:#fff4e1
    style M fill:#fff4e1

阶段一：PDF 层级水印去除

适用场景

文字型 PDF（pdf_type='txt'）：PDF 内部包含可提取的文字层，水印通常以 XObject 形式叠加在文字上方。

原理

PDF 文件中的水印通常通过以下两种 XObject 实现：

Form XObject：矢量绘图对象，包含旋转、透明度等变换矩阵
Image XObject：位图对象，通常是半透明的全页背景图

通过 PyMuPDF (fitz) 直接操作 PDF 内部结构，清空或替换水印 XObject 的内容流，而不影响文字层的可搜索性。

水印 XObject 判断规则

Form XObject 水印判断 (`_is_watermark_xobj`)

满足以下条件之一即判定为水印：

规则	说明	原理
旋转变换	内容流中 `cm` 指令的 sin/cos 分量非零	水印通常斜向 45° 放置
透明度组 + 透明操作符	`/Group` 存在且内容流含 `ca/CA`	水印具有半透明效果
透明度组 + 大体积流	`/Group` 存在且流体积 > 2KB	大量重复绘图 = 平铺水印

# 判断逻辑伪代码
def _is_watermark_xobj(doc, xref, obj_str):
    if "/Form" not in obj_str:
        return False
    
    stream_text = doc.xref_stream(xref).decode("latin-1")
    
    # 规则1：旋转变换
    if has_rotation_transform(stream_text):
        return True
    
    # 规则2-3：透明度组相关
    if "/Group" in obj_str:
        if has_transparency_operators(stream_text):
            return True
        if len(stream_text) > 2048:
            return True
    
    return False

Image XObject 水印判断 (`_is_watermark_image_xobj`)

必须同时满足以下条件：

条件	说明
`/Subtype /Image`	确认是图像对象
存在 `/SMask`	有透明通道（半透明）
宽 >= 600 且高 >= 800	全页尺寸（排除小图标）
像素均值 >= 240	近乎全白（水印文字稀疏）

处理方法

def remove_txt_pdf_watermark(pdf_bytes: bytes) -> Optional[bytes]:
    """
    对文字型 PDF 执行原生水印去除
    
    处理方式：
    - Form XObject：清空内容流 (update_stream(b""))
    - Image XObject：替换为全白像素 + 移除 DecodeParms
    
    Returns:
        去水印后的 PDF bytes，若未发现水印返回 None
    """
    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    
    for page in doc:
        # 处理 Form XObject 水印
        for xref, name, *_ in page.get_xobjects():
            if _is_watermark_xobj(doc, xref, obj_str):
                doc.update_stream(xref, b"")  # 清空内容流
        
        # 处理 Image XObject 水印
        for img_tuple in page.get_images(full=True):
            img_xref = img_tuple[0]
            if _is_watermark_image_xobj(doc, img_xref, obj_str):
                _blank_watermark_image(doc, img_xref)  # 替换为全白
    
    return doc.tobytes(garbage=4, deflate=True)

关键技术细节

移除 /DecodeParms 的必要性：

当 Image XObject 使用 Predictor 压缩时，必须先移除 /DecodeParms 再调用 update_stream，否则渲染器会尝试 Predictor 解码失败后回退原始数据，水印依然可见。

def _blank_watermark_image(doc, img_xref):
    # 关键：先移除 DecodeParms
    doc.xref_set_key(img_xref, "DecodeParms", "null")
    # 再更新为全白像素
    doc.update_stream(img_xref, bytes([255]) * (w * h * channels))

快速预扫描 (`scan_pdf_watermark_xobjs`)

对于大型 PDF（如财报），先执行只读扫描判断是否存在水印，避免不必要的全量处理：

def scan_pdf_watermark_xobjs(pdf_bytes: bytes, sample_pages: int = 3) -> bool:
    """
    快速扫描前 N 页，判断是否含水印 XObject
    
    Args:
        sample_pages: 扫描页数上限，默认 3（银行流水通常前几页有水印）
    
    Returns:
        True 表示发现水印 XObject
    """
    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    for i in range(min(sample_pages, len(doc))):
        # 检查 Form XObject 和 Image XObject
        ...
    return False

阶段二：图像级水印去除

适用场景

扫描件/图片（pdf_type='ocr'）：无法从 PDF 内部结构处理，只能对渲染后的图像进行像素级处理。

原理

银行流水等金融文档的水印特征：

颜色浅：灰度值通常在 160-220 之间（介于正文和背景之间）
角度斜：通常 45° 斜向排列
文字稀疏：水印文字占比较小

基于这些特征，采用阈值化处理：将灰度值高于阈值的像素置为白色，保留深色正文。

水印检测 (`detect_watermark`)

采用两阶段检测策略：

中间调检测：统计灰度在 100-220 之间的像素占比

斜向验证：使用 Hough 直线变换验证是否存在斜向纹理

def detect_watermark(image, midtone_low=100, midtone_high=220, ratio_threshold=0.03):
"""
检测图像中是否存在浅色斜向文字水印
    
步骤：
1. 提取中间调像素（100-220），计算占比
2. 若占比 > 3%，进行斜向验证
3. 使用 Canny 边缘检测 + Hough 直线变换
4. 统计 30-60° 斜向直线数量
"""
gray = to_grayscale(image)
    
# 步骤1：中间调检测
midtone_mask = (gray > midtone_low) & (gray < midtone_high)
if midtone_mask.sum() / gray.size < ratio_threshold:
    return False
    
# 步骤2：斜向验证
edges = cv2.Canny(midtone_mask, 50, 150)
lines = cv2.HoughLines(edges, rho=1, theta=np.pi/180, threshold=80)
    
# 统计斜向（30-60°）直线
diagonal_count = count_diagonal_lines(lines, angle_range=(30, 60))
return diagonal_count >= 2

水印去除 (`remove_watermark_from_image`)

def remove_watermark_from_image(image, threshold=160, morph_close_kernel=0):
    """
    去除图像中的浅色斜向文字水印
    
    原理：
    - 正文为深黑色（灰度 < threshold）
    - 水印为浅灰（灰度 > threshold）
    - 将高于阈值的像素置为白色（255）
    
    Args:
        threshold: 灰度阈值，建议 140-180，默认 160
        morph_close_kernel: 形态学闭运算核，0 表示跳过
    """
    gray = to_grayscale(image)
    
    # 阈值化：保留深色正文
    cleaned = gray.copy()
    cleaned[gray > threshold] = 255
    
    # 可选：形态学闭运算填补字符断裂
    if morph_close_kernel > 0:
        kernel = np.ones((morph_close_kernel, morph_close_kernel), np.uint8)
        cleaned = cv2.morphologyEx(cleaned, cv2.MORPH_CLOSE, kernel)
    
    return cleaned

参数说明

参数	默认值	说明	调整建议
`threshold`	160	灰度阈值	140-180，越大越保守（可能残留水印）
`morph_close_kernel`	0	形态学核大小	非二值图建议设为 0（闭运算会适得其反）

配置说明

完整配置示例

# 输入配置 - PDF 层级去水印
input:
  dpi: 200
  txt_pdf_watermark_removal:
    enabled: true        # 是否启用 PDF 层级去水印
    sample_pages: 3      # 快速预扫描页数

# 预处理配置 - 图像级去水印
preprocessor:
  module: "mineru"
  orientation_classifier:
    enabled: true
  watermark_removal:
    enabled: true           # 是否启用图像级去水印
    threshold: 160          # 灰度阈值
    morph_close_kernel: 0   # 形态学核大小（建议 0）

配置项详解

配置路径	类型	默认值	说明
`input.txt_pdf_watermark_removal.enabled`	bool	`false`	PDF 层级去水印开关
`input.txt_pdf_watermark_removal.sample_pages`	int	3	预扫描页数
`preprocessor.watermark_removal.enabled`	bool	`false`	图像级去水印开关
`preprocessor.watermark_removal.threshold`	int	160	灰度阈值
`preprocessor.watermark_removal.morph_close_kernel`	int	0	形态学核大小

注意：两个配置均无默认值，必须在 YAML 中显式配置 enabled: true 才会触发。

触发条件

阶段一触发条件

# pipeline_manager_v2.py: process_document()

if is_pdf:
    wm_cfg = config.get('input', {}).get('txt_pdf_watermark_removal', {})
    if wm_cfg.get('enabled', False):                    # 条件①
        if scan_pdf_watermark_xobjs(pdf_bytes, sample_pages=3):  # 条件②
            cleaned = remove_txt_pdf_watermark(pdf_bytes)

触发条件：

文件是 PDF
enabled: true
扫描发现水印 XObject

阶段二触发条件

# pipeline_manager_v2.py: _process_single_page()

if pdf_type == 'ocr':  # 条件①：仅扫描件
    detection_image, angle = self.preprocessor.process(original_image)

# mineru_adapter.py: MinerUPreprocessor.process()

if config.get('watermark_removal', {}).get('enabled', False):  # 条件②
    image = remove_watermark_from_image_rgb(image, threshold=160)

触发条件：

PDF 类型为 ocr（扫描件）
preprocessor.watermark_removal.enabled: true

两阶段对比

维度	阶段一（PDF 层级）	阶段二（图像级）
处理对象	文字型 PDF	扫描件/图片
处理层级	PDF XObject	图像像素
保留文字可搜索性	✅ 是	❌ 否
无损处理	✅ 是	❌ 否（像素修改）
处理时机	渲染前	渲染后、检测前
依赖库	PyMuPDF (fitz)	OpenCV, NumPy

代码集成

流水线集成

# pipeline_manager_v2.py

from ocr_utils.watermark_utils import (
    scan_pdf_watermark_xobjs,
    remove_txt_pdf_watermark
)

class EnhancedDocPipeline:
    def process_document(self, doc_path):
        # 阶段一：PDF 层级去水印
        _pdf_bytes_override = None
        if is_pdf and config['input']['txt_pdf_watermark_removal']['enabled']:
            raw_bytes = doc_path.read_bytes()
            if scan_pdf_watermark_xobjs(raw_bytes):
                _pdf_bytes_override = remove_txt_pdf_watermark(raw_bytes)
        
        # 渲染 PDF（使用去水印后的 bytes）
        images, pdf_type, pdf_doc = PDFUtils.load_and_classify_document(
            doc_path, pdf_bytes=_pdf_bytes_override
        )
        
        # 逐页处理
        for page_idx, original_image in enumerate(images):
            # 阶段二：图像级去水印（在 preprocessor.process 中）
            if pdf_type == 'ocr':
                detection_image, angle = self.preprocessor.process(original_image)
            
            # Layout 检测、OCR...

预处理器集成

# models/adapters/mineru_adapter.py

from ocr_utils.watermark_utils import remove_watermark_from_image_rgb

class MinerUPreprocessor:
    def process(self, image):
        # 图像级水印去除（在方向校正之前）
        if self.config.get('watermark_removal', {}).get('enabled', False):
            threshold = self.config.get('watermark_removal', {}).get('threshold', 160)
            image = remove_watermark_from_image_rgb(image, threshold=threshold)
        
        # 方向校正
        if self.orientation_classifier:
            angle = self.orientation_classifier.predict(image)
            image = self._apply_rotation(image, angle)
        
        return image, angle

使用示例

命令行

# 处理含水印的银行流水 PDF
python main_v2.py -i bank_statement.pdf -c config/bank_statement_yusys_v4.yaml --scene bank_statement

# 配置文件中已启用：
# input.txt_pdf_watermark_removal.enabled: true
# preprocessor.watermark_removal.enabled: true

Python API

from core.pipeline_manager_v2 import EnhancedDocPipeline

# 使用包含水印去除配置的 YAML
with EnhancedDocPipeline("config/bank_statement_yusys_v4.yaml") as pipeline:
    results = pipeline.process_document("document.pdf")

调试与验证

日志输出

# 阶段一日志
logger.info(f"🧹 文字型 PDF 原生去水印完成（{doc_path.name}）")
logger.debug(f"  [Form XObject] 清空水印 xref={xref}, name={name}")
logger.debug(f"  [Image XObject] 替换水印图像 xref={img_xref}")

# 阶段二日志
logger.info(f"🧹 Watermark removed (threshold={threshold})")

可视化验证

在 debug 模式下，可以通过对比去水印前后的图像来验证效果：

# 开启 debug 模式
python main_v2.py -i doc.pdf -c config.yaml --scene bank_statement --debug

# 输出文件：
# {doc}_pdf_page_001.png  - 渲染后的页面图像（去水印后）
# {doc}_page_001_layout.png - Layout 可视化

注意事项

两个阶段是互补的：阶段一处理文字型 PDF，阶段二处理扫描件，实际不会重复执行
阈值选择：threshold=160 适用于大多数银行流水，如果误删浅色文字可适当提高
形态学运算：morph_close_kernel=0 是推荐值，非二值图时闭运算可能引入噪声
大文件优化：sample_pages=3 快速预扫描，避免对无水印的大文件全量处理
依赖要求：PDF 层级去水印需要 PyMuPDF，图像级需要 OpenCV

参考资料

ocr_utils/watermark_utils.py - 水印工具函数实现
core/pipeline_manager_v2.py - 流水线集成
models/adapters/mineru_adapter.py - 预处理器集成
config/bank_statement_*.yaml - 配置示例

水印去除技术文档.md 15 KB Histórico Raw

水印去除技术文档

概述

处理流程

阶段一：PDF 层级水印去除

适用场景

原理

水印 XObject 判断规则

Form XObject 水印判断 (_is_watermark_xobj)

Image XObject 水印判断 (_is_watermark_image_xobj)

处理方法

关键技术细节

快速预扫描 (scan_pdf_watermark_xobjs)

阶段二：图像级水印去除

适用场景

原理

水印检测 (detect_watermark)

水印去除 (remove_watermark_from_image)

参数说明

配置说明

完整配置示例

配置项详解

触发条件

阶段一触发条件

阶段二触发条件

两阶段对比

代码集成

流水线集成

预处理器集成

使用示例

命令行

Python API

调试与验证

日志输出

可视化验证

注意事项

参考资料

水印去除技术文档.md 15 KB

Histórico Raw

Form XObject 水印判断 (`_is_watermark_xobj`)

Image XObject 水印判断 (`_is_watermark_image_xobj`)

快速预扫描 (`scan_pdf_watermark_xobjs`)

水印检测 (`detect_watermark`)

水印去除 (`remove_watermark_from_image`)