From 86b19d1e071e3ef1d08467496cb34f2ace979277 Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Tue, 7 Jul 2026 05:47:55 -0400
Subject: [PATCH] =?UTF-8?q?feat(adapters):=20OCRProvider=20Protocol=20+=20?=
 =?UTF-8?q?MonkeyOCRClient=20=E5=BC=82=E6=AD=A5=E9=80=82=E9=85=8D=E5=99=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- app/ports.py: 新增 OCRProvider Protocol（runtime_checkable，与
  EmbeddingProvider 同级），定义 async transcribe_frames 端口
- adapters/ocr.py: 从 TRM4 core/tree/ocr.py 保真迁移 MonkeyOCRClient
  - assert → ValueError（P5 防御性校验）
  - 公开方法改 async（asyncio.to_thread 包装同步 HTTP）
  - 内部逻辑不变：多端点轮询、线程安全 Session、单帧降级、行去重
- tests/unit/test_ocr_adapter.py: 17 个测试覆盖 Protocol 合规、
  构造校验、健康检查、转录、降级、去重、轮询

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 adapters/ocr.py                | 128 ++++++++++++++
 app/ports.py                   |  18 ++
 tests/unit/test_ocr_adapter.py | 302 +++++++++++++++++++++++++++++++++
 3 files changed, 448 insertions(+)
 create mode 100644 adapters/ocr.py
 create mode 100644 tests/unit/test_ocr_adapter.py

diff --git a/adapters/ocr.py b/adapters/ocr.py
new file mode 100644
index 0000000..0a2fbc3
--- /dev/null
+++ b/adapters/ocr.py
@@ -0,0 +1,128 @@
+"""MonkeyOCR HTTP 客户端 — 帧文字转录的异构硬证据源。
+
+服务由用户在 LAN 部署（双端点轮询）；请求必须绕过代理（trust_env=False）。
+实现 OCRProvider Protocol（app/ports.py）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import itertools
+import threading
+from pathlib import Path  # noqa: TC003 — 运行时需要（方法签名 + open()）
+
+import requests
+from loguru import logger
+
+_TIMEOUT_S = 15
+
+
+class MonkeyOCRClient:
+    """MonkeyOCR 服务客户端：多端点轮询、单帧失败降级为跳过。
+
+    关键实现细节：实例可被多线程共享——端点轮询加锁、Session 线程局部
+    （A/B 评测会以 4 线程并发调用同一实例）。
+
+    参数:
+        urls: 服务端点列表（如 ["http://10.77.0.20:7866", ...]），非空。
+
+    异常:
+        ValueError: urls 为空时抛出。
+    """
+
+    def __init__(self, urls: list[str]) -> None:
+        if not urls:
+            raise ValueError("MonkeyOCR 端点列表不能为空")
+        self._urls = [u.rstrip("/") for u in urls]
+        self._rr = itertools.cycle(self._urls)
+        self._rr_lock = threading.Lock()
+        self._local = threading.local()
+
+    def _get_session(self) -> requests.Session:
+        """返回当前线程专属的 Session（惰性创建并复用，trust_env=False 绕代理）。"""
+        session = getattr(self._local, "session", None)
+        if session is None:
+            session = requests.Session()
+            session.trust_env = False  # LAN 直连，绕过代理
+            self._local.session = session
+        return session
+
+    def _check_health_sync(self) -> None:
+        """同步预检所有端点，任一不可达即抛错（供 asyncio.to_thread 调用）。
+
+        异常:
+            RuntimeError: 端点不可达或 /health 非 2xx。
+        """
+        for url in self._urls:
+            try:
+                resp = self._get_session().get(f"{url}/health", timeout=5)
+            except requests.RequestException as e:
+                raise RuntimeError(f"MonkeyOCR 端点不可达: {url}: {e}") from e
+            if not resp.ok:
+                raise RuntimeError(f"MonkeyOCR 健康检查失败: {url}: {resp.status_code}")
+
+    async def check_health(self) -> None:
+        """异步预检所有端点，任一不可达即抛错（A/B qtr_ocr 臂启动门）。
+
+        异常:
+            RuntimeError: 端点不可达或 /health 非 2xx。
+        """
+        await asyncio.to_thread(self._check_health_sync)
+
+    def _transcribe_frames_sync(self, frame_paths: list[Path]) -> str:
+        """同步逐帧转录并拼接（供 asyncio.to_thread 调用）。
+
+        参数:
+            frame_paths: 帧文件路径列表。
+
+        返回:
+            "帧1: <行1> | <行2>\\n帧2: ..." 格式文本；无任何有效结果时空串。
+        """
+        parts: list[str] = []
+        for i, path in enumerate(frame_paths, 1):
+            lines = self._transcribe_one(path)
+            if lines:
+                parts.append(f"帧{i}: " + " | ".join(lines))
+        return "\n".join(parts)
+
+    async def transcribe_frames(self, frame_paths: list[Path]) -> str:
+        """异步逐帧转录并拼接为注入文本；单帧失败跳过，全失败返回空串。
+
+        参数:
+            frame_paths: 帧文件路径列表。
+
+        返回:
+            "帧1: <行1> | <行2>\\n帧2: ..." 格式文本；无任何有效结果时空串。
+        """
+        return await asyncio.to_thread(self._transcribe_frames_sync, frame_paths)
+
+    def _transcribe_one(self, path: Path) -> list[str]:
+        """单帧转录：空结果/单字符行过滤 + 帧内行级去重。
+
+        参数:
+            path: 帧文件路径。
+
+        返回:
+            过滤去重后的文本行列表；请求失败或无有效行时空列表。
+        """
+        with self._rr_lock:
+            url = next(self._rr)
+        try:
+            with open(path, "rb") as f:
+                resp = self._get_session().post(
+                    f"{url}/ocr/text", files={"file": f}, timeout=_TIMEOUT_S
+                )
+            resp.raise_for_status()
+            content = resp.json().get("content", "")
+        except (requests.RequestException, ValueError) as e:
+            logger.warning("MonkeyOCR 单帧转录失败，跳过 {}: {}", path.name, e)
+            return []
+        seen: set[str] = set()
+        lines: list[str] = []
+        for ln in content.splitlines():
+            ln = ln.strip()
+            if len(ln) <= 1 or ln in seen:
+                continue
+            seen.add(ln)
+            lines.append(ln)
+        return lines
diff --git a/app/ports.py b/app/ports.py
index 147f109..2584c17 100644
--- a/app/ports.py
+++ b/app/ports.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from pathlib import Path  # noqa: TC003 — runtime_checkable Protocol 需运行时可见
 from typing import TYPE_CHECKING, Protocol, runtime_checkable
 
 if TYPE_CHECKING:
@@ -56,3 +57,20 @@ class QuestionGenerator(Protocol):
         *,
         exemplars: list[GeneratedQuestion],
     ) -> GeneratedQuestion: ...
+
+
+@runtime_checkable
+class OCRProvider(Protocol):
+    """帧文字转录端口。
+
+    实现方负责将帧图像发送给 OCR 服务并返回拼接后的文本。
+    单帧失败应降级跳过，不得抛出异常中断整体流程。
+
+    参数:
+        frame_paths: 帧文件路径列表。
+
+    返回:
+        "帧1: <行1> | <行2>\\n帧2: ..." 格式文本；无有效结果时空串。
+    """
+
+    async def transcribe_frames(self, frame_paths: list[Path]) -> str: ...
diff --git a/tests/unit/test_ocr_adapter.py b/tests/unit/test_ocr_adapter.py
new file mode 100644
index 0000000..1c641f1
--- /dev/null
+++ b/tests/unit/test_ocr_adapter.py
@@ -0,0 +1,302 @@
+"""MonkeyOCRClient 适配器单元测试。
+
+覆盖范围：Protocol 合规性、单帧转录、失败降级、健康检查、
+多端点轮询、行去重过滤。使用 responses 库 mock HTTP。
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path  # noqa: TC003 — pytest tmp_path fixture 类型注解需运行时
+
+import pytest
+import requests
+import responses
+
+from adapters.ocr import MonkeyOCRClient
+from app.ports import OCRProvider
+
+# ---------------------------------------------------------------------------
+# Protocol 合规性
+# ---------------------------------------------------------------------------
+
+
+class TestOCRProviderProtocol:
+    """MonkeyOCRClient 必须满足 OCRProvider Protocol。"""
+
+    def test_is_runtime_checkable_instance(self) -> None:
+        """MonkeyOCRClient 实例应通过 isinstance(_, OCRProvider) 检查。"""
+        client = MonkeyOCRClient(urls=["http://localhost:7866"])
+        assert isinstance(client, OCRProvider)
+
+    def test_has_transcribe_frames(self) -> None:
+        """MonkeyOCRClient 必须暴露 transcribe_frames 方法。"""
+        assert hasattr(MonkeyOCRClient, "transcribe_frames")
+
+
+# ---------------------------------------------------------------------------
+# 构造函数校验
+# ---------------------------------------------------------------------------
+
+
+class TestConstructor:
+    """构造函数参数校验。"""
+
+    def test_empty_urls_raises_value_error(self) -> None:
+        """空端点列表应抛 ValueError（P5：不用 assert）。"""
+        with pytest.raises(ValueError, match="端点列表不能为空"):
+            MonkeyOCRClient(urls=[])
+
+    def test_trailing_slash_stripped(self) -> None:
+        """URL 尾部斜杠应被去除。"""
+        client = MonkeyOCRClient(urls=["http://host:7866/"])
+        assert client._urls == ["http://host:7866"]
+
+
+# ---------------------------------------------------------------------------
+# 健康检查
+# ---------------------------------------------------------------------------
+
+
+class TestCheckHealth:
+    """check_health 预检所有端点。"""
+
+    @responses.activate
+    def test_healthy_endpoints(self) -> None:
+        """所有端点返回 200 → 无异常。"""
+        url = "http://10.0.0.1:7866"
+        responses.add(responses.GET, f"{url}/health", status=200)
+        client = MonkeyOCRClient(urls=[url])
+        asyncio.get_event_loop().run_until_complete(client.check_health())
+
+    @responses.activate
+    def test_unhealthy_endpoint_raises(self) -> None:
+        """端点返回 500 → RuntimeError。"""
+        url = "http://10.0.0.1:7866"
+        responses.add(responses.GET, f"{url}/health", status=500)
+        client = MonkeyOCRClient(urls=[url])
+        with pytest.raises(RuntimeError, match="健康检查失败"):
+            asyncio.get_event_loop().run_until_complete(client.check_health())
+
+    @responses.activate
+    def test_unreachable_endpoint_raises(self) -> None:
+        """端点连接失败 → RuntimeError。"""
+        url = "http://10.0.0.1:7866"
+        responses.add(
+            responses.GET,
+            f"{url}/health",
+            body=requests.ConnectionError("refused"),
+        )
+        client = MonkeyOCRClient(urls=[url])
+        with pytest.raises(RuntimeError, match="端点不可达"):
+            asyncio.get_event_loop().run_until_complete(client.check_health())
+
+    @responses.activate
+    def test_multiple_endpoints_all_checked(self) -> None:
+        """多端点时全部预检，任一失败即报错。"""
+        url_a = "http://10.0.0.1:7866"
+        url_b = "http://10.0.0.2:7866"
+        responses.add(responses.GET, f"{url_a}/health", status=200)
+        responses.add(responses.GET, f"{url_b}/health", status=503)
+        client = MonkeyOCRClient(urls=[url_a, url_b])
+        with pytest.raises(RuntimeError, match="健康检查失败"):
+            asyncio.get_event_loop().run_until_complete(client.check_health())
+
+
+# ---------------------------------------------------------------------------
+# 单帧转录
+# ---------------------------------------------------------------------------
+
+
+class TestTranscribeFrames:
+    """transcribe_frames 核心逻辑。"""
+
+    @responses.activate
+    def test_single_frame(self, tmp_path: Path) -> None:
+        """单帧正常转录 → '帧1: line_a | line_b' 格式。"""
+        url = "http://ocr:7866"
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": "Hello World\nOCR Test"},
+            status=200,
+        )
+        frame = tmp_path / "frame_001.jpg"
+        frame.write_bytes(b"\xff\xd8\xff\xe0fake")
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
+        assert result == "帧1: Hello World | OCR Test"
+
+    @responses.activate
+    def test_multiple_frames(self, tmp_path: Path) -> None:
+        """多帧转录 → 每帧一行，帧号递增。"""
+        url = "http://ocr:7866"
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": "Line A"},
+            status=200,
+        )
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": "Line B"},
+            status=200,
+        )
+        frames = []
+        for i in range(2):
+            f = tmp_path / f"frame_{i}.jpg"
+            f.write_bytes(b"\xff\xd8data")
+            frames.append(f)
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames(frames))
+        assert "帧1: Line A" in result
+        assert "帧2: Line B" in result
+
+    @responses.activate
+    def test_empty_frames_returns_empty(self) -> None:
+        """空帧列表 → 空串。"""
+        client = MonkeyOCRClient(urls=["http://ocr:7866"])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([]))
+        assert result == ""
+
+
+# ---------------------------------------------------------------------------
+# 失败降级
+# ---------------------------------------------------------------------------
+
+
+class TestFailureDegradation:
+    """单帧失败跳过，不影响其余帧。"""
+
+    @responses.activate
+    def test_single_frame_failure_returns_empty(self, tmp_path: Path) -> None:
+        """唯一帧请求失败 → 返回空串（不抛异常）。"""
+        url = "http://ocr:7866"
+        responses.add(responses.POST, f"{url}/ocr/text", status=500)
+        frame = tmp_path / "frame.jpg"
+        frame.write_bytes(b"\xff\xd8data")
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
+        assert result == ""
+
+    @responses.activate
+    def test_partial_failure_skips_bad_frame(self, tmp_path: Path) -> None:
+        """第一帧失败、第二帧成功 → 仅输出第二帧。"""
+        url = "http://ocr:7866"
+        responses.add(responses.POST, f"{url}/ocr/text", status=500)
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": "Good"},
+            status=200,
+        )
+        frames = []
+        for i in range(2):
+            f = tmp_path / f"frame_{i}.jpg"
+            f.write_bytes(b"\xff\xd8data")
+            frames.append(f)
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames(frames))
+        # 帧1 失败被跳过，帧2 成功但输出为 "帧2: Good"
+        assert "帧1" not in result
+        assert "帧2: Good" in result
+
+
+# ---------------------------------------------------------------------------
+# 行去重过滤
+# ---------------------------------------------------------------------------
+
+
+class TestLineDedup:
+    """帧内行级去重与短行过滤。"""
+
+    @responses.activate
+    def test_duplicate_lines_removed(self, tmp_path: Path) -> None:
+        """帧内重复行只保留首次出现。"""
+        url = "http://ocr:7866"
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": "重复行\n重复行\n不同行"},
+            status=200,
+        )
+        frame = tmp_path / "frame.jpg"
+        frame.write_bytes(b"\xff\xd8data")
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
+        assert result == "帧1: 重复行 | 不同行"
+
+    @responses.activate
+    def test_single_char_lines_filtered(self, tmp_path: Path) -> None:
+        """单字符行被过滤（长度 <= 1）。"""
+        url = "http://ocr:7866"
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": "A\nAB\n.\nCD"},
+            status=200,
+        )
+        frame = tmp_path / "frame.jpg"
+        frame.write_bytes(b"\xff\xd8data")
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
+        # "A" 和 "." 被过滤（长度 <= 1），保留 "AB" 和 "CD"
+        assert result == "帧1: AB | CD"
+
+    @responses.activate
+    def test_empty_content_skipped(self, tmp_path: Path) -> None:
+        """OCR 返回空 content → 该帧跳过。"""
+        url = "http://ocr:7866"
+        responses.add(
+            responses.POST,
+            f"{url}/ocr/text",
+            json={"content": ""},
+            status=200,
+        )
+        frame = tmp_path / "frame.jpg"
+        frame.write_bytes(b"\xff\xd8data")
+        client = MonkeyOCRClient(urls=[url])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
+        assert result == ""
+
+
+# ---------------------------------------------------------------------------
+# 多端点轮询
+# ---------------------------------------------------------------------------
+
+
+class TestRoundRobin:
+    """多端点轮询行为。"""
+
+    @responses.activate
+    def test_round_robin_alternation(self, tmp_path: Path) -> None:
+        """两端点交替使用。"""
+        url_a = "http://host-a:7866"
+        url_b = "http://host-b:7866"
+        # 为两个端点各注册响应
+        responses.add(
+            responses.POST,
+            f"{url_a}/ocr/text",
+            json={"content": "From A"},
+            status=200,
+        )
+        responses.add(
+            responses.POST,
+            f"{url_b}/ocr/text",
+            json={"content": "From B"},
+            status=200,
+        )
+        frames = []
+        for i in range(2):
+            f = tmp_path / f"frame_{i}.jpg"
+            f.write_bytes(b"\xff\xd8data")
+            frames.append(f)
+        client = MonkeyOCRClient(urls=[url_a, url_b])
+        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames(frames))
+        assert "帧1: From A" in result
+        assert "帧2: From B" in result
+        # 验证两个端点都被调用
+        called_urls = [c.request.url for c in responses.calls]
+        assert any(url_a in u for u in called_urls)
+        assert any(url_b in u for u in called_urls)