feat(engine): fps 强制自测（忽略外部传入，从 decoder 元数据读）+ extract() 通用文本提取入口（ExtractionResult/ExtractedSegment dataclass）；建仓评估脚本（引擎依赖 7 模块零应用泄漏，ocr_engine.py 不随引擎）

Author: chr431

tools/archive/_engine_api_report.py

@@ -0,0 +1,43 @@
+"""列出引擎 FieldExtractor 方法、__init__ 参数、extract 状态。"""
+import ast
+from pathlib import Path
+
+root = Path(__file__).resolve().parent.parent.parent
+src = (root / "video_ocr_engine" / "extractor.py").read_text(encoding="utf-8")
+t = ast.parse(src)
+cls = next(n for n in t.body if isinstance(n, ast.ClassDef)
+           and n.name == "FieldExtractor")
+methods = [n for n in cls.body if isinstance(n, ast.FunctionDef)]
+
+print("=== FieldExtractor 方法 ===")
+for m in methods:
+    decs = " ".join(ast.unparse(d) for d in m.decorator_list)
+    print(f"  {decs or '-':28} def {m.name}")
+    doc = ast.get_docstring(m)
+    if doc:
+        first = doc.strip().splitlines()[0][:70]
+        print(f"      {first}")
+
+init = next(m for m in methods if m.name == "__init__")
+pos = [a.arg for a in init.args.args if a.arg != "self"]
+kw = [a.arg for a in init.args.kwonlyargs]
+print("\n=== __init__ 参数 ===")
+print("  位置参数:", ", ".join(pos))
+print("  仅关键字:", ", ".join(kw))
+
+print("\n=== extract() 是否已实现 ===")
+ext = next((m for m in methods if m.name == "extract"), None)
+print("  存在:", ext is not None)
+if ext:
+    body = ast.unparse(ext.body[0]) if ext.body else "(空)"
+    print("  首个语句:", body)
+
+# SegmentPipeline 怎么调它
+print("\n=== SegmentPipeline.__init__ 调 super() 的参数 ===")
+sf_src = (root / "segment_flow.py").read_text(encoding="utf-8")
+if "super().__init__(" in sf_src:
+    i = sf_src.index("super().__init__(")
+    j = sf_src.index(")", i)
+    print(sf_src[i:j + 1])
+else:
+    print("  未发现 super().__init__ 调用")

tools/archive/_engine_repo_assess.py

@@ -0,0 +1,41 @@
+"""建独立仓库评估：引擎依赖的 8 个模块各自的 import（找应用层泄漏）。
+
+目标：确认 video_ocr_engine 独立仓库需要哪些文件、哪些模块需净化。
+"""
+import ast
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+CANDIDATES = ["engine_config", "segmentation", "hybrid_decode", "ocr_native",
+              "ocr_trt", "ocr_engine", "video_utils", "constants"]
+# 应用层（不随引擎走）
+APP_MODULES = {"config", "segment_flow", "seg_correction", "ocr_text",
+               "csv_io", "gui", "signals", "monitor", "headless",
+               "export_controller", "analysis"}
+
+report = {}
+for name in CANDIDATES:
+    p = ROOT / f"{name}.py"
+    if not p.exists():
+        report[name] = {"error": "missing"}
+        continue
+    tree = ast.parse(p.read_text(encoding="utf-8"))
+    imports = []
+    for n in ast.walk(tree):
+        if isinstance(n, ast.ImportFrom) and n.module:
+            imports.append(n.module.split(".")[0])
+        elif isinstance(n, ast.Import):
+            imports.append("")
+    leaks = sorted(m for m in imports if m in APP_MODULES)
+    report[name] = {
+        "top_imports": sorted(m for m in imports if m),
+        "app_leaks": leaks,
+        "size_kb": round(p.stat().st_size / 1024, 1),
+    }
+
+for name, r in report.items():
+    print(f"\n=== {name} ({r.get('size_kb')}KB) ===")
+    print(f"  顶层 import: {r.get('top_imports')}")
+    leaks = r.get("app_leaks")
+    print(f"  应用泄漏: {leaks if leaks else '无'}")

video_ocr_engine/__init__.py

@@ -11,7 +11,9 @@
     for seg in result.segments:
         print(seg.text, seg.confidence)
 """
-from video_ocr_engine.extractor import FieldExtractor  # noqa: F401
+from video_ocr_engine.extractor import (  # noqa: F401
+    FieldExtractor, ExtractedSegment, ExtractionResult,
+)
 from video_ocr_engine import _version  # noqa: F401
 
 __version__ = _version.__version__

video_ocr_engine/extractor.py

@@ -11,7 +11,9 @@
 import os as _os
 import threading
 import time
+from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Any, Optional
 
 import numpy as np
 
@@ -37,6 +39,30 @@ def _ocr_batch_size() -> int:
     return config.OCR_BATCH_SIZE
 
 
+@dataclass
+class ExtractedSegment:
+    """引擎输出的单个文本字段段（原有字段区间 + 代表帧 + 原始文本）。"""
+
+    start: int                      # 段首帧号
+    end: int                        # 段末帧号
+    frames: tuple = ()              # 段内帧号序列
+    rep_frame: int = -1             # 代表帧号（段内最清晰帧）
+    text: Optional[str] = None      # OCR 原始文本（None=未读出）
+    confidence: float = 0.0         # OCR 置信度 0-1
+    rep_crop: Any = None            # 代表帧 ROI 图像（YUV420 或 RGB）
+
+
+@dataclass
+class ExtractionResult:
+    """引擎通用提取结果（无领域语义）。"""
+
+    segments: list = field(default_factory=list)  # list[ExtractedSegment]
+    frames: list = field(default_factory=list)     # 全部采样帧号
+    fps: float = 0.0                               # 自测帧率
+    timing: dict = field(default_factory=dict)     # 各阶段耗时
+    meta: dict = field(default_factory=dict)       # backend/codec/引擎版本等
+
+
 class FieldExtractor:
     """从视频固定区域提取文本的通用引擎（识别链：解码∥分段∥OCR）。
 
@@ -55,7 +81,10 @@ def __init__(self, video_path: str, roi: tuple, *, frame_start=None,
                  yuv_output: bool = False):
         self._video_path = Path(video_path)
         self._roi = tuple(roi)
-        self._fps = fps  # 外部给定时直接用（truth 头），否则识别链推导
+        # fps 强制自测：open decoder 后从 get_avg_fps/get_fps 读，忽略外部
+        # 传入（truth 头的 fps 可能与视频实际帧率偏离；自测无额外解码开销，
+        # 只在打开时读一次元数据）。fps 参数保留仅为 API 兼容（已废弃）。
+        self._fps = None
         self._frame_start = frame_start or 0
         self._frame_end = frame_end
         self._force_aspect = force_aspect
@@ -100,8 +129,31 @@ def __init__(self, video_path: str, roi: tuple, *, frame_start=None,
         # 后处理参数由子类（SegmentPipeline）在构造时设置；引擎识别链不读。
 
     def extract(self):
-        """通用文本提取入口（待精修：解码∥分段∥OCR → 每段 text/conf 结果）。"""
-        raise NotImplementedError
+        """通用文本提取：解码∥分段∥OCR → 结构化结果（每段原始文本+置信度）。
+
+        引擎的正式通用入口（无任何领域语义）。返回 ExtractionResult：
+          - segments: list[ExtractedSegment]（start/end/rep_frame/text/confidence/
+            rep_crop）
+          - frames / fps / timing / meta
+        识别层不解析文本含义（速度/数值由上层应用处理）。ffis 强制自测。
+        """
+        frames, segs, texts, confs, rep_frames = self._run_pipelined()
+        segments = [
+            ExtractedSegment(
+                start=seg[0], end=seg[-1], frames=tuple(seg),
+                rep_frame=rep_frames[i],
+                text=texts[i] if i < len(texts) else None,
+                confidence=confs[i] if i < len(confs) else 0.0,
+                rep_crop=self.crops.get(rep_frames[i]))
+            for i, seg in enumerate(segs)
+        ]
+        return ExtractionResult(
+            segments=segments, frames=frames, fps=self._fps or 0.0,
+            timing=dict(self.timing),
+            meta={"backend": self._backend,
+                  "ocr_backend": self._ocr_backend_used,
+                  "codec": self._codec,
+                  "n_segments": len(segments)})
 
     @property
     def frames(self) -> list: