OCR 文字识别：CRNN, PaddleOCR

from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class OCRResult:
    """OCR 单条识别结果"""
    bbox: List[Tuple[int, int]]  # 四边形顶点 [(x1,y1), (x2,y2), (x3,y3), (x4,y4)]
    text: str                     # 识别出的文本
    confidence: float             # 识别置信度 [0, 1]

@dataclass
class OCRPageResult:
    """整页 OCR 结果"""
    image_width: int
    image_height: int
    results: List[OCRResult]

    def get_sorted_text(self, reading_order: str = "ltr-ttb") -> str:
        """按阅读顺序拼接文本"""
        if reading_order == "ltr-ttb":
            sorted_results = sorted(
                self.results,
                key=lambda r: (min(p[1] for p in r.bbox), min(p[0] for p in r.bbox))
            )
        return "\n".join(r.text for r in sorted_results)

    @property
    def jsonl(self) -> str:
        import json
        return "\n".join(json.dumps({
            "bbox": r.bbox, "text": r.text, "confidence": r.confidence
        }, ensure_ascii=False) for r in self.results)

# OCR 技术演进时间线
ocr_evolution = {
    "1990s": "模板匹配 + 特征工程（Tesseract 1.x）",
    "2000s": "统计学习 + SVM/CRF（Tesseract 3.x）",
    "2015": "CNN 文字识别（CRNN 提出）",
    "2017": "两阶段检测识别（CTPN + CRNN）",
    "2018": "端到端检测（EAST, MaskTextSpotter）",
    "2019": "DB 可微二值化检测器",
    "2020": "PaddleOCR 开源（工业级 OCR 系统）",
    "2021": "ABINet 语言模型增强识别",
    "2022": "PARSENet 多模态文档理解",
    "2023": "多模态大模型（LayoutLMv3, DocLLM）",
}

print("=== OCR 技术演进 ===")
for year, desc in ocr_evolution.items():
    print(f"  {year}: {desc}")
# 从模板匹配到大模型，OCR 经历了 30 年技术迭代
# 核心转变：手工特征 → 深度学习 → 多模态理解

import cv2
import pytesseract
import numpy as np

# 基本使用
img = cv2.imread("document.png")
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

# 预处理：二值化 + 去噪
_, binary = cv2.threshold(gray, 0, 255,
    cv2.THRESH_BINARY + cv2.THRESH_OTSU)
denoised = cv2.fastNlMeansDenoising(binary, h=10)

# 倾斜校正
coords = np.column_stack(np.where(denoised > 0))
angle = cv2.minAreaRect(coords)[-1]
if angle < -45:
    angle = -(90 + angle)
else:
    angle = -angle

(h, w) = denoised.shape[:2]
center = (w // 2, h // 2)
M = cv2.getRotationMatrix2D(center, angle, 1.0)
rotated = cv2.warpAffine(denoised, M, (w, h),
    flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)

# Tesseract 识别
config = "--oem 3 --psm 6 -l chi_sim+eng"
text = pytesseract.image_to_string(rotated, config=config)
print(text)

# Tesseract Page Segmentation Mode (PSM) 详解
psm_modes = {
    0: "OSD 仅（方向+脚本检测）",
    1: "自动 OSD",
    2: "自动页面分割 + OSD",
    3: "全自动页面分割（默认）",
    4: "假设单列可变大小文本",
    5: "假设单一均匀垂直列文本",
    6: "假设统一文本块",
    7: "单行文本",
    8: "单词",
    9: "圆中的单个词",
    10: "单个字符",
    11: "稀疏文本",
    12: "稀疏文本 + OSD",
    13: "原始行（跳过 hack）",
}

# OCR Engine Mode (OEM)
oem_modes = {
    0: "仅传统引擎（LSTM 关闭）",
    1: "仅神经网络 LSTM 引擎",
    2: "传统 + LSTM 组合",
    3: "默认（基于可用内容自动选择）",
}

# 获取详细输出：字级别位置和置信度
data = pytesseract.image_to_data(
    rotated, config="--oem 3 --psm 6 -l chi_sim+eng",
    output_type=pytesseract.Output.DICT
)
# data 包含: level, page_num, block_num, par_num,
#           line_num, word_num, left, top, width, height,
#           conf, text
for i, conf in enumerate(data["conf"]):
    if int(conf) > 0:
        print(f"  [{conf}%] {data['text'][i]} "
              f"at ({data['left'][i]},{data['top'][i]})")

# EAST 文字检测实现
import cv2
import numpy as np

class EASTDetector:
    """EAST 场景文字检测器"""
    def __init__(self, model_path, conf_threshold=0.5, nms_threshold=0.2):
        self.net = cv2.dnn.readNet(model_path)
        self.conf_threshold = conf_threshold
        self.nms_threshold = nms_threshold

    def detect(self, image):
        """检测文字区域，返回旋转框列表"""
        orig_h, orig_w = image.shape[:2]
        # EAST 要求输入尺寸为 32 的倍数
        new_w = (orig_w // 32) * 32
        new_h = (orig_h // 32) * 32
        ratio_w = orig_w / new_w
        ratio_h = orig_h / new_h

        blob = cv2.dnn.blobFromImage(image, 1.0, (new_w, new_h),
            (123.68, 116.78, 103.94), swapRB=True, crop=False)
        self.net.setInput(blob)

        # EAST 输出两层: 分数图 + 几何图
        (scores, geometry) = self.net.forward([
            "feature_fusion/Conv_7/Sigmoid",
            "feature_fusion/concat_3"
        ])

        return self._decode_predictions(scores, geometry, ratio_w, ratio_h)

    def _decode_predictions(self, scores, geometry, ratio_w, ratio_h):
        """解码 EAST 输出为旋转矩形框"""
        boxes, confidences = [], []
        num_rows, num_cols = scores.shape[2:4]

        for y in range(num_rows):
            scores_data = scores[0, 0, y]
            geo_data = geometry[0, :, y]

            for x in range(num_cols):
                if scores_data[x] < self.conf_threshold:
                    continue

                # 解码旋转框参数
                offset_x = x * 4.0
                offset_y = y * 4.0
                angle = geo_data[4, x]
                h = geo_data[1, x] + geo_data[3, x]
                w = geo_data[2, x] + geo_data[0, x]

                center_x = offset_x + geo_data[0, x]
                center_y = offset_y + geo_data[1, x]

                boxes.append({
                    "center_x": center_x * ratio_w,
                    "center_y": center_y * ratio_h,
                    "w": w * ratio_w,
                    "h": h * ratio_h,
                    "angle": angle,
                    "conf": scores_data[x]
                })

        # NMS 去重
        indices = cv2.dnn.NMSBoxesRotated(
            boxes, [b["conf"] for b in boxes],
            self.conf_threshold, self.nms_threshold
        )
        return [boxes[i] for i in indices]

# DBNet 可微二值化核心实现
import torch
import torch.nn as nn
import torch.nn.functional as F

class DifferentiableBinarization(nn.Module):
    """DBNet 可微二值化模块"""
    def __init__(self, k=50):
        super().__init__()
        self.k = k  # 锐化系数，越大二值化越锐利

    def forward(self, prob_map, thresh_map):
        """
        prob_map: 概率图 [B, 1, H, W]
        thresh_map: 阈值图 [B, 1, H, W]
        返回: 近似二值图 [B, 1, H, W]
        """
        # 可微近似: B_hat = 1 / (1 + exp(-k * (P - T)))
        # 当 k→∞ 时趋近于硬二值化 step(P - T)
        binary = self._step_function(prob_map, thresh_map, self.k)
        return binary

    @staticmethod
    def _step_function(x, thresh, k=50):
        """sigmoid 近似 step 函数"""
        return torch.sigmoid(k * (x - thresh))

class DBNet(nn.Module):
    """DBNet 简化版"""
    def __init__(self, backbone="resnet50"):
        super().__init__()
        # 特征提取 backbone
        self.backbone = self._build_backbone(backbone)
        # 特征金字塔 FPN
        self.fpn = self._build_fpn()
        # 两个头: 概率图 + 阈值图
        self.prob_head = nn.Conv2d(64, 1, 3, padding=1)
        self.thresh_head = nn.Sequential(
            nn.Conv2d(64, 64 // 4, 3, padding=1),
            nn.BatchNorm2d(64 // 4),
            nn.ReLU(inplace=True),
            nn.ConvTranspose2d(64 // 4, 64 // 4, 2, 2),
            nn.BatchNorm2d(64 // 4),
            nn.ReLU(inplace=True),
            nn.Conv2d(64 // 4, 1, 3, padding=1),
            nn.Sigmoid()
        )
        self.binarize = DifferentiableBinarization(k=50)

    def forward(self, x):
        features = self.backbone(x)
        features = self.fpn(features)
        prob_map = torch.sigmoid(self.prob_head(features))
        thresh_map = self.thresh_head(features)
        binary_map = self.binarize(prob_map, thresh_map)
        return prob_map, thresh_map, binary_map

import torch
import torch.nn as nn
import torch.nn.functional as F

class CRNN(nn.Module):
    """CRNN: CNN + BiLSTM + CTC"""
    def __init__(self, img_h=32, nclass=3952, nh=256, n_rnn=2):
        super().__init__()
        # CNN 特征提取 (类似 VGG)
        self.cnn = nn.Sequential(
            # conv1: [B, 1, 32, W] -> [B, 64, 16, W/2]
            nn.Conv2d(1, 64, 3, 1, 1), nn.ReLU(True),
            nn.MaxPool2d(2, 2),
            # conv2: [B, 64, 16, W/2] -> [B, 128, 8, W/4]
            nn.Conv2d(64, 128, 3, 1, 1), nn.ReLU(True),
            nn.MaxPool2d(2, 2),
            # conv3: [B, 128, 8, W/4] -> [B, 256, 4, W/8]
            nn.Conv2d(128, 256, 3, 1, 1), nn.BatchNorm2d(256), nn.ReLU(True),
            nn.Conv2d(256, 256, 3, 1, 1), nn.ReLU(True),
            nn.MaxPool2d((2, 1), (2, 1)),  # 只降低高度
            # conv4: [B, 256, 2, W/8] -> [B, 512, 1, W/8]
            nn.Conv2d(256, 512, 3, 1, 1), nn.BatchNorm2d(512), nn.ReLU(True),
            nn.MaxPool2d((2, 1), (2, 1)),
            # conv5: [B, 512, 1, W/16] -> [B, 512, 1, W/16]
            nn.Conv2d(512, 512, (2, 1), 1, 0), nn.BatchNorm2d(512), nn.ReLU(True),
        )
        # 映射层: 512 -> nh
        self.mapping = nn.Sequential(
            nn.Conv2d(512, nh, 1, 1, 0), nn.ReLU(True)
        )
        # RNN 序列建模
        self.rnn = self._build_rnn(nh, n_rnn)
        # CTC 分类头
        self.fc = nn.Linear(nh, nclass)

    def _build_rnn(self, nh, n_rnn):
        return nn.LSTM(nh, nh // 2, bidirectional=True,
                       num_layers=n_rnn, batch_first=False)

    def forward(self, x):
        conv = self.cnn(x)          # [B, 512, 1, W/16]
        conv = self.mapping(conv)    # [B, nh, 1, W/16]
        b, c, h, w = conv.size()
        conv = conv.view(b, -1, w)   # [B, nh, W/16] 序列
        conv = conv.permute(2, 0, 1) # [W/16, B, nh]
        rnn_out, _ = self.rnn(conv)  # [W/16, B, nh]
        output = self.fc(rnn_out)    # [W/16, B, nclass]
        return output  # CTC 输入

# CTC 解码详解
def ctc_decode(predictions, blank_index=0):
    """
    CTC 贪婪解码 (Greedy Decode)
    predictions: [seq_len, nclass] - 每个时间步的字符概率
    返回: 解码后的文本
    """
    # 每步取最大概率的字符
    pred_indices = predictions.argmax(dim=-1)  # [seq_len]

    # 去除连续重复 + 去除 blank
    decoded = []
    prev = None
    for idx in pred_indices:
        if idx != blank_index and idx != prev:
            decoded.append(idx.item())
        prev = idx
    return decoded

def ctc_beam_search(predictions, beam_width=10, blank_index=0):
    """
    CTC 束搜索解码 (Beam Search Decode)
    比贪婪解码更准确，考虑多个候选路径
    """
    import math
    # 简化版: 维护 top-k 路径
    seq_len, nclass = predictions.shape
    probs = F.softmax(predictions, dim=-1)

    # 初始: 空路径概率为 1
    beams = {(): 1.0}

    for t in range(seq_len):
        new_beams = {}
        for path, prob in beams.items():
            for c in range(nclass):
                new_prob = prob * probs[t, c].item()
                if c == blank_index:
                    # blank: 路径不变
                    new_beams[path] = new_beams.get(path, 0) + new_prob
                elif path and path[-1] == c:
                    # 重复字符: 跳过 (CTC 规则)
                    new_beams[path] = new_beams.get(path, 0) + new_prob
                else:
                    # 新字符: 追加
                    new_path = path + (c,)
                    new_beams[new_path] = new_beams.get(new_path, 0) + new_prob

        # 保留 top-k
        beams = dict(sorted(new_beams.items(),
                            key=lambda x: -x[1])[:beam_width])

    # 返回最高概率路径
    best_path = max(beams, key=beams.get)
    return list(best_path), beams[best_path]

# 字符集映射
charset = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
charset = ["<blank>"] + list(charset)

def indices_to_text(indices):
    return "".join(charset[i] for i in indices if i < len(charset))

decoded_indices = ctc_decode(torch.randn(25, 65))
print(f"识别结果: {indices_to_text(decoded_indices)}")

class ABINetLanguageModel(nn.Module):
    """ABINet 语言模型（纠错分支）"""
    def __init__(self, n_layers=4, d_model=512, n_head=8,
                 d_inner=2048, dropout=0.1, max_length=25):
        super().__init__()
        self.max_length = max_length
        self.embedding = nn.Embedding(39, d_model)  # 38 chars + PAD

        # Transformer Decoder
        decoder_layer = nn.TransformerDecoderLayer(
            d_model=d_model, nhead=n_head,
            dim_feedforward=d_inner, dropout=dropout
        )
        self.transformer = nn.TransformerDecoder(
            decoder_layer, num_layers=n_layers
        )
        self.generator = nn.Linear(d_model, 39)

        # 位置编码
        self.pos_enc = nn.Parameter(
            torch.randn(max_length, d_model) * 0.01
        )

    def forward(self, visual_features, tgt=None):
        """
        visual_features: [B, seq_len, d_model] 视觉分支输出
        tgt: 目标序列（训练时）或 None（推理时自回归）
        """
        b, seq_len, _ = visual_features.size()

        # 自回归生成
        if tgt is None:
            tgt = torch.full((b, 1), 0, dtype=torch.long,
                             device=visual_features.device)  # SOS token
            for _ in range(self.max_length - 1):
                tgt_emb = self.embedding(tgt) + \
                    self.pos_enc[:tgt.size(1)].unsqueeze(0)
                tgt_emb = tgt_emb.permute(1, 0, 2)
                vis_perm = visual_features.permute(1, 0, 2)

                out = self.transformer(tgt_emb, vis_perm)
                logits = self.generator(out[-1:])  # 最后一步
                next_token = logits.argmax(-1)
                tgt = torch.cat([tgt, next_token.transpose(0, 1)], dim=1)

                if (next_token == 1).all():  # EOS token
                    break
            return tgt

        # 训练模式: teacher forcing
        tgt_emb = self.embedding(tgt) + self.pos_enc[:tgt.size(1)].unsqueeze(0)
        tgt_emb = tgt_emb.permute(1, 0, 2)
        vis_perm = visual_features.permute(1, 0, 2)

        out = self.transformer(tgt_emb, vis_perm)
        logits = self.generator(out)
        return logits

class EndToEndOCR(nn.Module):
    """简化的端到端 OCR 系统"""
    def __init__(self, num_classes=37):
        super().__init__()
        # 共享 Backbone
        self.backbone = self._build_backbone()
        # 检测头
        self.det_head = nn.Sequential(
            nn.Conv2d(256, 128, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(128, 1, 1)  # 文字区域概率图
        )
        # RoI Align 将不规则区域转为固定大小
        self.roi_align = nn.Sequential(
            nn.AdaptiveAvgPool2d((8, 32)),  # 标准化为 8x32
        )
        # 识别头 (CRNN)
        self.rec_head = nn.Sequential(
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(),
            nn.LSTM(256, 128, bidirectional=True, batch_first=True),
            nn.Linear(256, num_classes)
        )

    def forward(self, image, proposals=None):
        features = self.backbone(image)  # [B, 256, H/16, W/16]

        # 检测
        if proposals is None:
            score_map = torch.sigmoid(self.det_head(features))
            proposals = self._extract_proposals(score_map)

        # 对每个 proposal 做 RoI Align + 识别
        results = []
        for proposal in proposals:
            roi_feat = self._crop_and_resize(features, proposal)
            roi_feat = self.roi_align(roi_feat)
            b, c, h, w = roi_feat.size()
            seq = roi_feat.view(b, c, -1).permute(0, 2, 1)
            rec_out = self.rec_head(seq)
            results.append(rec_out)

        return results

from paddleocr import PaddleOCR
import cv2

# 初始化 OCR 引擎
ocr = PaddleOCR(
    use_angle_cls=True,      # 启用文字方向分类器
    lang="ch",               # 中英文识别
    use_gpu=True,            # GPU 加速
    det_model_dir="./models/det/",
    rec_model_dir="./models/rec/",
    cls_model_dir="./models/cls/",
    det_db_thresh=0.3,       # 检测阈值
    det_db_box_thresh=0.6,   # 检测框置信度阈值
    rec_batch_num=6,         # 识别 batch size
    max_text_length=25,      # 最大文本长度
)

# 推理
image_path = "sample.jpg"
result = ocr.ocr(image_path, cls=True)

# 解析结果
img = cv2.imread(image_path)
for line in result:
    if line is None:
        continue
    for word_info in line:
        bbox = word_info[0]       # 四边形顶点
        text = word_info[1][0]    # 识别文本
        score = word_info[1][1]   # 置信度

        # 绘制结果
        pts = np.array(bbox, dtype=np.int32)
        cv2.polylines(img, [pts], True, (0, 255, 0), 2)
        cv2.putText(img, f"{text} ({score:.2f})",
                     (int(bbox[0][0]), int(bbox[0][1]) - 5),
                     cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1)

        print(f"  文本: {text} | 置信度: {score:.4f}")

cv2.imwrite("ocr_result.jpg", img)

# PaddleOCR 模型量化与部署
from paddle.static import InputSpec
from paddle.fluid.contrib.slim.quant import quant_aware

# 1. 导出 inference 模型
# !paddleocr --export_model --output_dir ./inference/

# 2. INT8 量化（减小模型体积 4 倍）
# !paddle2onnx --model_dir ./inference/ \
#     --model_filename inference.pdmodel \
#     --params_filename inference.pdiparams \
#     --save_file ocr.onnx \
#     --opset_version 11

# 3. ONNX Runtime 推理
import onnxruntime as ort

session = ort.InferenceSession("ocr.onnx")
input_name = session.get_inputs()[0].name

def inference_with_onnx(image, session, input_name):
    """ONNX Runtime 推理"""
    # 预处理
    img = preprocess(image)  # resize, normalize
    img = img.transpose(2, 0, 1).astype("float32")
    img = np.expand_dims(img, axis=0)

    # 推理
    outputs = session.run(None, {input_name: img})
    return outputs

# 4. 性能对比
print("=== PaddleOCR 模型对比 ===")
models = {
    "PP-OCRv4 det":  {"size": "3.1MB", "speed": "15ms", "accuracy": "84.5%"},
    "PP-OCRv4 rec":  {"size": "3.5MB", "speed": "8ms", "accuracy": "80.1%"},
    "PP-OCRv4 server det": {"size": "110MB", "speed": "25ms", "accuracy": "87.3%"},
    "PP-OCRv4 server rec": {"size": "98MB", "speed": "15ms", "accuracy": "84.2%"},
}
for name, m in models.items():
    print(f"  {name:<20} | {m['size']:>7} | {m['speed']:>7} | {m['accuracy']}")

from transformers import TrOCRProcessor, VisionEncoderDecoderModel
import torch
from PIL import Image

# TrOCR: Transformer-based OCR 手写识别
class HandwritingRecognizer:
    """基于 TrOCR 的手写文字识别器"""
    def __init__(self, model_name="microsoft/trocr-base-handwritten"):
        self.processor = TrOCRProcessor.from_pretrained(model_name)
        self.model = VisionEncoderDecoderModel.from_pretrained(model_name)
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model.to(self.device)

    def recognize(self, image_path):
        """识别手写文字"""
        image = Image.open(image_path).convert("RGB")

        # 预处理
        pixel_values = self.processor(
            image, return_tensors="pt"
        ).pixel_values.to(self.device)

        # 生成文本
        generated_ids = self.model.generate(
            pixel_values,
            max_length=64,
            num_beams=5,         # Beam Search
            early_stopping=True,
            no_repeat_ngram_size=3,  # 防止重复
        )
        text = self.processor.batch_decode(
            generated_ids, skip_special_tokens=True
        )[0]
        return text

    def batch_recognize(self, image_paths):
        """批量识别"""
        images = [Image.open(p).convert("RGB") for p in image_paths]
        pixel_values = self.processor(
            images, return_tensors="pt", padding=True
        ).pixel_values.to(self.device)

        generated_ids = self.model.generate(
            pixel_values, max_length=64, num_beams=5
        )
        return self.processor.batch_decode(
            generated_ids, skip_special_tokens=True
        )

# 多语言字符集管理
class MultiLanguageCharset:
    """多语言字符集管理"""
    def __init__(self):
        # 各语言字符集
        self.charsets = {
            "en": "0123456789abcdefghijklmnopqrstuvwxyz"
                  "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
            "zh_cn": self._load_chinese_chars(),  # 6500 常用汉字
            "ja": self.charsets["en"] + self._load_japanese_chars(),
            "ar": self._load_arabic_chars(),  # 从右向左
            "ko": self._load_korean_chars(),
        }
        # 合并字符集
        self.unified = self._build_unified_charset()
        self.char2idx = {c: i for i, c in enumerate(self.unified)}
        self.idx2char = {i: c for i, c in enumerate(self.unified)}

    def _load_chinese_chars(self):
        """加载常用汉字（示例前 100 个）"""
        return "的一了是我不在人们有来他这上着个地到大里说" \
               "就去子得也那要下看天时过出么然还起都实" \
               "现所面前着她而里后以自会家可方成当没动" \
               "行物生应知水明问力理尔点文几定本公特外" \
               "儿相因小社者经此发十如西去种两都法想情" \
               "回但开已其第些作前新想又进所好无只才"

    def _build_unified_charset(self):
        """构建统一字符集"""
        all_chars = set()
        for lang, chars in self.charsets.items():
            all_chars.update(chars)
        return "".join(sorted(all_chars))

    def encode_text(self, text, lang=None):
        """文本 → 索引序列"""
        charset = self.charsets.get(lang, self.unified)
        return [self.char2idx.get(c, 0) for c in text]

# 多语言识别策略
strategies = {
    "统一模型": {
        "描述": "单一模型处理所有语言",
        "优点": "部署简单，无需语言判断",
        "缺点": "模型大、字符间相互干扰",
        "适用": "语言混合场景"
    },
    "语言路由": {
        "描述": "先判断语言，再调用专用模型",
        "优点": "每种语言精度最优",
        "缺点": "需要语言分类器，系统复杂",
        "适用": "单语文档批量处理"
    },
    "零样本多模态": {
        "描述": "GPT-4V / Qwen-VL 等直接识别",
        "优点": "无需训练，支持任意语言",
        "缺点": "成本高，延迟大，不可控",
        "适用": "低频长尾场景"
    }
}