基于docker+python+paddleocr构建自己本地化ocr服务

1、使用FastAPI创建服务实例

1.1、正常程序

from fastapi import FastAPI, UploadFile, File, HTTPException
from typing import List
from paddleocr import PaddleOCR
import numpy as np
from PIL import Image
import io
import logging

app = FastAPI(title="游戏截图OCR识别接口")
ocr = PaddleOCR(lang='ch', use_angle_cls=True)  # 启用方向分类器
logger = logging.getLogger("uvicorn.error")

@app.post("/ocrImages")
async def process_game_screenshots(
    images: List[UploadFile] = File(...,description="上传游戏截图（支持PNG/JPG格式）")
):
    """
    游戏截图文本识别接口
    - 支持同时上传最多10张截图
    - 单文件大小限制：5MB
    - 响应时间：约2秒/张（取决于文本复杂度）
    """
    results = []
    
    for screenshot in images:
        try:
            # 验证文件类型
            if screenshot.content_type not in ["image/jpeg", "image/png"]:
                raise HTTPException(400, detail=f"不支持的文件格式: {screenshot.filename}")

            # 读取并验证文件大小
            contents = await screenshot.read()
            if len(contents) > 5 * 1024 * 1024:
                raise HTTPException(413, detail=f"文件过大: {screenshot.filename}")

            # 转换图像格式
            img = Image.open(io.BytesIO(contents)).convert('RGB')
            img_array = np.array(img)

            # 执行OCR识别
            ocr_result = ocr.ocr(img_array, cls=True)
            
            # 提取文本内容（过滤低置信度结果）
            texts = []
            if ocr_result:
                for page in ocr_result:
                    if page:  # 忽略空页
                        texts.extend(line[1][0] for line in page if line[1][1] > 0.6)

            results.append({
                "filename": screenshot.filename,
                "texts": texts
            })

        except HTTPException as he:
            logger.warning(f"验证失败: {he.detail}")
            results.append({
                "filename": screenshot.filename,
                "error": he.detail
            })
        except Exception as e:
            logger.error(f"处理失败: {screenshot.filename} - {str(e)}")
            results.append({
                "filename": screenshot.filename,
                "error": "内部处理错误"
            })
    
    return {"results": results}

# 运行服务（生产环境建议使用uvicorn命令行启动）
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8888)

1.2、考虑并发

from fastapi import FastAPI, UploadFile, File, HTTPException
from typing import List
from paddleocr import PaddleOCR
import numpy as np
from PIL import Image
import io
import logging
import asyncio

app = FastAPI(title="游戏截图OCR识别接口")
logger = logging.getLogger("uvicorn.error")

# 调整OCR配置，根据实际情况启用GPU或MKLDNN
ocr = PaddleOCR(
    lang='ch',
    use_angle_cls=True,
    use_gpu=False,  # 根据环境调整，若无GPU则设为False
    det_limit_side_len=1600,
    det_db_thresh=0.4,
    det_db_box_thresh=0.6,
    rec_batch_num=5,
    enable_mkldnn=True  # 若CPU不支持MKLDNN则禁用
)

async def async_ocr(img_array: np.ndarray):
    """异步OCR处理，添加线程锁确保安全"""
    loop = asyncio.get_event_loop()
    # 使用线程锁（需导入threading）
    from threading import Lock
    lock = Lock()
    with lock:  # 确保同一时间仅一个线程调用OCR
        return await loop.run_in_executor(
            None, 
            lambda: ocr.ocr(img_array, cls=True)
        )

def preprocess_image(img: Image.Image) -> np.ndarray:
    # 保持不变
    w, h = img.size
    if max(w, h) > 1600:
        scale = 1600 / max(w, h)
        img = img.resize((int(w*scale), int(h*scale)), Image.Resampling.LANCZOS)
    return np.array(img)

@app.post("/ocrImages")
async def process_game_screenshots(
    images: List[UploadFile] = File(...,description="上传游戏截图（支持PNG/JPG格式）")
):
    results = []
    tasks = []
    for screenshot in images:
        if screenshot.content_type not in ["image/jpeg", "image/png"]:
            results.append({"filename": screenshot.filename, "error": "不支持的格式"})
            continue
        tasks.append(process_single_image(screenshot))
    
    # 限制最大并发数，例如设置为2
    from concurrent.futures import ThreadPoolExecutor
    executor = ThreadPoolExecutor(max_workers=2)
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 处理异常
    processed_results = []
    for res in results:
        if isinstance(res, Exception):
            logger.error(f"处理异常: {str(res)}")
            processed_results.append({"error": "内部错误"})
        else:
            processed_results.append(res)
    return {"results": processed_results}

async def process_single_image(screenshot: UploadFile):
    try:
        contents = await screenshot.read()
        if len(contents) > 5 * 1024 * 1024:
            raise HTTPException(413, detail="文件过大")
        
        img = Image.open(io.BytesIO(contents)).convert('RGB')
        img_array = preprocess_image(img)
        
        ocr_result = await async_ocr(img_array)
        
        texts = []
        for page in ocr_result:
            if page:
                texts.extend(
                    line[1][0] 
                    for line in page 
                    if line[1][1] > 0.65
                )
        
        return {
            "filename": screenshot.filename,
            "texts": texts,
            "warning": "检测到低精度内容" if len(texts)<3 else None
        }
    
    except Exception as e:
        logger.error(f"处理失败: {screenshot.filename} - {str(e)}")
        return {"filename": screenshot.filename, "error": "处理失败"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8888)

2、使用Docker打包镜像

2.1、编写Dockerfile

# 使用 Python 基础镜像
# 第一阶段：构建环境
FROM python:3.9-slim as builder

# 创建APT源文件
RUN echo "deb http://mirrors.aliyun.com/debian/ bullseye main non-free contrib" > /etc/apt/sources.list && \
    echo "deb http://mirrors.aliyun.com/debian-security bullseye-security main" >> /etc/apt/sources.list && \
    echo "deb http://mirrors.aliyun.com/debian/ bullseye-updates main" >> /etc/apt/sources.list

# 安装基础工具
RUN apt-get update && apt-get install -y \
    ca-certificates \
    gnupg2 \
    && apt-get clean

# 安装编译依赖
RUN apt-get update && apt-get install -y \
    gcc \
    python3-dev \
    libgl1-mesa-glx \
    libglib2.0-0 \
    libgomp1 \
    libsm6 \
    libxext6 \
    libxrender-dev \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .

# 使用国内PyPI镜像
RUN pip install --user --no-cache-dir -r requirements.txt \
    -i https://pypi.tuna.tsinghua.edu.cn/simple \
    --trusted-host pypi.tuna.tsinghua.edu.cn

# 第二阶段：生产镜像
FROM python:3.9-slim

# 初始化生产环境源文件
RUN echo "deb http://mirrors.aliyun.com/debian/ bullseye main non-free contrib" > /etc/apt/sources.list && \
    echo "deb http://mirrors.aliyun.com/debian-security bullseye-security main" >> /etc/apt/sources.list && \
    echo "deb http://mirrors.aliyun.com/debian/ bullseye-updates main" >> /etc/apt/sources.list

# 安装运行时依赖
RUN apt-get update && apt-get install -y \
    libgl1-mesa-glx \
    libglib2.0-0 \
    libgomp1 \
    && rm -rf /var/lib/apt/lists/*

# 复制构建结果
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

# 设置时区
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone

# 复制应用代码
COPY . .

EXPOSE 6012

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "6012"]

2.2 requirements.txt

fastapi
uvicorn
python-multipart
paddlepaddle
paddleocr
opencv-python-headless

2.3 构建镜像&运行容器

# 构建镜像（使用国内缓存）
docker build -t ocr-server:v1 .

启动

docker run -d -p 6012:6012 \
  --name ocr-service \
  --restart always \
  ocr-server:v1

目录结构
your_project/
├── app.py # FastAPI 主程序
├── Dockerfile # Docker 构建文件
└── requirements.txt # 依赖列表