PokemonCardSearch/code/serve_lang_judge.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
语种/方向判断 sidecar 服务（子系统 C）—— PaddleOCR 常驻 HTTP 服务

用途：
  主识别接口 serve_recognition_api.py 运行在 pytorch 环境，无法 import PaddleOCR
  （paddleocr==3.2.0 与 pytorch 环境的包互不兼容，见框架文档第三之二节）。
  故把「OCR + 语种判定 + 方向角」拆成独立进程，跑在 paddleocr conda 环境，
  由主接口通过 HTTP 调用。两模型各自常驻内存，单图实时延迟最低。

运行环境：paddleocr conda 环境（paddleocr 3.2.0 + paddlepaddle-gpu 3.2.0）
启动：
  CUDA_VISIBLE_DEVICES=<空闲V100 UUID> \
    ~/miniconda3/envs/paddleocr/bin/python serve_lang_judge.py --port 8100

接口：
  POST /lang_judge   multipart 上传字段 image=<YOLO裁剪图>
      → {"language": "tcg us"|"tcg jp"|"简中"|"繁中"|null,
         "angle": 0|90|180|270,
         "detail": {...字符统计...},
         "text": "OCR文本(≤500字符)"}
  GET  /health       → {"status": "ok"}

依赖：pip install flask   （paddleocr / hanzidentifier 环境已有）

说明：核心 ocr_full_text() 与 judge() 逻辑直接取自本框架 ocr_judge.py:40-95，
      原样保留判定规则（假名≥10→jp；cjk*3>latin→中文再判简繁；否则→us），
      仅去掉其面向 CSV 批处理的主循环，改为按请求单张处理。不改动 ocr_judge.py。
"""
import os
import sys
import argparse
import tempfile
import traceback

# 强制 UTF-8 输出（Windows 控制台默认 GBK，打印中文会报错）
sys.stdout.reconfigure(encoding="utf-8")

MAX_TEXT_LEN = 500  # OCR 文本最大字符数（与 ocr_judge.py 一致）

# ==================== 启动时一次性加载 PaddleOCR（常驻内存）====================
print("[sidecar] 初始化 PaddleOCR 3.2.0 ...", flush=True)
from paddleocr import PaddleOCR

# use_doc_orientation_classify=True 必要：关闭时旋转卡 OCR 会乱码；开启后顺带给出方向角
ocr = PaddleOCR(
    use_doc_orientation_classify=True,
    use_doc_unwarping=False,
    use_textline_orientation=True,
)
print("[sidecar] PaddleOCR 加载完成", flush=True)

import hanzidentifier as hz


# ==================== OCR 文本 + 方向角（取自 ocr_judge.py:40-69）====================
def ocr_full_text(img_path):
    """OCR 识别，返回 (完整文本, 方向角度)。
    角度来自 doc_preprocessor_res.angle（0/90/180/270，输入相对正立顺时针的旋转）。"""
    result = ocr.predict(img_path)
    texts, angle = [], 0
    for res in result:
        if isinstance(res, dict):
            t = res.get("rec_texts")
            if t:
                texts.extend(list(t))
            dpr = res.get("doc_preprocessor_res")
            if dpr is not None:
                a = dpr.get("angle") if isinstance(dpr, dict) else getattr(dpr, "angle", None)
                if a is not None:
                    try:
                        angle = int(a)
                    except Exception:
                        angle = a
        elif hasattr(res, "rec_texts") and res.rec_texts:
            texts.extend(list(res.rec_texts))
            dpr = getattr(res, "doc_preprocessor_res", None)
            if dpr is not None:
                a = getattr(dpr, "angle", None)
                if a is not None:
                    try:
                        angle = int(a)
                    except Exception:
                        angle = a
    return "".join(texts), angle


# ==================== 语种判定（取自 ocr_judge.py:76-95）====================
def judge(text):
    """语种判定逻辑:
       ① 假名>=10 → tcg jp (日文铁证)
       ② 中文占比>25% (cjk*3 > latin) → 是中文卡 → hanzidentifier 判简繁
       ③ 否则 (cjk*3 <= latin) → tcg us (英文)
    卡牌上英文元素(HP/Weakness/版权/编号)天然拉高latin，故中文阈值放宽到25%。"""
    kana = sum(1 for ch in text if 0x3040 <= ord(ch) < 0x3100)
    cjk = sum(1 for ch in text if 0x4E00 <= ord(ch) < 0x9FFF)
    latin = sum(1 for ch in text if ch.isascii() and ch.isalpha())
    if kana >= 10:
        return "tcg jp", {"kana": kana, "cjk": cjk, "latin": latin}
    if cjk * 3 > latin:
        s = sum(1 for ch in text if hz.identify(ch) == 2)  # 简体字
        t = sum(1 for ch in text if hz.identify(ch) == 1)  # 繁体字
        script = "简中" if s >= t else "繁中"
        return script, {"kana": kana, "s": s, "t": t, "cjk": cjk}
    return "tcg us", {"kana": kana, "latin": latin, "cjk": cjk}


def judge_image(img_path):
    """单张裁剪图 → {language, angle, detail, text}。异常时 language=None。"""
    try:
        full_text, angle = ocr_full_text(img_path)
        text = full_text[:MAX_TEXT_LEN]
        lang, det = judge(text)
        return {"language": lang, "angle": angle, "detail": det, "text": text}
    except Exception as e:
        traceback.print_exc()
        return {"language": None, "angle": 0, "detail": {"err": str(e)[:120]}, "text": ""}


# ==================== HTTP 服务（Flask）====================
def create_app():
    from flask import Flask, request, jsonify
    app = Flask(__name__)

    @app.get("/health")
    def health():
        return jsonify({"status": "ok"})

    @app.post("/lang_judge")
    def lang_judge():
        """接收 multipart 图片(字段名 image)，OCR + 判语种/方向，返回 JSON。"""
        up = request.files.get("image")
        if not up or not up.filename:
            return jsonify({"error": "缺少图片：请以 multipart 字段 image 上传"}), 400
        blob = up.read()
        if not blob:
            return jsonify({"error": "上传文件为空"}), 400
        # 存临时文件供 PaddleOCR predict（推理后删）
        fd, tmp = tempfile.mkstemp(suffix=".jpg")
        try:
            with os.fdopen(fd, "wb") as f:
                f.write(blob)
            return jsonify(judge_image(tmp))
        finally:
            if os.path.exists(tmp):
                try:
                    os.remove(tmp)
                except Exception:
                    pass

    return app


if __name__ == "__main__":
    p = argparse.ArgumentParser(description="语种/方向判断 sidecar (PaddleOCR)")
    p.add_argument("--host", default="127.0.0.1", help="监听地址，默认 127.0.0.1（仅本机主接口调用）")
    p.add_argument("--port", type=int, default=8100, help="监听端口，默认 8100")
    args = p.parse_args()
    print("[sidecar] 就绪，监听 %s:%d，等待请求" % (args.host, args.port), flush=True)
    create_app().run(host=args.host, port=args.port, threaded=True)