feat(adapters): OCRProvider Protocol + MonkeyOCRClient 异步适配器

- 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>
This commit is contained in:
2026-07-07 05:47:55 -04:00
parent 60e737e2fc
commit 86b19d1e07
3 changed files with 448 additions and 0 deletions
+128
View File
@@ -0,0 +1,128 @@
"""MonkeyOCR HTTP 客户端 — 帧文字转录的异构硬证据源。
服务由用户在 LAN 部署(双端点轮询);请求必须绕过代理(trust_env=False)。
实现 OCRProvider Protocolapp/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
+18
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from pathlib import Path # noqa: TC003 — runtime_checkable Protocol 需运行时可见
from typing import TYPE_CHECKING, Protocol, runtime_checkable from typing import TYPE_CHECKING, Protocol, runtime_checkable
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -56,3 +57,20 @@ class QuestionGenerator(Protocol):
*, *,
exemplars: list[GeneratedQuestion], exemplars: list[GeneratedQuestion],
) -> 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: ...
+302
View File
@@ -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:
"""空端点列表应抛 ValueErrorP5:不用 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)