iomgaa
86b19d1e07
- 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>
77 lines
1.8 KiB
Python
77 lines
1.8 KiB
Python
"""应用层 Protocol 端口定义。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path # noqa: TC003 — runtime_checkable Protocol 需运行时可见
|
|
from typing import TYPE_CHECKING, Protocol, runtime_checkable
|
|
|
|
if TYPE_CHECKING:
|
|
import numpy as np
|
|
|
|
from app.tree.index import TreeIndex
|
|
from core.types import GeneratedQuestion
|
|
|
|
|
|
@runtime_checkable
|
|
class EmbeddingProvider(Protocol):
|
|
"""文本嵌入端口。
|
|
|
|
属性:
|
|
dim: 嵌入维度 D。
|
|
"""
|
|
|
|
@property
|
|
def dim(self) -> int: ...
|
|
|
|
def embed(self, texts: str | list[str]) -> np.ndarray:
|
|
"""文本 → 嵌入向量（L2 归一化）。
|
|
|
|
参数:
|
|
texts: 单条文本或文本列表。
|
|
|
|
返回:
|
|
[N, D] ndarray，每行 L2 范数为 1.0。
|
|
"""
|
|
...
|
|
|
|
|
|
@runtime_checkable
|
|
class QuestionGenerator(Protocol):
|
|
"""LLM 驱动的题目生成端口（预留接口）。
|
|
|
|
参数:
|
|
video_id: 视频标识。
|
|
task_type: 题型。
|
|
tree: 视频树索引，提供锚节点上下文。
|
|
exemplars: 风格示例题目列表。
|
|
|
|
返回:
|
|
生成的单条题目。
|
|
"""
|
|
|
|
async def generate(
|
|
self,
|
|
video_id: str,
|
|
task_type: str,
|
|
tree: TreeIndex,
|
|
*,
|
|
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: ...
|