obsidian-plugin/lllin000_PaperForge
1
0
Fork 0
mirror of https://github.com/lllin000/PaperForge.git synced 2026-07-22 06:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(25-01): refactor run_status to read canonical index + add doctor Index Health

Browse source 
- run_status() reads canonical index via summarize_index() for lifecycle,
  health, and maturity aggregates; falls back to filesystem when index missing
- JSON output includes lifecycle_level_counts, health_aggregate,
  maturity_distribution (or None when falling back)
- Text output shows lifecycle and health lines when index is present
- run_doctor() shows Index Health section with PDF/OCR/Note/Asset Health
  counts and status per dimension
- Brownfield detection: legacy schema, old Base templates, partial OCR assets
- Fixed status_tag mapping to support 'info' status (existing bug)
- 6 new tests covering index-backed JSON output, fallback, text output,
  Index Health with/without index, and mixed health counts

Phase 25-01, Tasks 2+3
This commit is contained in:
Research Assistant 2026-05-04 11:26:39 +08:00
parent d79275aadc
commit e2ef4e9f02
2 changed files with 371 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

123
paperforge/worker/status.py
View file

@ -395,6 +395,94 @@ def run_doctor(vault: Path, verbose: bool = False) -> int:
else:
add_check("Agent 脚本", "warn", "literature-qa skill 目录未找到", "确认 agent_config_dir 配置正确")
# --- Index Health section (Phase 25: derived from canonical index) ---
try:
from paperforge.worker.asset_index import read_index as _read_idx, summarize_index as _summarize_idx
_summary = _summarize_idx(vault)
except Exception:
_summary = None
if _summary and _summary.get("paper_count", 0) > 0:
_health = _summary.get("health_aggregate", {})
def _health_status(dim: dict) -> tuple[str, str]:
healthy = dim.get("healthy", 0)
unhealthy = dim.get("unhealthy", 0)
total = healthy + unhealthy
if total == 0:
return ("pass", "0 papers")
if unhealthy == 0:
return ("pass", f"{healthy}/{total} healthy")
if healthy == 0:
return ("fail", f"0/{total} healthy -- run doctor/repair")
return ("warn", f"{healthy}/{total} healthy, {unhealthy} issues")
for _dim_name, _dim_key in [
("PDF Health", "pdf_health"),
("OCR Health", "ocr_health"),
("Note Health", "note_health"),
("Asset Health", "asset_health"),
]:
_dim = _health.get(_dim_key, {})
if not isinstance(_dim, dict):
_dim = {"healthy": 0, "unhealthy": 0}
_st, _msg = _health_status(_dim)
add_check("Index Health", _st, f"{_dim_name}: {_msg}")
# Brownfield migration detection (MIG-01, MIG-03)
_idx_data = _read_idx(vault)
if isinstance(_idx_data, dict):
_sv = _idx_data.get("schema_version", "0")
try:
if int(_sv) < 2:
add_check(
"Index Health", "warn",
f"Legacy index schema v{_sv} -- consider rebuild to v2",
"Run `paperforge sync --rebuild-index`",
)
except (ValueError, TypeError):
pass
# Legacy Base templates (pre-lifecycle format)
_bases_dir = paths.get("bases")
if _bases_dir and _bases_dir.exists():
_legacy = 0
for _bp in _bases_dir.glob("*.base"):
try:
_content = _bp.read_text(encoding="utf-8")
if "has_pdf" in _content and "lifecycle" not in _content:
_legacy += 1
except Exception:
continue
if _legacy > 0:
add_check(
"Index Health", "warn",
f"{_legacy} Base file(s) use legacy columns (has_pdf, do_ocr) instead of lifecycle",
"Run `paperforge sync` to regenerate Base views",
)
# Partial OCR assets
_ocr_dir = paths.get("ocr")
if _ocr_dir and _ocr_dir.exists():
_partial = 0
for _mp in _ocr_dir.glob("*/meta.json"):
try:
_meta = json.loads(_mp.read_text(encoding="utf-8"))
_os = str(_meta.get("ocr_status", "")).strip().lower()
if _os == "done_incomplete":
_partial += 1
except Exception:
continue
if _partial > 0:
add_check(
"Index Health", "warn",
f"{_partial} partial OCR asset(s) found",
"Re-run `paperforge ocr` on affected items",
)
else:
add_check("Index Health", "info", "No canonical index -- run `paperforge sync` to generate")
print("PaperForge Doctor")
print("=" * 40)
current_category = ""
@ -404,7 +492,7 @@ def run_doctor(vault: Path, verbose: bool = False) -> int:
if current_category:
print()
current_category = category
status_tag = {"pass": "[PASS]", "fail": "[FAIL]", "warn": "[WARN]"}[status]
status_tag = {"pass": "[PASS]", "fail": "[FAIL]", "warn": "[WARN]", "info": "[INFO]"}.get(status, "[INFO]")
print(f"{status_tag} {category} — {message}")
if status == "fail" and fix:
fix_map.setdefault(category, [])
@ -447,6 +535,14 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
paths = pipeline_paths(vault)
cfg = load_vault_config(vault)
config = load_domain_config(paths)
# Phase 25: read canonical index summary (falls back gracefully)
try:
from paperforge.worker.asset_index import summarize_index
summary = summarize_index(vault)
except Exception:
summary = None
ensure_base_views(vault, paths, config)
export_files = sorted(paths["exports"].glob("*.json"))
record_count = sum(1 for _ in paths["library_records"].rglob("*.md")) if paths["library_records"].exists() else 0
@ -524,6 +620,15 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
except Exception:
continue
# Phase 25: index-derived aggregates (None when index unavailable)
lifecycle_level_counts = None
health_aggregate = None
maturity_distribution = None
if summary is not None:
lifecycle_level_counts = summary["lifecycle_level_counts"]
health_aggregate = summary["health_aggregate"]
maturity_distribution = summary["maturity_distribution"]
if json_output:
data = {
"version": __import__("paperforge").__version__,
@ -544,6 +649,10 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
},
"path_errors": path_error_count,
"env_configured": len(env_found) > 0,
# Phase 25: lifecycle/health/maturity from canonical index (None when unavailable)
"lifecycle_level_counts": lifecycle_level_counts,
"health_aggregate": health_aggregate,
"maturity_distribution": maturity_distribution,
}
print(_json.dumps(data, indent=2, ensure_ascii=False))
return 0
@ -559,6 +668,18 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
print(f"- library_records: {record_count}")
print(f"- formal_notes: {note_count}")
print(f"- bases: {base_count}")
# Phase 25: index section (only when canonical index available)
if summary is not None:
lc = lifecycle_level_counts
print(f"- index: {summary['paper_count']} papers")
print(f" lifecycle: indexed={lc['indexed']} pdf_ready={lc['pdf_ready']} "
f"fulltext_ready={lc['fulltext_ready']} deep_read={lc['deep_read_done']} "
f"ai_ready={lc['ai_context_ready']}")
ha = health_aggregate
print(f" health: pdf={ha['pdf_health']['healthy']}/{ha['pdf_health']['unhealthy']} "
f"ocr={ha['ocr_health']['healthy']}/{ha['ocr_health']['unhealthy']} "
f"note={ha['note_health']['healthy']}/{ha['note_health']['unhealthy']} "
f"asset={ha['asset_health']['healthy']}/{ha['asset_health']['unhealthy']}")
print(f"- ocr: {ocr_done}/{ocr_total} done (pending: {ocr_pending}, processing: {ocr_processing}, failed: {ocr_failed})")
print(f"- path_errors: {path_error_count}")
if path_error_count > 0:

249
tests/test_status.py Normal file
View file

@ -0,0 +1,249 @@
"""Tests for status.py — run_status with canonical index, run_doctor Index Health.
Phase 25-01 Task 2: run_status() reads from canonical index
Phase 25-01 Task 3: run_doctor() Index Health section
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _minimal_vault(tmp_path: Path) -> Path:
"""Create a minimal vault with paperforge.json for path resolution."""
vault = tmp_path / "test_vault"
vault.mkdir(parents=True, exist_ok=True)
pf_json = vault / "paperforge.json"
pf_json.write_text(
json.dumps(
{
"version": "1.2.0",
"vault_config": {
"system_dir": "99_System",
"resources_dir": "03_Resources",
"literature_dir": "Literature",
"control_dir": "LiteratureControl",
"base_dir": "05_Bases",
"skill_dir": ".opencode/skills",
},
},
indent=2,
),
encoding="utf-8",
)
return vault
def _ensure_domain_config(vault: Path) -> None:
"""Create domain config so load_domain_config returns a valid config."""
from paperforge.config import paperforge_paths as _pp
paths = _pp(vault)
config_dir = paths["paperforge"] / "config"
config_dir.mkdir(parents=True, exist_ok=True)
config_path = config_dir / "domain-collections.json"
config_path.write_text(
json.dumps(
{
"collections": {},
"domains": [],
},
indent=2,
),
encoding="utf-8",
)
def _write_index(vault: Path, items: list[dict]) -> None:
"""Write a canonical index envelope to the vault."""
from paperforge.worker.asset_index import atomic_write_index, build_envelope, get_index_path
idx_path = get_index_path(vault)
idx_path.parent.mkdir(parents=True, exist_ok=True)
envelope = build_envelope(items)
atomic_write_index(idx_path, envelope)
def _ensure_library_records(vault: Path) -> Path:
"""Ensure library_records dir exists and return its path."""
from paperforge.worker._utils import pipeline_paths
paths = pipeline_paths(vault)
records_dir = paths["library_records"]
records_dir.mkdir(parents=True, exist_ok=True)
return records_dir
def _ensure_exports(vault: Path) -> None:
"""Create empty exports dir (needed for run_status filesystem counts)."""
from paperforge.worker._utils import pipeline_paths
paths = pipeline_paths(vault)
paths["exports"].mkdir(parents=True, exist_ok=True)
# ---------------------------------------------------------------------------
# Task 2: status --json reads from canonical index
# ---------------------------------------------------------------------------
class TestStatusJsonIndexSource:
"""run_status(json_output=True) with canonical index"""
def test_status_json_includes_lifecycle_health_maturity(self, tmp_path: Path) -> None:
"""JSON output includes lifecycle/health/maturity aggregates from index."""
from paperforge.worker.status import run_status
vault = _minimal_vault(tmp_path)
_ensure_domain_config(vault)
_ensure_exports(vault)
# Create a canonical index with 2 entries
items = [
{
"zotero_key": "AAA",
"lifecycle": "ai_context_ready",
"health": {
"pdf_health": "healthy",
"ocr_health": "healthy",
"note_health": "healthy",
"asset_health": "healthy",
},
"maturity": {"level": 6},
},
{
"zotero_key": "BBB",
"lifecycle": "indexed",
"health": {
"pdf_health": "healthy",
"ocr_health": "healthy",
"note_health": "Formal note missing",
"asset_health": "Missing workspace paths",
},
"maturity": {"level": 1},
},
]
_write_index(vault, items)
code = run_status(vault, json_output=True)
assert code == 0
def test_status_json_fallback_when_index_missing(self, tmp_path: Path, capsys) -> None:
"""When no canonical index, lifecycle/health/maturity fields are None."""
from paperforge.worker.status import run_status
vault = _minimal_vault(tmp_path)
_ensure_domain_config(vault)
_ensure_exports(vault)
code = run_status(vault, json_output=True)
captured = capsys.readouterr().out
assert code == 0
data = json.loads(captured)
assert data["lifecycle_level_counts"] is None
assert data["health_aggregate"] is None
assert data["maturity_distribution"] is None
def test_status_text_output_includes_index_section(self, tmp_path: Path, capsys) -> None:
"""Text output contains 'lifecycle:' line when index is present."""
from paperforge.worker.status import run_status
vault = _minimal_vault(tmp_path)
_ensure_domain_config(vault)
_ensure_exports(vault)
_write_index(vault, [])
code = run_status(vault, json_output=False)
captured = capsys.readouterr().out
assert code == 0
assert "lifecycle:" in captured
assert "health:" in captured
# ---------------------------------------------------------------------------
# Task 3: doctor Index Health section
# ---------------------------------------------------------------------------
class TestDoctorIndexHealth:
"""run_doctor() Index Health section"""
def _run_doctor(self, vault: Path) -> str:
"""Run doctor and return captured stdout."""
from paperforge.worker.status import run_doctor
import io
import sys
old_stdout = sys.stdout
sys.stdout = buf = io.StringIO()
try:
run_doctor(vault)
finally:
sys.stdout = old_stdout
return buf.getvalue()
def test_doctor_includes_index_health_section(self, tmp_path: Path) -> None:
"""Doctor output includes 'Index Health' section when index present."""
vault = _minimal_vault(tmp_path)
_ensure_domain_config(vault)
_write_index(vault, [])
output = self._run_doctor(vault)
assert "Index Health" in output
def test_doctor_index_health_with_mixed_health(self, tmp_path: Path) -> None:
"""Mixed health entries produce correct counts."""
vault = _minimal_vault(tmp_path)
_ensure_domain_config(vault)
items = [
{
"zotero_key": "AAA",
"lifecycle": "ai_context_ready",
"health": {
"pdf_health": "healthy",
"ocr_health": "healthy",
"note_health": "healthy",
"asset_health": "healthy",
},
"maturity": {"level": 6},
},
{
"zotero_key": "BBB",
"lifecycle": "indexed",
"health": {
"pdf_health": "healthy",
"ocr_health": "OCR failed",
"note_health": "Formal note missing",
"asset_health": "healthy",
},
"maturity": {"level": 1},
},
]
_write_index(vault, items)
output = self._run_doctor(vault)
assert "Index Health" in output
# PDF Health: both healthy
assert "PDF Health" in output
# OCR Health: one unhealthy
assert "OCR Health" in output
# Note Health: one unhealthy
assert "Note Health" in output
def test_doctor_index_health_no_index(self, tmp_path: Path) -> None:
"""When no index, doctor shows info message instead of crashing."""
vault = _minimal_vault(tmp_path)
_ensure_domain_config(vault)
output = self._run_doctor(vault)
assert "Index Health" in output
assert "No canonical index" in output