diff --git a/paperforge/skills/paperforge/scripts/pf_deep.py b/paperforge/skills/paperforge/scripts/pf_deep.py index 99c2d0f3..a775eab8 100644 --- a/paperforge/skills/paperforge/scripts/pf_deep.py +++ b/paperforge/skills/paperforge/scripts/pf_deep.py @@ -1439,7 +1439,7 @@ def prepare_deep_reading(vault: Path, zotero_key: str, force: bool = False) -> d meta = _read_json(meta_path) ocr_status = str(meta.get("ocr_status", "pending")).strip().lower() - if ocr_status != "done": + if ocr_status not in ("done", "done_degraded"): result["message"] = f"[ERROR] OCR status='{ocr_status}', not 'done'. Wait for OCR or check meta.json." return result @@ -1531,7 +1531,7 @@ def scan_deep_reading_queue(vault: Path) -> list[dict]: # Sort: OCR completed first, then by domain, then by key queue.sort( key=lambda row: ( - 0 if row["ocr_status"] == "done" else 1, + 0 if row["ocr_status"] in ("done", "done_degraded") else 1, row["domain"], row["zotero_key"], ) @@ -1737,8 +1737,8 @@ def main() -> int: serializable.append(r) print(json.dumps(serializable, ensure_ascii=False, indent=2)) else: - ready = [row for row in queue if row["ocr_status"] == "done"] - blocked = [row for row in queue if row["ocr_status"] != "done"] + ready = [row for row in queue if row["ocr_status"] in ("done", "done_degraded")] + blocked = [row for row in queue if row["ocr_status"] not in ("done", "done_degraded")] print(f"# 待精读队列 ({len(queue)} 篇)") print() if ready: diff --git a/paperforge/worker/ocr.py b/paperforge/worker/ocr.py index 0d83c0c3..71155d18 100644 --- a/paperforge/worker/ocr.py +++ b/paperforge/worker/ocr.py @@ -59,6 +59,14 @@ from paperforge.worker.ocr_blocks import ( write_raw_blocks_jsonl, write_structured_blocks_jsonl, ) +from paperforge.worker.ocr_errors import ( + OCRAPISchemaError, + OCRArtifactIntegrityError, + OCRNetworkError, + OCRPDFResolveError, + OCRPostprocessError, + classify_ocr_error, +) from paperforge.worker.ocr_figures import ( build_figure_inventory, write_figure_inventory, @@ -79,6 +87,19 @@ from paperforge.worker.sync import ( logger = logging.getLogger(__name__) +OCR_QUEUE_STATUSES = { + "pending", "queued", "running", "done", "done_degraded", + "retryable_error", "fatal_error", "nopdf", "blocked", +} + + +def apply_ocr_error_state(row: dict, meta: dict, error_state: dict) -> None: + row["queue_status"] = error_state["status"] + meta["error_type"] = error_state["error_type"] + meta["error_stage"] = error_state["error_stage"] + meta["retryable"] = error_state["retryable"] + meta["last_error"] = error_state["last_error"] + def ensure_ocr_meta(vault: Path, row: dict) -> dict: paths = pipeline_paths(vault) @@ -2184,11 +2205,8 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec try: response = retry_with_meta(_do_poll, meta_path_poll, meta["ocr_job_id"], token) except Exception as e: - meta["ocr_status"] = "pending" - meta["error"] = str(e) - meta["last_error"] = str(e) - meta["retry_count"] = int(meta.get("retry_count", 0)) + 1 - queue_row["queue_status"] = "pending" + error_state = classify_ocr_error(OCRNetworkError(str(e)), stage="poll") + apply_ocr_error_state(queue_row, meta, error_state) write_json(paths["ocr"] / key / "meta.json", meta) changed += 1 active_submitted = max(0, active_submitted - 1) @@ -2197,13 +2215,17 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec payload = response.json()["data"] state = payload["state"] except (json.JSONDecodeError, KeyError) as e: - meta["ocr_status"] = "pending" - meta["error"] = f"API schema mismatch during polling: {e}" - meta["last_error"] = meta["error"] + error_state = classify_ocr_error( + OCRAPISchemaError(f"API schema mismatch during polling: {e}"), stage="poll" + ) + apply_ocr_error_state(queue_row, meta, error_state) meta["raw_response"] = response.text[:1000] - meta["retry_count"] = int(meta.get("retry_count", 0)) + 1 - queue_row["queue_status"] = "pending" write_json(paths["ocr"] / key / "meta.json", meta) + json_dir = paths["ocr"] / key / "json" + json_dir.mkdir(parents=True, exist_ok=True) + raw_error_dir = paths["ocr"] / key / "raw" + raw_error_dir.mkdir(parents=True, exist_ok=True) + write_json(raw_error_dir / "api_error_payload.json", {"raw_response": response.text[:5000]}) changed += 1 active_submitted = max(0, active_submitted - 1) continue @@ -2224,16 +2246,33 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec page_num, markdown_path, json_path, fulltext_md_path = postprocess_ocr_result( vault, key, all_results ) - except Exception as e: - meta["ocr_status"] = "pending" - meta["error"] = str(e) - meta["retry_count"] = int(meta.get("retry_count", 0)) + 1 - queue_row["queue_status"] = "pending" + except KeyboardInterrupt: + raise + except SystemExit: + raise + except BaseException: + raise + except (OCRPDFResolveError, OCRArtifactIntegrityError, OCRPostprocessError) as exc: + error_state = classify_ocr_error(exc, stage="postprocess") + apply_ocr_error_state(queue_row, meta, error_state) write_json(paths["ocr"] / key / "meta.json", meta) changed += 1 active_submitted = max(0, active_submitted - 1) continue - meta["ocr_status"] = "done" + except Exception as exc: + error_state = classify_ocr_error(OCRPostprocessError(str(exc)), stage="postprocess") + apply_ocr_error_state(queue_row, meta, error_state) + write_json(paths["ocr"] / key / "meta.json", meta) + changed += 1 + active_submitted = max(0, active_submitted - 1) + continue + health_path_poll = paths["ocr"] / key / "health" / "ocr_health.json" + health_report_poll = read_json(health_path_poll) if health_path_poll.exists() else {} + if health_report_poll.get("degraded_reasons"): + meta["ocr_status"] = "done_degraded" + meta["degraded_reason"] = "; ".join(health_report_poll["degraded_reasons"]) + else: + meta["ocr_status"] = "done" # Per D-01: auto_analyze_after_ocr opt-in workflow streamlining cfg_path = vault / "paperforge.json" if cfg_path.exists(): @@ -2435,7 +2474,13 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec page_num, md_path, json_path, fulltext_md_path = postprocess_ocr_result(vault, key, results) except Exception: continue - meta["ocr_status"] = "done" + health_path_loop = paths["ocr"] / key / "health" / "ocr_health.json" + health_report_loop = read_json(health_path_loop) if health_path_loop.exists() else {} + if health_report_loop.get("degraded_reasons"): + meta["ocr_status"] = "done_degraded" + meta["degraded_reason"] = "; ".join(health_report_loop["degraded_reasons"]) + else: + meta["ocr_status"] = "done" meta["ocr_finished_at"] = datetime.now(timezone.utc).isoformat() meta["page_count"] = page_num meta["markdown_path"] = md_path diff --git a/paperforge/worker/ocr_errors.py b/paperforge/worker/ocr_errors.py new file mode 100644 index 00000000..ae8dbdb2 --- /dev/null +++ b/paperforge/worker/ocr_errors.py @@ -0,0 +1,48 @@ +from __future__ import annotations + + +class OCRRecoverableError(Exception): + pass + + +class OCRFatalError(Exception): + pass + + +class OCRNetworkError(OCRRecoverableError): + pass + + +class OCRAPISchemaError(OCRRecoverableError): + pass + + +class OCRPDFResolveError(OCRFatalError): + pass + + +class OCRPostprocessError(OCRFatalError): + pass + + +class OCRArtifactIntegrityError(OCRFatalError): + pass + + +def classify_ocr_error(exc: Exception, *, stage: str) -> dict: + retryable = isinstance(exc, OCRRecoverableError) + return { + "status": "retryable_error" if retryable else "fatal_error", + "error_type": exc.__class__.__name__, + "error_stage": stage, + "retryable": retryable, + "last_error": str(exc), + } + + +def normalize_ocr_status_for_reader(status: str) -> str: + if status in {"retryable_error", "fatal_error"}: + return "error" + if status == "done_degraded": + return "done" + return status diff --git a/paperforge/worker/ocr_health.py b/paperforge/worker/ocr_health.py index 74ba41e8..6de61b82 100644 --- a/paperforge/worker/ocr_health.py +++ b/paperforge/worker/ocr_health.py @@ -107,6 +107,16 @@ def build_ocr_health( } report.update(decision_summary) + degraded_reasons = [] + if span.get("coverage_quality", "weak") == "weak": + degraded_reasons.append(f"weak span coverage ({span.get('coverage_ratio', 0):.0%})") + if spine.get("_meta", {}).get("quality", "weak") == "weak": + degraded_reasons.append("weak body spine") + if layout.get("status", "unknown") == "fail": + degraded_reasons.append(f"layout audit failed ({layout.get('anomaly_count', 0)} anomalies)") + + report["degraded_reasons"] = degraded_reasons + def _score_distribution(scores: list[float]) -> dict: return { "high": sum(1 for s in scores if s >= 0.75), diff --git a/paperforge/worker/status.py b/paperforge/worker/status.py index 23b67bb8..2f1d32fb 100644 --- a/paperforge/worker/status.py +++ b/paperforge/worker/status.py @@ -782,7 +782,7 @@ def run_doctor(vault: Path, verbose: bool = False, json_output: bool = False) -> if not _ws_dir.exists(): _missing_workspace += 1 continue - if _e.get("ocr_status") == "done": + if _e.get("ocr_status") in ("done", "done_degraded"): _ft = _ws_dir / "fulltext.md" if not _ft.exists(): _missing_fulltext += 1 @@ -1018,7 +1018,7 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) -> ocr_failed += 1 continue status = str(meta.get("ocr_status", "")).strip().lower() - if status == "done": + if status in ("done", "done_degraded"): ocr_done += 1 elif status in ("queued", "running", "processing"): ocr_processing += 1 @@ -1038,7 +1038,7 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) -> continue ocr_total += 1 status = str(meta.get("ocr_status", "")).strip().lower() - if status == "done": + if status in ("done", "done_degraded"): ocr_done += 1 elif status in ("queued", "running", "processing"): ocr_processing += 1 diff --git a/tests/test_ocr_errors.py b/tests/test_ocr_errors.py new file mode 100644 index 00000000..d0430fff --- /dev/null +++ b/tests/test_ocr_errors.py @@ -0,0 +1,54 @@ +from __future__ import annotations + + +def test_network_error_maps_to_retryable_status() -> None: + from paperforge.worker.ocr_errors import OCRNetworkError, classify_ocr_error + + result = classify_ocr_error(OCRNetworkError("timeout"), stage="poll") + assert result["status"] == "retryable_error" + assert result["retryable"] is True + assert result["error_type"] == "OCRNetworkError" + + +def test_postprocess_error_maps_to_fatal_status() -> None: + from paperforge.worker.ocr_errors import OCRPostprocessError, classify_ocr_error + + result = classify_ocr_error(OCRPostprocessError("bad role graph"), stage="postprocess") + assert result["status"] == "fatal_error" + assert result["retryable"] is False + assert result["error_stage"] == "postprocess" + + +def test_ocr_queue_statuses_include_error_taxonomy() -> None: + from paperforge.worker.ocr import OCR_QUEUE_STATUSES + + for status in ["pending", "queued", "running", "done", "done_degraded", "retryable_error", "fatal_error", "nopdf", "blocked"]: + assert status in OCR_QUEUE_STATUSES + + +def test_apply_ocr_error_state_updates_queue_and_meta() -> None: + from paperforge.worker.ocr import apply_ocr_error_state + + row = {"queue_status": "running"} + meta = {} + apply_ocr_error_state(row, meta, {"status": "fatal_error", "error_type": "OCRPostprocessError", "error_stage": "postprocess", "retryable": False, "last_error": "bad"}) + assert row["queue_status"] == "fatal_error" + assert meta["error_type"] == "OCRPostprocessError" + + +def test_legacy_error_status_maps_to_new_taxonomy_for_readers() -> None: + from paperforge.worker.ocr_errors import normalize_ocr_status_for_reader + + assert normalize_ocr_status_for_reader("retryable_error") == "error" + assert normalize_ocr_status_for_reader("fatal_error") == "error" + assert normalize_ocr_status_for_reader("done_degraded") == "done" + + +def test_postprocess_failure_does_not_return_to_pending() -> None: + from paperforge.worker.ocr_errors import OCRPostprocessError, classify_ocr_error + + state = classify_ocr_error(OCRPostprocessError("bad structure"), stage="postprocess") + assert state["status"] == "fatal_error" + assert state["retryable"] is False + + diff --git a/tests/test_ocr_health.py b/tests/test_ocr_health.py index 47b4941c..a02aacb0 100644 --- a/tests/test_ocr_health.py +++ b/tests/test_ocr_health.py @@ -238,3 +238,18 @@ def test_ocr_health_includes_confidence_distributions() -> None: assert tbl_dist["high"] == 1 assert tbl_dist["medium"] == 1 assert tbl_dist["low"] == 1 + + +def test_health_report_includes_degraded_reasons() -> None: + from paperforge.worker.ocr_health import build_ocr_health + + report = build_ocr_health( + page_count=1, + raw_blocks_count=0, + structured_blocks=[], + figure_inventory={}, + table_inventory={}, + ) + + assert "degraded_reasons" in report + assert len(report["degraded_reasons"]) > 0