Refactor corpus I/O, parsers, and pipeline orchestration

.github/workflows/test.yml

@@ -25,5 +25,11 @@ jobs:
           python -m pip install --upgrade pip
           pip install -e ".[dev]"
 
+      - name: Ruff
+        run: ruff check medbots tests
+
+      - name: Mypy
+        run: mypy medbots
+
       - name: Run pytest
         run: pytest tests/ -q

medbots/cli.py

@@ -1,8 +1,11 @@
 #!/usr/bin/env python3
 """CLI entry points for medbots-core."""
+
 from __future__ import annotations
 
 import argparse
+import json
+import os
 import subprocess
 import sys
 from pathlib import Path
@@ -12,9 +15,14 @@
 from medbots.extract_pdf_text import run as run_extract
 from medbots.import_apple_health import run_import
 from medbots.init_instance import init as run_init
+from medbots.local_structure_pdfs import run as run_structure
+from medbots.log_config import get_logger, setup_logging
 from medbots.pipeline.run import run_pipeline
+from medbots.pipeline.validate_corpus import main as validate_corpus_main
 from medbots.scan_sources import scan as run_scan
 
+logger = get_logger(__name__)
+
 
 def _cmd_init(args: argparse.Namespace) -> int:
     run_init(args.path, force=args.force)
@@ -53,41 +61,45 @@ def _cmd_import_apple_health(args: argparse.Namespace) -> int:
 
 def _cmd_validate_apple_health(args: argparse.Namespace) -> int:
     corpus = resolve_corpus(args.corpus)
-    return subprocess.call([sys.executable, "-m", "medbots.validate_apple_health", "--corpus", str(corpus)])
+    return subprocess.call(
+        [sys.executable, "-m", "medbots.validate_apple_health", "--corpus", str(corpus)]
+    )
 
 
 def _cmd_structure(args: argparse.Namespace) -> int:
     root = Path(args.bot_root).resolve() if args.bot_root else find_bot_root()
-    corpus = resolve_corpus(args.corpus) if args.corpus else resolve_corpus(root / "structured_database")
-    cmd = [
-        sys.executable,
-        "-m",
-        "medbots.local_structure_pdfs",
-        "--corpus",
-        str(corpus),
-    ]
-    if args.force:
-        cmd.append("--force")
-    for src in args.source:
-        cmd.extend(["--source", src])
-    if args.dry_run:
-        cmd.append("--dry-run")
-    return subprocess.call(cmd)
+    corpus = (
+        resolve_corpus(args.corpus) if args.corpus else resolve_corpus(root / "structured_database")
+    )
+    source_filter = {s.strip().lower() for s in args.source if s.strip()} or None
+    stats = run_structure(
+        corpus,
+        dry_run=args.dry_run,
+        force=args.force,
+        sources=source_filter,
+    )
+    print(json.dumps(stats, ensure_ascii=False, indent=2))
+    if stats.get("errors"):
+        for err in stats["errors"]:
+            logger.warning("%s", err)
+    return 0
 
 
 def _cmd_pipeline(args: argparse.Namespace) -> int:
     root = Path(args.bot_root).resolve() if args.bot_root else find_bot_root()
     corpus = resolve_corpus(args.corpus) if args.corpus else None
     try:
         run_pipeline(bot_root=root, corpus=corpus)
-    except subprocess.CalledProcessError as exc:
-        return exc.returncode or 1
+    except SystemExit as exc:
+        code = exc.code
+        return code if isinstance(code, int) else 1
     return 0
 
 
 def _cmd_validate(args: argparse.Namespace) -> int:
     corpus = resolve_corpus(args.corpus)
-    return subprocess.call([sys.executable, "-m", "medbots.pipeline.validate_corpus", "--corpus", str(corpus)])
+    os.environ["MEDBOTS_CORPUS_PATH"] = str(corpus)
+    return validate_corpus_main()
 
 
 def _cmd_patient_dob(args: argparse.Namespace) -> int:
@@ -97,6 +109,7 @@ def _cmd_patient_dob(args: argparse.Namespace) -> int:
 
 
 def main(argv: list[str] | None = None) -> int:
+    setup_logging()
     parser = argparse.ArgumentParser(
         prog="medbots",
         description="Medical corpus tools: scan PDFs, extract text, parse labs, run pipeline",
@@ -126,10 +139,14 @@ def main(argv: list[str] | None = None) -> int:
     p_ah.add_argument("--zip", type=Path, required=True, help="Apple Health export.zip from iPhone")
     p_ah.add_argument("--bot-root", type=Path)
     p_ah.add_argument("--corpus", type=Path)
-    p_ah.add_argument("--copy-zip", action="store_true", help="Archive zip under sources/apple_health/")
+    p_ah.add_argument(
+        "--copy-zip", action="store_true", help="Archive zip under sources/apple_health/"
+    )
     p_ah.set_defaults(func=_cmd_import_apple_health)
 
-    p_ah_val = sub.add_parser("validate-apple-health", help="Check fitness/ after Apple Health import")
+    p_ah_val = sub.add_parser(
+        "validate-apple-health", help="Check fitness/ after Apple Health import"
+    )
     p_ah_val.add_argument("--corpus", type=Path)
     p_ah_val.set_defaults(func=_cmd_validate_apple_health)
 

medbots/cli_args.py

@@ -0,0 +1,38 @@
+"""Shared argparse helpers for corpus CLI entrypoints."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from collections.abc import Callable
+from pathlib import Path
+
+from medbots.corpus_io import default_corpus_root
+
+
+def add_corpus_argument(parser: argparse.ArgumentParser, default: Path | None = None) -> None:
+    parser.add_argument(
+        "--corpus",
+        type=Path,
+        default=default,
+        help="structured_database path override",
+    )
+
+
+def resolve_corpus_from_args(args: argparse.Namespace) -> Path:
+    corpus_arg = getattr(args, "corpus", None)
+    if corpus_arg is None:
+        corpus_arg = default_corpus_root()
+    corpus = corpus_arg.expanduser().resolve()
+    if not corpus.is_dir():
+        print(f"ERROR: corpus not found: {corpus}", file=sys.stderr)
+        sys.exit(1)
+    return corpus
+
+
+def corpus_main_wrapper(description: str, run_fn: Callable[[Path], int]) -> int:
+    ap = argparse.ArgumentParser(description=description)
+    add_corpus_argument(ap)
+    args = ap.parse_args()
+    corpus = resolve_corpus_from_args(args)
+    return run_fn(corpus)

medbots/config.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 """Load bot_config.json and resolve feature flags (with env overrides)."""
+
 from __future__ import annotations
 
 import json

medbots/corpus_io.py

@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 """Shared corpus I/O: manifest, labs, patient profile, vendor index."""
+
 from __future__ import annotations
 
 import json
 import os
+import tempfile
 from pathlib import Path
 from typing import Any
 
@@ -43,35 +45,55 @@ def resolve_corpus(path: Path | str | None = None) -> Path:
     return Path(path).expanduser().resolve()
 
 
+def atomic_write_text(path: Path, text: str) -> None:
+    """Write text atomically via temp file in the same directory."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(
+        dir=path.parent,
+        prefix=f".{path.name}.",
+        suffix=".tmp",
+    )
+    tmp_path = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            fh.write(text)
+        os.replace(tmp_path, path)
+    except Exception:
+        tmp_path.unlink(missing_ok=True)
+        raise
+
+
+def load_json(path: Path, default: Any) -> Any:
+    if path.exists():
+        return json.loads(path.read_text(encoding="utf-8"))
+    return default
+
+
+def write_json(path: Path, data: Any, *, trailing_newline: bool = False) -> None:
+    text = json.dumps(data, ensure_ascii=False, indent=2)
+    if trailing_newline and not text.endswith("\n"):
+        text += "\n"
+    atomic_write_text(path, text)
+
+
 def empty_manifest() -> dict[str, Any]:
     return {"version": 1, "pdfs": [], "images": [], "meta": {}}
 
 
 def load_manifest(corpus: Path) -> dict[str, Any]:
-    p = corpus / "manifest.json"
-    if p.exists():
-        return json.loads(p.read_text(encoding="utf-8"))
-    return empty_manifest()
+    return load_json(corpus / "manifest.json", empty_manifest())
 
 
 def write_manifest(corpus: Path, data: dict[str, Any]) -> None:
-    (corpus / "manifest.json").write_text(
-        json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
-    )
+    write_json(corpus / "manifest.json", data)
 
 
 def load_labs(corpus: Path) -> dict[str, Any]:
-    p = corpus / "LABS_NORMALIZED.json"
-    if p.exists():
-        return json.loads(p.read_text(encoding="utf-8"))
-    return {"rows": []}
+    return load_json(corpus / "LABS_NORMALIZED.json", {"rows": []})
 
 
 def write_labs(corpus: Path, data: dict[str, Any]) -> None:
-    text = json.dumps(data, ensure_ascii=False, indent=2)
-    if not text.endswith("\n"):
-        text += "\n"
-    (corpus / "LABS_NORMALIZED.json").write_text(text, encoding="utf-8")
+    write_json(corpus / "LABS_NORMALIZED.json", data, trailing_newline=True)
 
 
 def load_patient_dob(corpus: Path) -> str:

medbots/corpus_writers.py

@@ -1,15 +1,18 @@
 """Shared corpus writers (doc_text, labs, supplements) — no Grok/LLM."""
+
 from __future__ import annotations
 
 import json
 import re
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
+from medbots.corpus_io import load_labs, write_labs
+
 
 def _utc_date_slug() -> str:
-    return datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    return datetime.now(UTC).strftime("%Y-%m-%d")
 
 
 def _load_patient_profile(corpus: Path) -> dict[str, Any]:
@@ -50,8 +53,7 @@ def _write_to_extracted_images_md(corpus: Path, section_id: str, extracted: dict
     else:
         md_path.write_text(
             "# Извлечение из Telegram ingest\n\n"
-            f"Пациент: **{patient}**, д.р. **{dob}**.\n"
-            + entry.lstrip("\n"),
+            f"Пациент: **{patient}**, д.р. **{dob}**.\n" + entry.lstrip("\n"),
             encoding="utf-8",
         )
 
@@ -146,29 +148,15 @@ def _write_doc_text_md(
 def _append_lab_rows(corpus: Path, lab_rows: list[dict], source_path_rel: str) -> int:
     if not lab_rows:
         return 0
-    labs_path = corpus / "LABS_NORMALIZED.json"
-    if labs_path.exists():
-        labs = json.loads(labs_path.read_text(encoding="utf-8"))
-    else:
-        labs = {"rows": []}
+    labs = load_labs(corpus)
     existing_rows: list[dict] = list(labs.get("rows") or [])
-    existing_keys = {
-        (r.get("canonical_key", ""), r.get("specimen_date", ""))
-        for r in existing_rows
-    }
-    added = 0
     for row in lab_rows:
         row = dict(row)
         row["source_path"] = source_path_rel
-        key = (row.get("canonical_key", ""), row.get("specimen_date", ""))
-        if key in existing_keys:
-            continue
         existing_rows.append(row)
-        existing_keys.add(key)
-        added += 1
     labs["rows"] = existing_rows
-    labs_path.write_text(json.dumps(labs, ensure_ascii=False, indent=2), encoding="utf-8")
-    return added
+    write_labs(corpus, labs)
+    return len(lab_rows)
 
 
 def _append_supplement_mentions(corpus: Path, mentions: list[dict], source_path_rel: str) -> int: