sansenhoshi/ocr-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

首次提交

Browse Source
This commit is contained in:
sansen
2026-08-31 16:41:12 +08:00
parent 680e028a0b
commit 9a479b6832
10 changed files with 2735 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -174,3 +174,4 @@ cython_debug/
# PyPI configuration file # PyPI configuration file
.pypirc .pypirc
/CLAUDE.md
.idea/.gitignore
Generated
+10
View File
@@ -0,0 +1,10 @@
# 默认忽略的文件
/shelf/
/workspace.xml
# 已忽略包含查询文件的默认文件夹
/queries/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml
# 基于编辑器的 HTTP 客户端请求
/httpRequests/
RAPIDOCR_PARAMETERS.md
+1000
View File
File diff suppressed because it is too large Load Diff
README_GPU.md
+241
View File
@@ -0,0 +1,241 @@
# GPU 推理部署指南
## 配置参数
在 `.env` 文件中设置：
```env
# 推理设备: cpu / cuda / dml / tensorrt
OCR_DEVICE=cpu
# GPU设备编号（多GPU时指定，默认0）
OCR_DEVICE_ID=0
# CPU线程数（仅 cpu 模式生效）
OCR_INTRA_THREADS=4
OCR_INTER_THREADS=2
```
---
## 方案一：CUDA（NVIDIA 显卡）
适用：有 NVIDIA 独立显卡（GTX/RTX/Quadro/Tesla 等）。
### 1. 确认显卡支持 CUDA
```powershell
nvidia-smi
```
确保 CUDA Version >= 11.8。
### 2. 安装 CUDA Toolkit 和 cuDNN
**方法A（推荐）：直接安装 onnxruntime-gpu**
onnxruntime-gpu 已内置必要的 CUDA/cuDNN 依赖（Windows 上通过 DirectML 或 CUDA EP）：
```bash
pip uninstall onnxruntime
pip install onnxruntime-gpu
```
**方法B：手动安装 CUDA + cuDNN**
1. 下载 [CUDA Toolkit 11.8](https://developer.nvidia.com/cuda-11-8-0-download-archive)
2. 下载 [cuDNN 8.x for CUDA 11.x](https://developer.nvidia.com/cudnn)
3. 安装后将 cuDNN 的 bin/lib/include 复制到 CUDA 安装目录
### 3. 配置 .env
```env
OCR_DEVICE=cuda
OCR_DEVICE_ID=0
```
### 4. 验证
启动服务后查看日志：
```
推理设备: CUDA (device_id=0)
```
或用 Python 测试：
```python
import onnxruntime as ort
print(ort.get_available_providers())
# 应包含: ['CUDAExecutionProvider', 'CPUExecutionProvider']
```
---
## 方案二：DirectML（Windows 任意 GPU）
适用：Windows 系统，任意显卡（NVIDIA/AMD/Intel 核显），无需安装 CUDA。
### 1. 安装 onnxruntime-directml
```bash
pip uninstall onnxruntime
pip install onnxruntime-directml
```
### 2. 配置 .env
```env
OCR_DEVICE=dml
```
### 3. 验证
```python
import onnxruntime as ort
print(ort.get_available_providers())
# 应包含: ['DmlExecutionProvider', 'CPUExecutionProvider']
```
> **注意**：首次推理时 ONNX Runtime 会编译 DML 算子，可能耗时 10-30 秒，后续推理恢复正常速度。
### 4. 核显部署机（无独显）完整流程
开发机与部署机分离时，部署机常为只有核显的轻薄本/迷你主机。核显同样支持 DirectML 加速（Intel UHD / Iris Xe、AMD Radeon 核显均满足），无需任何 NVIDIA/AMD 驱动层面的额外配置。
**前置条件**
| 项目 | 要求 |
|------|------|
| 系统 | Windows 10 1903（Build 18362）及以上（RapidOCR 会检查，不满足则自动回退 CPU） |
| 显卡 | 支持 DirectX 12 的核显（现代 Intel/AMD 核显均满足） |
| Python | 3.10+ |
**全新环境安装**
```powershell
# 1. 安装项目依赖
pip install -r requirements.txt
# 2. 安装 DirectML 后端（替换掉 rapidocr 自带的 CPU 版 onnxruntime）
pip uninstall onnxruntime
pip install onnxruntime-directml
# 3. 验证后端可用
python -c "import onnxruntime as ort; print(ort.get_available_providers())"
# 应输出: ['DmlExecutionProvider', 'CPUExecutionProvider']
```
**配置 .env**
```env
OCR_DEVICE=dml
# OCR_INTRA_THREADS / OCR_INTER_THREADS 仅 cpu 模式生效，dml 模式无需设置
```
**启动并验证**
```powershell
uvicorn main:app --host 0.0.0.0 --port 8000
```
- 启动日志应出现 `推理设备: DirectML`
- 首次识别会编译着色器，等待 30-60 秒属正常现象，之后恢复正常速度
**核显加速预期**
| 场景 | 说明 |
|------|------|
| 加速比 | 约为 CPU 的 1.5-3 倍（Intel Iris Xe 单张 A4 约 1-2s） |
| CPU 较强的机器 | 核显与 CPU 共享内存带宽，加速比更接近下限 1.5x |
| 双显卡机器 | DML 使用系统默认适配器（通常为独显）；当前代码未将 `OCR_DEVICE_ID` 透传给 DML，该参数对 dml 模式不生效 |
---
## 方案三：TensorRT（NVIDIA 显卡，极致性能）
适用：追求最快推理速度的 NVIDIA GPU 用户。
### 1. 安装依赖
```bash
pip uninstall onnxruntime
pip install onnxruntime-gpu
# 安装 TensorRT
pip install tensorrt
# 或从 NVIDIA 官网下载: https://developer.nvidia.com/tensorrt
```
### 2. 配置 .env
```env
OCR_DEVICE=tensorrt
OCR_DEVICE_ID=0
```
### 3. 首次运行
TensorRT 首次加载模型时需要构建 engine（耗时 30 秒到几分钟），之后会缓存到默认模型目录。
---
## 方案四：OpenVINO（Intel CPU/GPU）
如果你使用 Intel CPU（特别是第 10 代及更新），OpenVINO 比默认 ONNX Runtime CPU 更快。
### 1. 安装
```bash
pip install openvino
```
### 2. 修改 `engine/ocr_engine.py`
在 `_build_device_params()` 中添加 OpenVINO 分支（目前代码模板已预留结构）。
---
## 性能预期
| 设备 | 单张 A4 耗时 | vs CPU | 适用场景 |
|------|-------------|--------|---------|
| CPU (i7-12700) | ~2-5s | 基准 | 低并发、开发调试 |
| CUDA (RTX 3060) | ~0.3-0.8s | 3-8x | 生产环境 NVIDIA GPU |
| DirectML (RTX 3060) | ~0.5-1.0s | 2-5x | Windows 任意 GPU |
| DirectML (Intel Iris Xe) | ~1-2s | 1.5-3x | 轻薄本核显 |
| TensorRT (RTX 3060) | ~0.15-0.4s | 5-15x | 高吞吐生产环境 |
---
## 故障排查
### CUDA 模式下报错 "CUDAExecutionProvider not found"
```bash
# 确认安装了 onnxruntime-gpu
pip show onnxruntime-gpu
# 确认 CUDA 可用
python -c "import onnxruntime; print(onnxruntime.get_available_providers())"
```
### DirectML 下首次推理卡住
首次推理需要编译着色器，等待 30-60 秒即可。后续推理恢复正常。
### GPU 显存不足
服务使用整图 OCR。如果处理大图时出现 OOM：
- 在调用服务前降低输入图片分辨率
- 降低并发请求数量
- 或降低 `.env` 中 `OCR_REC_BATCH_NUM` 的值
### 多 GPU 环境
设置 `OCR_DEVICE_ID` 指定使用哪张卡：
```env
OCR_DEVICE=cuda
OCR_DEVICE_ID=1 # 使用第2张GPU
```
config/config.py
+62
View File
@@ -0,0 +1,62 @@
from pathlib import Path
from typing import Annotated, Literal
from pydantic import AfterValidator, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
PROJECT_ROOT = Path(__file__).resolve().parents[1]
def resolve_model_dir(value: Path) -> Path:
"""将模型目录解析为项目根目录下的绝对路径。"""
path = value.expanduser()
if not path.is_absolute():
path = PROJECT_ROOT / path
path = path.resolve()
if not path.is_dir():
raise ValueError(f"OCR模型目录不存在: {path}")
return path
ModelDir = Annotated[Path, AfterValidator(resolve_model_dir)]
class Settings(BaseSettings):
"""应用配置"""
model_config = SettingsConfigDict(
env_file=PROJECT_ROOT / ".env",
case_sensitive=False,
extra="ignore",
)
host: str = "0.0.0.0"
port: int = 15651
debug: bool = False
# OCR阶段
ocr_use_det: bool = True
ocr_use_cls: bool = False
ocr_use_rec: bool = True
# 模型与常用调优参数
ocr_model_dir: ModelDir = Path("model")
ocr_text_score: float = Field(default=0.5, ge=0.0, le=1.0)
ocr_max_side_len: int = Field(default=2000, ge=32, le=16384)
ocr_det_limit_side_len: int = Field(default=736, ge=32, le=8192)
ocr_det_box_thresh: float = Field(default=0.5, ge=0.0, le=1.0)
ocr_det_thresh: float = Field(default=0.3, ge=0.0, le=1.0)
ocr_rec_batch_num: int = Field(default=6, ge=1, le=256)
# 推理设备
ocr_device: Literal["cpu", "cuda", "dml", "tensorrt"] = "cpu"
ocr_device_id: int = Field(default=0, ge=0)
ocr_intra_threads: int = Field(default=4, ge=1)
ocr_inter_threads: int = Field(default=2, ge=1)
# 竖排文字补充通道（边缘带旋转90°检测，需 model/PP-OCRv4 目录下的 server 模型）
ocr_enable_vertical: bool = False
settings = Settings()
engine/ocr_engine.py
+526
View File
@@ -0,0 +1,526 @@
"""
OCR引擎模块
使用 RapidOCR 对整张图片进行文字识别。
推理设备支持：CPU / CUDA / DirectML / TensorRT
"""
import numpy as np
import cv2
from rapidocr import EngineType, LangCls, LangDet, LangRec, ModelType, OCRVersion, RapidOCR
from rapidocr.utils.output import RapidOCROutput
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class OCRProcessor:
"""整图 OCR 处理器"""
def __init__(
self,
use_det: bool = True,
use_cls: bool = False,
use_rec: bool = True,
model_dir: str | Path = "model",
text_score: float = 0.5,
max_side_len: int = 2000,
det_limit_side_len: int = 736,
det_box_thresh: float = 0.5,
det_thresh: float = 0.3,
rec_batch_num: int = 6,
# 推理设备
device: str = "cpu",
device_id: int = 0,
intra_threads: int = 4,
inter_threads: int = 2,
# 竖排文字补充通道
enable_vertical: bool = False,
):
self.use_det = use_det
self.use_cls = use_cls
self.use_rec = use_rec
self.device = device
self.device_id = device_id
self.intra_threads = intra_threads
self.inter_threads = inter_threads
self.text_score = text_score
self.max_side_len = max_side_len
self.det_limit_side_len = det_limit_side_len
self.det_box_thresh = det_box_thresh
self.det_thresh = det_thresh
self.rec_batch_num = rec_batch_num
self.enable_vertical = enable_vertical
self._vert_engine = None
# 模型路径
model_path = Path(model_dir).resolve()
self.rec_model_path = model_path / "ch_PP-OCRv5_rec_mobile.onnx"
self.det_model_path = model_path / "ch_PP-OCRv5_det_mobile.onnx"
self.cls_model_path = model_path / "ch_PP-LCNet_x0_25_textline_ori_cls_mobile.onnx"
missing_models = [
str(path)
for path in (self.det_model_path, self.cls_model_path, self.rec_model_path)
if not path.is_file()
]
if missing_models:
raise FileNotFoundError(f"OCR模型文件不存在: {', '.join(missing_models)}")
# 初始化引擎
self.engine = self._init_engine()
def _build_device_params(self) -> dict:
"""根据 device 配置构建推理引擎参数"""
params = {}
if self.device == "cpu":
# CPU: 控制线程数以优化并发
params.update({
"EngineConfig.onnxruntime.intra_op_num_threads": self.intra_threads,
"EngineConfig.onnxruntime.inter_op_num_threads": self.inter_threads,
})
logger.info(f"推理设备: CPU (intra_threads={self.intra_threads}, inter_threads={self.inter_threads})")
elif self.device == "cuda":
# NVIDIA GPU with CUDA
params.update({
"EngineConfig.onnxruntime.use_cuda": True,
"EngineConfig.onnxruntime.cuda_ep_cfg.device_id": self.device_id,
})
logger.info(f"推理设备: CUDA (device_id={self.device_id})")
elif self.device == "dml":
# Windows DirectML (任意GPU)
params.update({
"EngineConfig.onnxruntime.use_dml": True,
})
logger.info("推理设备: DirectML")
elif self.device == "tensorrt":
# NVIDIA TensorRT (需要预先构建engine)
params.update({
"Det.engine_type": EngineType.TENSORRT,
"Rec.engine_type": EngineType.TENSORRT,
"Cls.engine_type": EngineType.TENSORRT,
"EngineConfig.tensorrt.device_id": self.device_id,
"EngineConfig.tensorrt.use_fp16": True,
})
logger.info(f"推理设备: TensorRT (device_id={self.device_id}, fp16=True)")
return params
def _init_engine(self) -> RapidOCR:
"""初始化OCR引擎"""
params = {
# 全局与常用调优参数
"Global.text_score": self.text_score,
"Global.max_side_len": self.max_side_len,
"Det.limit_side_len": self.det_limit_side_len,
"Det.box_thresh": self.det_box_thresh,
"Det.thresh": self.det_thresh,
"Rec.rec_batch_num": self.rec_batch_num,
# 模型契约
"Det.model_path": str(self.det_model_path),
"Det.engine_type": EngineType.ONNXRUNTIME,
"Det.lang_type": LangDet.CH,
"Det.model_type": ModelType.MOBILE,
"Det.ocr_version": OCRVersion.PPOCRV5,
"Rec.model_path": str(self.rec_model_path),
"Rec.engine_type": EngineType.ONNXRUNTIME,
"Rec.lang_type": LangRec.CH,
"Rec.model_type": ModelType.MOBILE,
"Rec.ocr_version": OCRVersion.PPOCRV5,
"Cls.engine_type": EngineType.ONNXRUNTIME,
"Cls.lang_type": LangCls.CH,
"Cls.model_type": ModelType.MOBILE,
"Cls.ocr_version": OCRVersion.PPOCRV5,
"Cls.model_path": str(self.cls_model_path),
}
# 合并设备参数
params.update(self._build_device_params())
return RapidOCR(params=params)
def process(self, image: np.ndarray) -> RapidOCROutput:
"""标准OCR识别（整图直接OCR）"""
return self.engine(
image,
use_det=self.use_det,
use_rec=self.use_rec,
use_cls=self.use_cls,
return_word_box=True
)
# ─── 边缘/竖排文字补充通道（纯 CV，CPU 友好）──────────
@staticmethod
def _row_bands(gray_img: np.ndarray, dark_thresh: int = 160, char_h: int | None = None):
"""行投影找文字行带（字高自适应）。
手写/毛笔大字内部笔画断裂会产生大间隙，固定空白阈值会把一个字拆成多段。
流程：小阈值粗切 → 按字高比例合并相邻带、丢弃碎片带。
char_h 为字高（px）；None 时自动估计（取主带——≥0.35×最大带高——的平均高，
避免被小噪声碎片带拉偏）。
返回 [(y0, y1), ...]（区域内坐标）
"""
row_dark = (gray_img < dark_thresh).sum(axis=1)
bands = []
in_b = False
for y, d in enumerate(row_dark):
if d > 2 and not in_b:
in_b = True
b0 = y
elif d <= 2 and in_b:
in_b = False
if y - b0 >= 2:
bands.append((b0, y))
if in_b:
bands.append((b0, len(row_dark)))
if not bands:
return []
if char_h is None:
hmax = max(y1 - y0 for y0, y1 in bands)
mains = [y1 - y0 for y0, y1 in bands if y1 - y0 >= hmax * 0.35]
char_h = round(sum(mains) / len(mains)) if mains else hmax
gap_min = max(2, int(char_h * 0.35)) # 相邻带空白 <= 该值 -> 笔画断裂，合并
band_min = max(4, int(char_h * 0.5)) # 低于该高度的带为噪声碎片，丢弃
merged = []
last = None
for y0, y1 in bands:
if y1 - y0 < band_min:
continue
if last is not None and y0 - last[1] <= gap_min:
last[1] = y1
else:
last = [y0, y1]
merged.append(last)
return [(y0, y1) for y0, y1 in merged]
def _cv_supplement(self, image: np.ndarray, main_boxes: list, band_h: int = 160) -> list:
"""边缘带补充检测（纯 OpenCV，无模型依赖，CPU 毫秒级）：
四边带 → 多级低阈值二值化 → 膨胀 → 连通域 → 文字状块过滤。
解决主流程漏检的边缘小字/竖排手写文字。
返回 [(box_np, text), ...]（原图坐标）"""
if not self.enable_vertical:
return []
H, W = image.shape[:2]
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
bands = [
(0, 0, W, min(band_h, H)), # 上边带
(0, max(0, H - band_h), W, H), # 下边带
(0, 0, min(band_h, W), H), # 左边带
(max(0, W - band_h), 0, W, H), # 右边带
]
extra = []
for bx1, by1, bx2, by2 in bands:
if bx2 - bx1 < 40 or by2 - by1 < 40:
continue
band = gray[by1:by2, bx1:bx2]
for th in (155, 145, 135):
_, bw = cv2.threshold(band, th, 255, cv2.THRESH_BINARY_INV)
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 5))
closed = cv2.morphologyEx(bw, cv2.MORPH_CLOSE, kernel, iterations=2)
contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
for c in contours:
x, y, w, h = cv2.boundingRect(c)
if not (8 <= w <= 110 and 10 <= h <= 130):
continue
if closed[y:y + h, x:x + w].mean() / 255.0 < 0.12:
continue
cx, cy = bx1 + x + w / 2, by1 + y + h / 2
# 块中心落在任一主流程检测框内（含4px容差）则跳过
if any(b[:, 0].min() - 4 <= cx <= b[:, 0].max() + 4
and b[:, 1].min() - 4 <= cy <= b[:, 1].max() + 4 for b in main_boxes):
continue
if any(abs(cx - eb[:, 0].mean()) < 10 and abs(cy - eb[:, 1].mean()) < 10 for eb, _ in extra):
continue
box = np.array([[x, y], [x + w, y], [x + w, y + h], [x, y + h]], dtype=float)
box[:, 0] += bx1
box[:, 1] += by1
text = self._cv_box_text(gray, box) if h / max(w, 1) > 1.2 else ""
extra.append((box, text))
return extra
def _cv_box_text(self, gray: np.ndarray, box: np.ndarray) -> str:
"""补充框内逐字识别（mobile rec，CPU 可跑）：行投影切单字 → 逐字识别。
识别失败返回空串（标注优先，文本对错次要）。"""
H, W = gray.shape[:2]
x1, y1 = int(max(0, box[:, 0].min() - 3)), int(max(0, box[:, 1].min() - 3))
x2, y2 = int(min(W, box[:, 0].max() + 3)), int(min(H, box[:, 1].max() + 3))
if x2 - x1 < 4 or y2 - y1 < 4:
return ""
region = gray[y1:y2, x1:x2]
text = ""
for c0, c1 in self._row_bands(region, 160):
seg = region[c0:c1, :]
if seg.shape[0] < 8:
continue
big = cv2.resize(seg, None, fx=2, fy=2, interpolation=cv2.INTER_LANCZOS4)
r = self.engine(cv2.cvtColor(big, cv2.COLOR_GRAY2RGB), use_det=False, use_cls=False, use_rec=True)
if r.txts:
text += "".join(r.txts)
return text
def _vertical_box_text(self, gray: np.ndarray, box: np.ndarray) -> str:
"""竖排框内逐字识别：行投影（阈值160）切单字 → 逐字 mobile rec"""
H, W = gray.shape[:2]
x1, y1 = int(max(0, box[:, 0].min() - 3)), int(max(0, box[:, 1].min() - 3))
x2, y2 = int(min(W, box[:, 0].max() + 3)), int(min(H, box[:, 1].max() + 3))
if x2 - x1 < 4 or y2 - y1 < 4:
return ""
region = gray[y1:y2, x1:x2]
text = ""
for c0, c1 in self._row_bands(region, 160):
seg = region[c0:c1, :]
if seg.shape[0] < 8:
continue
big = cv2.resize(seg, None, fx=2, fy=2, interpolation=cv2.INTER_LANCZOS4)
r = self.engine(cv2.cvtColor(big, cv2.COLOR_GRAY2RGB), use_det=False, use_cls=False, use_rec=True)
if r.txts:
text += "".join(r.txts)
return text
def _vertical_longbox_supplement(self, image: np.ndarray, result: RapidOCROutput) -> RapidOCROutput:
"""竖排长框旋转识别补充：
检测把竖排一列文字框成超长行（>300px），Rec 压缩后丢字（如"殁"）。
按列合并长框为竖条 → 旋转90°变横排 → 完整OCR → 行框换算回原图 →
行内均分字坐标 → 补充原 word_results 缺失的位置。"""
if result.boxes is None or not result.word_results:
return result
H, W = image.shape[:2]
# 超长竖排框（>300px 高且高宽比 >2）
def is_longbox(b):
return (b[:, 1].max() - b[:, 1].min() > 300
and (b[:, 1].max() - b[:, 1].min()) / max(b[:, 0].max() - b[:, 0].min(), 1) > 2)
def flat_items(word_results):
"""展平 word_results；rapidocr 对超长竖框的字级切分会把整列切成
碎片（半字高），这些框的字级条目整框丢弃，交给下方长框通道重切"""
out = []
for i, line in enumerate(word_results):
if isinstance(line, (tuple, list)) and len(line) >= 3 and isinstance(line[0], str):
out.append((line[0], line[1], line[2]))
elif isinstance(line, (tuple, list)):
if result.boxes is not None and i < len(result.boxes) and is_longbox(result.boxes[i]):
continue
for it in line:
if isinstance(it, (tuple, list)) and len(it) >= 3 and isinstance(it[0], str):
out.append((it[0], it[1], it[2]))
return out
def center(coords):
return sum(p[0] for p in coords) / len(coords), sum(p[1] for p in coords) / len(coords)
orig_items = flat_items(result.word_results)
# 超长竖排框（>300px 高且高宽比 >2）按 x 中心聚类合并为竖条
longs = [b for b in result.boxes if is_longbox(b)]
if not longs:
return result
# 聚类容差按列宽自适应: 硬编码 60px 会把列间距较密的竖排相邻列误并为一组
col_w = float(np.median([b[:, 0].max() - b[:, 0].min() for b in longs]))
cluster_d = max(2, int(col_w * 0.8))
groups = []
for b in longs:
cx = b[:, 0].mean()
for g in groups:
if abs(cx - g[0]) < cluster_d:
g[1] = min(g[1], b[:, 1].min()); g[2] = max(g[2], b[:, 1].max())
g[3] = min(g[3], b[:, 0].min()); g[4] = max(g[4], b[:, 0].max())
break
else:
groups.append([cx, b[:, 1].min(), b[:, 1].max(), b[:, 0].min(), b[:, 0].max()])
for cx, gy1, gy2, gx1, gx2 in groups:
x1, y1 = max(0, int(gx1)), max(0, int(gy1))
x2, y2 = min(W, int(gx2)), min(H, int(gy2))
if x2 - x1 < 10 or y2 - y1 < 50:
continue
crop = image[y1:y2, x1:x2]
_, chw = crop.shape[:2]
# 竖排(上→下)需逆时针转90°: 顺时针会字头朝左、字序反转, rec 无法识别
r90 = cv2.rotate(crop, cv2.ROTATE_90_COUNTERCLOCKWISE)
big = cv2.resize(r90, None, fx=2, fy=2, interpolation=cv2.INTER_CUBIC)
# 旋转后为横排: 列投影切字(按字高比例合并断笔) -> 逐字识别。
# 整列 rec + 均分会把毛笔大字切成两半输出, 这里用真实字块边界。
gb = cv2.cvtColor(big, cv2.COLOR_RGB2GRAY)
col_dark = (gb < 160).sum(axis=0)
blocks, in_b, b0 = [], False, 0
for x, d in enumerate(col_dark):
if d > 2 and not in_b:
in_b, b0 = True, x
elif d <= 2 and in_b:
in_b = False
if x - b0 >= 2:
blocks.append([b0, x])
if in_b:
blocks.append([b0, len(col_dark)])
if not blocks:
continue
char_h = big.shape[0] # 旋转后字高 ≈ 原列宽, 作为切字尺子
gap_min = max(2, int(char_h * 0.2)) # 断笔间隙 <= 该值 -> 同字, 合并
merged = [blocks[0]]
for blk in blocks[1:]:
if blk[0] - merged[-1][1] <= gap_min:
merged[-1][1] = blk[1]
else:
merged.append(blk)
# 字宽尺子: 合并后块宽的中位数(断笔已合并, 单字块为主)。
# 竖排字宽不等于列宽(Det 框可能框住1~2列), 不能用 char_h 直接当字宽;
# 连笔列会合并成单个巨型块污染中位数, 字宽也不可能超过列宽, 故取上界
widths = sorted(b[1] - b[0] for b in merged)
w_med = widths[len(widths) // 2] if widths else char_h
w_med = max(2, min(w_med, int(0.9 * char_h)))
# 断笔残块并入相邻块: 飞白/横笔间隙切出的残块明显窄于字宽,
# 合并后按字宽整块识别, 避免半个字独立成框
i = 0
while i < len(merged):
if merged[i][1] - merged[i][0] < 0.65 * w_med:
left_gap = merged[i][0] - merged[i - 1][1] if i > 0 else float("inf")
right_gap = merged[i + 1][0] - merged[i][1] if i < len(merged) - 1 else float("inf")
if left_gap <= right_gap and i > 0:
merged[i - 1][1] = merged[i][1]
del merged[i]
i -= 1
elif i < len(merged) - 1:
merged[i][1] = merged[i + 1][1]
del merged[i + 1]
else:
i += 1
else:
i += 1
# 超宽块拆开: 合并后宽 > 1.4×字宽说明粘了多个字, 在最大间隙处递归拆
def split_wide(blk, subs):
if blk[1] - blk[0] <= 1.4 * w_med or len(subs) < 2:
return [blk]
gaps = [(subs[i + 1][0] - subs[i][1], i) for i in range(len(subs) - 1)]
gi = max(range(len(gaps)), key=lambda i: gaps[i][0])
if gaps[gi][0] < 0.2 * w_med:
return [blk]
cut = (subs[gi][1] + subs[gi + 1][0]) // 2
left = [blk[0], cut]; right = [cut, blk[1]]
return (split_wide(left, [s for s in subs if s[1] <= cut])
+ split_wide(right, [s for s in subs if s[0] >= cut]))
final_blocks = []
for blk in merged:
subs = [s for s in blocks if s[0] >= blk[0] - 1 and s[1] <= blk[1] + 1]
final_blocks.extend(split_wide(blk, subs))
new_items = []
for bx0, bx1 in final_blocks:
if bx1 - bx0 < 8:
continue
# 连笔/倾斜导致投影切不开的超宽块: 按字宽等份切开, 逐份识别
if bx1 - bx0 > 1.4 * w_med:
n = max(2, round((bx1 - bx0) / w_med))
cuts = [bx0 + (bx1 - bx0) * i // n for i in range(n + 1)]
blocks_2 = [[cuts[i], cuts[i + 1]] for i in range(n)]
else:
blocks_2 = [[bx0, bx1]]
for cx0, cx1 in blocks_2:
r = self.engine(big[:, cx0:cx1], use_det=False, use_cls=False, use_rec=True)
txt = "".join(r.txts) if r.txts else ""
if not txt:
continue
# 逆时针旋转90°后: big 的 x' 方向(宽2×chh)对应 crop 的 y 方向,
# y' 方向(高2×chw)对应 crop 的 x 方向。
# crop(x_c, y_c) = (chw-1-y'/2, x'/2), 字块占满 y' 全高 -> x_c 全宽
oy1 = y1 + cx0 / 2; oy2 = y1 + cx1 / 2
ox1 = x1; ox2 = x1 + chw - 1
coords = [[int(ox1), int(oy1)], [int(ox2), int(oy1)],
[int(ox2), int(oy2)], [int(ox1), int(oy2)]]
cxx, cyy = center(coords)
if any(abs(cxx - center(c)[0]) <= 10 and abs(cyy - center(c)[1]) <= 10
for _, _, c in orig_items + new_items):
continue
new_items.append((txt, 0.5, coords))
if new_items:
orig_items = orig_items + new_items
logger.info(f"竖排长框旋转补充 {len(new_items)} 字 (列x~{cx:.0f})")
# 主流程整列条目若被逐字补充充分覆盖(>=2个字且覆盖>=50%面积),
# 则丢弃, 只保留字级框, 避免前端画出一堆叠加框
n_new = len(new_items)
if n_new:
main_part = orig_items[:len(orig_items) - n_new]
new_coords = [c for _, _, c in new_items]
def covered(item):
t, s, c = item
if not isinstance(t, str) or not t:
return False
x0, y0 = min(p[0] for p in c), min(p[1] for p in c)
x1, y1 = max(p[0] for p in c), max(p[1] for p in c)
area = (x1 - x0) * (y1 - y0)
if area <= 0:
return False
hits, hit_area = 0, 0
for nc in new_coords:
nx0, ny0 = min(p[0] for p in nc), min(p[1] for p in nc)
nx1, ny1 = max(p[0] for p in nc), max(p[1] for p in nc)
ix0, iy0 = max(x0, nx0), max(y0, ny0)
ix1, iy1 = min(x1, nx1), min(y1, ny1)
if ix1 > ix0 and iy1 > iy0:
hits += 1
hit_area += (ix1 - ix0) * (iy1 - iy0)
return hits >= 2 and hit_area >= 0.5 * area
main_part = [item for item in main_part if not covered(item)]
# 列内相对过滤: 明显矮于同列中位的 0.5 分字块是残笔/等份碎片, 删除
if len(new_items) >= 3:
by_col = {}
for it in new_items:
cxx = sum(p[0] for p in it[2]) / len(it[2])
by_col.setdefault(round(cxx / 30), []).append(it)
for key, its in by_col.items():
if len(its) < 3:
continue
hs = sorted(max(p[1] for p in c) - min(p[1] for p in c) for _, _, c in its)
med = hs[len(hs) // 2]
by_col[key] = [it for it in its
if max(p[1] for p in it[2]) - min(p[1] for p in it[2]) >= med * 0.5]
new_items = [it for its in by_col.values() for it in its]
orig_items = main_part + new_items
result.word_results = tuple(
(t, s, c) if isinstance(t, str) else (t[0], t[1], t[2])
for t, s, c in orig_items
)
return result
def process_with_vertical(self, image: np.ndarray) -> RapidOCROutput:
"""标准 OCR + 边缘/竖排文字补充通道（纯 CV + 长框旋转识别，合并检测框与文本）"""
result = self.process(image)
if not self.enable_vertical:
return result
main_boxes = list(result.boxes) if result.boxes is not None else []
extra = self._cv_supplement(image, main_boxes)
if extra:
merged = []
for box, text in extra:
cx, cy = box[:, 0].mean(), box[:, 1].mean()
# 中心落在主流程框内则跳过
if any(b[:, 0].min() - 4 <= cx <= b[:, 0].max() + 4
and b[:, 1].min() - 4 <= cy <= b[:, 1].max() + 4 for b in main_boxes):
continue
# 补充块之间去重（多阈值可能检出同一区域）
if any(abs(cx - mx) < 15 and abs(cy - my) < 15 for _, _, _, mx, my in merged):
continue
coords = [[int(p[0]), int(p[1])] for p in box]
merged.append((text, 1.0, coords, cx, cy))
if merged:
existing = list(result.word_results) if result.word_results else []
result.word_results = tuple(existing + [(t, s, c) for t, s, c, _, _ in merged])
logger.info(f"竖排通道补充 {len(merged)} 条: {[t for t, _, _, _, _ in merged]}")
# 竖排长框旋转补充（恢复超长行识别丢字）
result = self._vertical_longbox_supplement(image, result)
return result
main.py
+558
View File
@@ -0,0 +1,558 @@
"""
OCR Internal Service
提供标准整图 OCR 识别服务。
"""
import io
import asyncio
from typing import List, Optional
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from PIL import Image
import numpy as np
from datetime import datetime
import logging
from config.config import settings
from engine.ocr_engine import OCRProcessor
# 配置日志
logging.basicConfig(
level=logging.DEBUG if settings.debug else logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger(__name__)
app = FastAPI(title="OCR Internal Service", version="2.0.0")
# 初始化OCR处理器
ocr = OCRProcessor(
use_det=settings.ocr_use_det,
use_cls=settings.ocr_use_cls,
use_rec=settings.ocr_use_rec,
model_dir=settings.ocr_model_dir,
text_score=settings.ocr_text_score,
max_side_len=settings.ocr_max_side_len,
det_limit_side_len=settings.ocr_det_limit_side_len,
det_box_thresh=settings.ocr_det_box_thresh,
det_thresh=settings.ocr_det_thresh,
rec_batch_num=settings.ocr_rec_batch_num,
device=settings.ocr_device,
device_id=settings.ocr_device_id,
intra_threads=settings.ocr_intra_threads,
inter_threads=settings.ocr_inter_threads,
enable_vertical=settings.ocr_enable_vertical,
)
# ─── 响应模型 ─────────────────────────────────────────────
class TextRegion(BaseModel):
confidence: float
text: str
text_region: List[List[int]] # [[x1,y1], [x2,y2], [x3,y3], [x4,y4]]
class OCRResponse(BaseModel):
code: int
message: str
data: Optional[List[TextRegion]] = None
timestamp: str
# ─── 辅助函数 ─────────────────────────────────────────────
def _build_response(word_results) -> OCRResponse:
"""将 word_results 构建为响应，兼容嵌套和扁平两种格式"""
ocr_result = []
if not word_results:
return OCRResponse(
code=0, message="success", data=ocr_result,
timestamp=datetime.now().isoformat(),
)
for line in word_results:
if isinstance(line, (tuple, list)) and len(line) >= 3 and isinstance(line[0], str):
# 扁平 item: 单条 (text, conf, coords)，line 本身就是 item
text, confidence, coords = line[0], line[1], line[2]
if coords is not None:
ocr_result.append(TextRegion(
confidence=confidence, text=text, text_region=coords,
))
elif isinstance(line, (tuple, list)):
# 嵌套结构: line = ((item1), (item2), ...)
for item in line:
if isinstance(item, (tuple, list)) and len(item) >= 3:
text, confidence, coords = item[0], item[1], item[2]
if coords is not None:
ocr_result.append(TextRegion(
confidence=confidence, text=text, text_region=coords,
))
return OCRResponse(
code=0,
message="success",
data=ocr_result,
timestamp=datetime.now().isoformat(),
)
def _error_response(msg: str) -> OCRResponse:
return OCRResponse(
code=500,
message=msg,
timestamp=datetime.now().isoformat(),
)
# ─── API 端点 ─────────────────────────────────────────────
@app.get("/ocr_system/health")
async def health_check():
"""健康检查"""
return {
"status": "ok",
"service": "ocr",
"version": "2.0.0",
"device": ocr.device,
"timestamp": datetime.now().isoformat(),
}
@app.post("/ocr_system/ocr", response_model=OCRResponse)
async def ocr_recognition(file: UploadFile = File(...)):
"""上传图片文件并执行整图 OCR 识别。"""
try:
# 读取文件
image_bytes = await file.read()
image_file = io.BytesIO(image_bytes)
image = Image.open(image_file)
image_np = np.array(image)
logger.info(f"收到OCR请求: size={image_np.shape}")
# 预处理（原图直通：灰度化+CLAHE+锐化会损失边缘小字细节，已实测降低召回）
loop = asyncio.get_event_loop()
im_np = image_np
logger.info("使用标准OCR模式")
result = await loop.run_in_executor(None, ocr.process_with_vertical, im_np)
if result is None:
return _error_response("OCR处理失败")
return _build_response(result.word_results)
except Exception as e:
logger.error(f"OCR处理异常: {str(e)}", exc_info=True)
return _error_response(f"处理异常: {str(e)}")
# ─── 可视化页面 ───────────────────────────────────────────
@app.get("/ocr_system/viewer", response_class=HTMLResponse)
async def viewer():
"""OCR结果可视化页面"""
return HTMLResponse(content=VIEWER_HTML)
VIEWER_HTML = r"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>OCR 可视化工具</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
background: #1a1a2e;
color: #e0e0e0;
height: 100vh;
display: flex; flex-direction: column;
overflow: hidden;
}
.header {
background: #16213e;
padding: 12px 24px;
display: flex; align-items: center; gap: 16px;
border-bottom: 1px solid #0f3460;
flex-shrink: 0;
}
.header h1 { font-size: 18px; font-weight: 600; }
/* ── 三栏布局 ── */
.main-layout {
display: flex; flex: 1; min-height: 0;
}
/* 左栏：控件 */
.panel-left {
width: 280px; min-width: 280px;
background: #16213e;
padding: 16px;
display: flex; flex-direction: column; gap: 12px;
border-right: 1px solid #0f3460;
overflow-y: auto; flex-shrink: 0;
}
.section-title {
font-size: 12px; font-weight: 600; color: #a0a0b0;
text-transform: uppercase; letter-spacing: 1px;
padding-top: 4px;
}
.upload-zone {
border: 2px dashed #0f3460; border-radius: 10px;
padding: 24px 16px; text-align: center; cursor: pointer;
transition: all .2s;
}
.upload-zone:hover { border-color: #53d8fb; background: rgba(83,216,251,0.05); }
.upload-zone.dragover { border-color: #53d8fb; background: rgba(83,216,251,0.1); }
.upload-zone .icon { font-size: 32px; margin-bottom: 4px; }
.upload-zone .text { font-size: 13px; color: #a0a0b0; }
.upload-zone .hint { font-size: 11px; color: #606080; margin-top: 2px; }
.upload-zone input { display: none; }
.btn {
padding: 9px 14px; border: none; border-radius: 6px;
cursor: pointer; font-size: 13px; font-weight: 500;
transition: all .2s; width: 100%;
}
.btn-primary { background: #53d8fb; color: #1a1a2e; }
.btn-primary:hover { background: #7ce4ff; }
.btn-primary:disabled { background: #303050; color: #606080; cursor: not-allowed; }
.status {
font-size: 12px; padding: 8px 10px; border-radius: 6px;
background: #0f3460; word-break: break-all;
}
.status.success { background: #0a3d2e; color: #4ade80; }
.status.error { background: #3d0a0a; color: #f87171; }
.status.loading { background: #3d2e0a; color: #fbbf24; }
.info-grid {
display: grid; grid-template-columns: 1fr 1fr; gap: 6px;
font-size: 12px;
}
.info-item { background: #0f3460; padding: 6px 10px; border-radius: 4px; }
.info-item .label { color: #606080; font-size: 10px; }
.info-item .value { color: #e0e0e0; font-weight: 500; }
/* 中栏：OCR 画布区 */
.panel-center {
flex: 1; display: flex; flex-direction: column;
background: #0d1117; min-width: 0;
}
.canvas-pane {
flex: 1; display: flex; align-items: center; justify-content: center;
position: relative; overflow: hidden;
}
.canvas-pane .pane-label {
position: absolute; top: 6px; left: 8px;
font-size: 11px; padding: 2px 6px; border-radius: 3px;
background: rgba(0,0,0,0.6); color: #a0a0b0;
z-index: 2; pointer-events: none;
}
.canvas-pane canvas {
max-width: 100%; max-height: 100%;
object-fit: contain; border-radius: 2px;
}
/* 右栏：文本列表 */
.panel-right {
width: 280px; min-width: 280px;
background: #16213e; border-left: 1px solid #0f3460;
display: flex; flex-direction: column; flex-shrink: 0;
}
.panel-right .section-title { padding: 16px 16px 8px; }
#text-list {
flex: 1; overflow-y: auto; padding: 0 12px 12px;
font-size: 12px;
}
.text-item {
padding: 6px 8px; margin-bottom: 4px;
background: #0f3460; border-radius: 4px;
cursor: pointer; transition: all .12s;
display: flex; justify-content: space-between; align-items: center;
}
.text-item:hover { background: #1a3a70; }
.text-item.highlight { background: #1a4a80; border: 1px solid #53d8fb; }
.text-item .text { flex: 1; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.text-item .conf {
font-size: 10px; color: #53d8fb; margin-left: 8px; flex-shrink: 0;
}
.empty-state {
text-align: center; color: #606080;
position: absolute; top: 50%; left: 50%; transform: translate(-50%,-50%);
}
.empty-state .icon { font-size: 48px; margin-bottom: 8px; }
.empty-state p { font-size: 14px; }
@media (max-width: 1100px) {
.panel-right { width: 220px; min-width: 220px; }
.panel-left { width: 240px; min-width: 240px; }
}
</style>
</head>
<body>
<div class="header">
<h1>OCR 可视化工具</h1>
</div>
<div class="main-layout">
<!-- 左栏 -->
<div class="panel-left">
<div class="section-title">上传图片</div>
<div class="upload-zone" id="upload-zone">
<div class="icon">+</div>
<div class="text">点击或拖拽上传</div>
<div class="hint">JPG / PNG / BMP / TIFF</div>
<input type="file" id="file-input" accept="image/*">
</div>
<button class="btn btn-primary" id="btn-recognize" disabled>开始识别</button>
<div id="status-area"></div>
<div class="info-grid" id="info-grid" style="display:none;">
<div class="info-item"><div class="label">识别数</div><div class="value" id="info-count">-</div></div>
<div class="info-item"><div class="label">耗时</div><div class="value" id="info-elapsed">-</div></div>
<div class="info-item"><div class="label">尺寸</div><div class="value" id="info-size">-</div></div>
</div>
</div>
<!-- 中栏 -->
<div class="panel-center" id="panel-center">
<div class="canvas-pane" id="pane-text">
<div class="pane-label">文字文本框</div>
<div class="empty-state" id="empty-text">
<div class="icon">T</div>
<p>上传图片开始识别</p>
</div>
<canvas id="text-canvas" style="display:none;"></canvas>
</div>
</div>
<!-- 右栏 -->
<div class="panel-right">
<div class="section-title">识别文本</div>
<div id="text-list">
<div style="color:#606080;text-align:center;padding-top:40px;">暂无识别结果</div>
</div>
</div>
</div>
<script>
let ocrData = [];
let imageElement = null;
let isProcessing = false;
const uploadZone = document.getElementById('upload-zone');
const fileInput = document.getElementById('file-input');
const btnRecognize = document.getElementById('btn-recognize');
const textCanvas = document.getElementById('text-canvas');
const textList = document.getElementById('text-list');
const statusArea = document.getElementById('status-area');
const infoGrid = document.getElementById('info-grid');
const emptyText = document.getElementById('empty-text');
// ── 文件选择 ──
uploadZone.addEventListener('click', () => fileInput.click());
fileInput.addEventListener('change', (e) => handleFile(e.target.files[0]));
uploadZone.addEventListener('dragover', (e) => { e.preventDefault(); uploadZone.classList.add('dragover'); });
uploadZone.addEventListener('dragleave', () => uploadZone.classList.remove('dragover'));
uploadZone.addEventListener('drop', (e) => {
e.preventDefault(); uploadZone.classList.remove('dragover');
if (e.dataTransfer.files[0]) handleFile(e.dataTransfer.files[0]);
});
function handleFile(file) {
if (!file || !file.type.startsWith('image/')) {
showStatus('请选择图片文件', 'error'); return;
}
const reader = new FileReader();
reader.onload = (e) => {
imageElement = new Image();
imageElement.onload = () => {
ocrData = [];
textList.innerHTML = '<div style="color:#606080;text-align:center;padding-top:40px;">暂无识别结果</div>';
infoGrid.style.display = 'none';
drawBaseImages();
btnRecognize.disabled = false;
showStatus(`已加载: ${file.name} (${imageElement.width}x${imageElement.height})`, 'success');
};
imageElement.src = e.target.result;
};
reader.readAsDataURL(file);
}
function drawBaseImages() {
if (!imageElement) return;
drawCanvas(textCanvas, emptyText);
}
function drawCanvas(canvas, emptyEl) {
const pane = canvas.parentElement;
const maxW = pane.clientWidth * 0.92;
const maxH = pane.clientHeight * 0.88;
let w = imageElement.width, h = imageElement.height;
const scale = Math.min(maxW / w, maxH / h, 1);
w = Math.floor(w * scale); h = Math.floor(h * scale);
canvas.width = w; canvas.height = h;
const ctx = canvas.getContext('2d');
ctx.drawImage(imageElement, 0, 0, w, h);
canvas.style.display = 'block';
if (emptyEl) emptyEl.style.display = 'none';
}
// ── OCR ──
btnRecognize.addEventListener('click', async () => {
if (!imageElement || isProcessing) return;
isProcessing = true;
btnRecognize.disabled = true;
btnRecognize.textContent = '识别中...';
showStatus('正在识别...', 'loading');
ocrData = [];
textList.innerHTML = '<div style="color:#606080;text-align:center;padding-top:40px;">识别中...</div>';
infoGrid.style.display = 'none';
try {
const blob = await fetch(imageElement.src).then(r => r.blob());
const formData = new FormData();
formData.append('file', blob, 'image.png');
const t0 = performance.now();
const resp = await fetch('/ocr_system/ocr', { method: 'POST', body: formData });
const t1 = performance.now();
const json = await resp.json();
if (json.code !== 0) {
showStatus(`识别失败: ${json.message}`, 'error'); return;
}
ocrData = json.data || [];
const elapsed = (t1 - t0).toFixed(0);
document.getElementById('info-count').textContent = ocrData.length;
document.getElementById('info-elapsed').textContent = `${elapsed}ms`;
document.getElementById('info-size').textContent = `${imageElement.width}x${imageElement.height}`;
infoGrid.style.display = 'grid';
redrawTextCanvas();
renderTextList();
if (ocrData.length > 0) {
showStatus(`完成: ${ocrData.length}条 / ${elapsed}ms`, 'success');
} else {
showStatus('未识别到文字', 'error');
}
} catch (err) {
showStatus(`请求失败: ${err.message}`, 'error');
} finally {
isProcessing = false;
btnRecognize.disabled = false;
btnRecognize.textContent = '开始识别';
}
});
// ── 文字框画布 ──
function redrawTextCanvas() {
if (!imageElement) return;
drawCanvas(textCanvas, emptyText);
if (ocrData.length === 0) return;
const ctx = textCanvas.getContext('2d');
const scaleX = textCanvas.width / imageElement.width;
const scaleY = textCanvas.height / imageElement.height;
const lw = Math.max(1, 1.5 * (textCanvas.width / 800));
ocrData.forEach((item, i) => {
const coords = item.text_region;
if (!coords || coords.length < 4) return;
ctx.strokeStyle = `hsl(${(i * 37) % 360}, 70%, 65%)`;
ctx.lineWidth = lw;
ctx.beginPath();
ctx.moveTo(coords[0][0] * scaleX, coords[0][1] * scaleY);
for (let j = 1; j < coords.length; j++) {
ctx.lineTo(coords[j][0] * scaleX, coords[j][1] * scaleY);
}
ctx.closePath();
ctx.stroke();
});
}
// ── 文本列表 ──
function renderTextList() {
if (ocrData.length === 0) {
textList.innerHTML = '<div style="color:#606080;text-align:center;padding-top:40px;">未识别到文字</div>';
return;
}
textList.innerHTML = ocrData.map((item, i) => `
<div class="text-item" data-index="${i}" onclick="highlightBox(${i})">
<span class="text">${escapeHtml(item.text)}</span>
<span class="conf">${(item.confidence * 100).toFixed(1)}%</span>
</div>
`).join('');
}
let highlightedIndex = -1;
function highlightBox(index) {
redrawTextCanvas();
const ctx = textCanvas.getContext('2d');
const scaleX = textCanvas.width / imageElement.width;
const scaleY = textCanvas.height / imageElement.height;
const coords = ocrData[index].text_region;
ctx.strokeStyle = '#ff6b6b';
ctx.lineWidth = 3;
ctx.beginPath();
ctx.moveTo(coords[0][0] * scaleX, coords[0][1] * scaleY);
for (let j = 1; j < coords.length; j++) {
ctx.lineTo(coords[j][0] * scaleX, coords[j][1] * scaleY);
}
ctx.closePath();
ctx.stroke();
document.querySelectorAll('.text-item').forEach(el => el.classList.remove('highlight'));
const target = document.querySelector(`.text-item[data-index="${index}"]`);
if (target) target.classList.add('highlight');
target?.scrollIntoView({ behavior: 'smooth', block: 'nearest' });
}
function escapeHtml(text) {
const d = document.createElement('div');
d.textContent = text;
return d.innerHTML;
}
function showStatus(msg, type) {
statusArea.innerHTML = `<div class="status ${type}">${msg}</div>`;
}
window.addEventListener('resize', () => {
if (imageElement && !isProcessing) {
redrawTextCanvas();
}
});
</script>
</body>
</html>
"""
# ─── 启动入口 ─────────────────────────────────────────────
if __name__ == "__main__":
import uvicorn
logger.info(f"启动OCR服务: {settings.host}:{settings.port}")
uvicorn.run(
"main:app",
host=settings.host,
port=settings.port,
reload=settings.debug
)
requirements.txt
+9
View File
@@ -0,0 +1,9 @@
rapidocr==3.8.1
opencv-python>=4.10.0.84
numpy>=2.1.2
pillow>=12.2.0
uvicorn>=0.32.0
fastapi>=0.115.3
pydantic>=2.9.2
pydantic-settings>=2.14.1
python-multipart
utils/image_utils.py
+212
View File
@@ -0,0 +1,212 @@
from PIL import Image
from PIL.Image import Image
import cv2
import numpy as np
from PIL import Image
def split_image(image_array: np.ndarray, dirt: int) -> (np.ndarray, np.ndarray):
# 确保图像是二维的灰度图
if len(image_array.shape) > 2:
# 如果是彩色图，转换为灰度图 (取平均值或者只取一个通道)
image_array = image_array.mean(axis=2) # 或者直接取第一个通道：image_array = image_array[:, :, 0]
# 将图片转为二值化（假设白色背景值为255，黑色文字值为0）
# 你可以根据图片实际情况调整阈值
_, binary_image = cv2.threshold(image_array, 127, 255, cv2.THRESH_BINARY)
# 寻找切割线
split_line = find_split_line(binary_image, dirt)
if dirt == 1:
# 竖着切
split1 = image_array[:, :split_line]
split2 = image_array[:, split_line:]
else:
# 横着切
split1 = image_array[:split_line, :]
split2 = image_array[split_line:, :]
return split1, split2
def find_split_line(image_array, dirt):
height, width = image_array.shape # 现在应该是 (height, width)
# 根据dirt值选择切割方向
if dirt == 1:
# 竖着切
middle = width // 2
for x in range(middle, width):
# 检查竖直方向的列是否全为空白
if np.all(image_array[:, x] > 200): # 假设200是空白阈值
return x
return middle # 如果找不到合适的切割线，返回中间
else:
# 横着切
middle = height // 2
for y in range(middle, height):
# 检查水平方向的行是否全为空白
if np.all(image_array[y, :] > 200): # 假设200是空白阈值
return y
return middle # 如果找不到合适的切割线，返回中间
async def image_split(image: np.ndarray, dirt: int):
part1, part2 = split_image(image, dirt)
# 将 NumPy 数组转换为 Pillow 图像对象
pil_image1 = Image.fromarray(part1)
pil_image2 = Image.fromarray(part2)
# 显示图像
pil_image1.show()
pil_image2.show()
return pil_image1, pil_image2
async def get_able_area(image, coords, text):
"""
根据给定的文本区域坐标，从图像中截取文本区域。
:param text:
:param image: 原始图像 (使用cv2.imread读取的图像)
:param coords: 文本区域的四个角坐标，格式为：
[[x1_top_left, y1_top_left],
[x2_top_right, y2_top_right],
[x3_bottom_right, y3_bottom_right],
[x4_bottom_left, y4_bottom_left]]
:return: 截取的文本区域图像
"""
# 将坐标转换为numpy数组并重新整理为 [[x1, y1], [x2, y2], [x3, y3], [x4, y4]] 的形式
pts = np.float32([coords[0], coords[1], coords[2], coords[3]])
# 计算变换后的目标矩形的宽和高
top_width = np.linalg.norm(np.array(coords[1]) - np.array(coords[0])) # 顶边长度
bottom_width = np.linalg.norm(np.array(coords[2]) - np.array(coords[3])) # 底边长度
height = np.linalg.norm(np.array(coords[2]) - np.array(coords[1])) # 高
# 取最大宽度作为目标宽度
max_width = max(top_width, bottom_width)
# 目标矩形的四个角点
dst = np.float32([[0, 0], [max_width - 1, 0], [max_width - 1, height - 1], [0, height - 1]])
# 计算透视变换矩阵
M = cv2.getPerspectiveTransform(pts, dst)
# 进行透视变换，获取裁剪后的图像
cropped_image = cv2.warpPerspective(image, M, (int(max_width), int(height)))
await get_box_size(cropped_image, text)
async def get_box_size(image, text):
# 预处理图像
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
_, binary = cv2.threshold(gray, 128, 255, cv2.THRESH_BINARY_INV)
# 查找轮廓
contours, _ = cv2.findContours(binary, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
# 存储所有文字大小
text_sizes = []
# 遍历每个轮廓
for cnt in contours:
# 获取包围盒的坐标和尺寸
x, y, w, h = cv2.boundingRect(cnt)
# 计算文字区域的面积
area = cv2.contourArea(cnt)
# 假设每个文字区域有一个平均字符数 (可以根据实际情况调整)
avg_char_width = w // len(text) # 假设每行有10个字符
avg_char_height = h // 1 # 假设每行有1个字符
# 计算字符的大小
character_size = (avg_char_width, avg_char_height)
text_sizes.append(character_size)
# 可视化结果
cv2.rectangle(image, (x, y), (x + h, y + h), (0, 255, 0), 2)
# 计算平均字符大小
if len(text_sizes) > 0:
avg_char_width = sum(size[0] for size in text_sizes) / len(text_sizes)
avg_char_height = sum(size[1] for size in text_sizes) / len(text_sizes)
print(f"平均字符宽度: {avg_char_width}, 平均字符高度: {avg_char_height}")
else:
print("未检测到文字区域")
# 显示结果图像
im = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
il_image1 = Image.fromarray(im)
il_image1.show()
#增强图片
async def positive_image(image):
# 1. 灰度化（注意：输入为 RGB，须用 COLOR_RGB2GRAY；误用 BGR 会翻转红蓝通道，彩色文字/背景灰度失真）
gray_img = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
# 2. 对比度增强（CLAHE）
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
enhanced = clahe.apply(gray_img)
# 3. 锐化（可选）
blurred = cv2.GaussianBlur(enhanced, (0, 0), 2)
sharpened = cv2.addWeighted(enhanced, 1.5, blurred, -0.5, 0)
return sharpened
def draw_text_box(image, coords):
pos1 = coords[0]
pos2 = coords[2]
x1 = int(pos1[0])
y1 = int(pos1[1])
x2 = int(pos2[0])
y2 = int(pos2[1])
# 可视化结果
cv2.rectangle(image, (x1, y1), (x2, y2), (0, 255, 0), 1)
# 倾斜矫正
async def tilt_correction(image):
# 转为灰度图
gray_img = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
# 二值化
_, binary_img = cv2.threshold(gray_img, 128, 255, cv2.THRESH_BINARY)
# 边缘检测
edges = cv2.Canny(binary_img, 50, 150)
# 提取轮廓
contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if len(contours) == 0:
print("No contours found.")
return image # 没有找到轮廓，返回原图
# 找到最大轮廓
largest_contour = max(contours, key=cv2.contourArea)
# 计算最小外接矩形的角度
rect = cv2.minAreaRect(largest_contour)
angle = rect[-1]
# 修正角度
if angle < -45:
angle = -(90 + angle) # 修正角度，使其在 [-90, 90] 范围内
else:
angle = -angle # 直接取反号
# 如果角度接近0度，则不进行旋转
if abs(angle) < 0.5:
return image # 角度非常小，几乎不需要旋转
# 旋转图像
(h, w) = image.shape[:2] # 使用原始图像的形状
center = (w // 2, h // 2)
M = cv2.getRotationMatrix2D(center, angle, 1.0)
rotated_img = cv2.warpAffine(image, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)
return rotated_img
utils/utils.py
+116
View File
@@ -0,0 +1,116 @@
import math
from PIL import Image as pli
from PIL.Image import Image
import cv2
import numpy as np
from PIL import Image
def distance(p1, p2):
return math.sqrt((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2)
def find_edge_lengths(points):
# 计算所有可能的边的长度
edge_lengths = []
for i in range(len(points)):
for j in range(i + 1, len(points)):
edge_lengths.append(distance(points[i], points[j]))
return edge_lengths
def find_opposite_edges(edge_lengths):
# 对边的长度应该是成对的
edge_lengths.sort()
return edge_lengths
async def determine_orientation(points):
# 提取所有点的x和y坐标
x_coords = [p[0] for p in points]
y_coords = [p[1] for p in points]
# 计算x和y坐标的最大值和最小值
x_max, x_min = max(x_coords), min(x_coords)
y_max, y_min = max(y_coords), min(y_coords)
# 计算宽度和高度
width = x_max - x_min
height = y_max - y_min
edge_lengths = find_edge_lengths(points)
# 计算对边长度
opposite_edges = find_opposite_edges(edge_lengths)
# 判断竖着还是横着
if height > width:
return 1, opposite_edges
else:
return 0, opposite_edges
async def split_text_box(direction, opposite_edges, text_content, points):
# 左上，右上，右下，左下
point_lu_x, point_lu_y = points[0]
point_ru_x, point_ru_y = points[1]
point_rd_x, point_rd_y = points[2]
point_ld_x, point_ld_y = points[3]
if direction == 0:
# 文本框 长度
box_length_l = point_ru_x - point_lu_x
box_length_r = point_ld_x - point_rd_x
text_length = len(text_content)
each_text_size = (box_length_l + box_length_r) / 2 / text_length
min_size = min(opposite_edges)
cha = min_size / 6
print(f'差值：{cha}')
word_box = []
point_lu_x_sp = point_lu_x
point_ru_x_sp = point_ru_x
point_rd_x_sp = point_rd_x
point_ld_x_sp = point_ld_x
for char in text_content:
word_const = {
"text": char,
"box": [[point_lu_x_sp, point_lu_y],
[point_ru_x_sp, point_ru_y],
[point_rd_x_sp, point_rd_y],
[point_ld_x_sp, point_ld_y]]
}
point_lu_x_sp += min_size + cha
point_ru_x_sp += min_size + cha
point_rd_x_sp += min_size + cha
point_ld_x_sp += min_size + cha
word_box.append(word_const)
return word_box
elif direction == 1:
# 文本框 长度
box_length_l = point_ru_y - point_lu_y
box_length_r = point_ld_y - point_rd_y
text_length = len(text_content)
each_text_size = (box_length_l + box_length_r) / 2 / text_length
min_size = min(opposite_edges)
cha = min_size / 5
print(f'差值：{cha}')
word_box = []
point_lu_y_sp = point_lu_y
point_ru_y_sp = point_ru_y
point_rd_y_sp = point_rd_y
point_ld_y_sp = point_ld_y
for char in text_content:
word_const = {
"text": char,
"box": [[point_lu_x, point_lu_y_sp],
[point_ru_x, point_ru_y_sp],
[point_rd_x, point_rd_y_sp],
[point_ld_x, point_ld_y_sp]]
}
point_lu_y_sp += min_size + cha
point_ru_y_sp += min_size + cha
point_rd_y_sp += min_size + cha
point_ld_y_sp += min_size + cha
word_box.append(word_const)
return word_box