feat(comments): parallel walker + DuckDB view (refs #253)

- walker.walk_and_extract iterates docket/cid dirs under root,
  parallelizes per-comment extraction across ThreadPoolExecutor.
  PyMuPDF releases the GIL during decode so threads scale.
  Returns {written, skipped, failed} counts; honors --docket and
  --limit; resume rule via existing combined.md.
- view.rebuild_index walks combined.md files and writes _index.csv;
  view.register exposes it as a DuckDB view (comments_index) for
  filtering/aggregation. Bodies stay on disk; view.body_path is the
  canonical lookup helper for downstream consumers (#254/#255).

CLI + integration test land in the next batch.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/rex/comments/view.py

@@ -0,0 +1,84 @@
+"""DuckDB view over the combined.md index.
+
+We don't load body text into DuckDB — bodies stay on disk. The view
+exposes per-comment metadata (status counts, chars) for filtering and
+aggregation. Downstream consumers read body text from disk via
+``body_path()``.
+"""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+
+import duckdb
+
+from rex.comments.combine import parse_combined
+
+_INDEX = "_index.csv"
+_VIEW = "comments_index"
+_COLUMNS = (
+    "comment_id",
+    "docket_id",
+    "extracted_at",
+    "chars",
+    "n_attachments",
+    "n_ok",
+    "n_ocr_needed",
+    "n_failed",
+)
+
+
+def rebuild_index(root: Path) -> Path:
+    """Walk *root* and write _index.csv. Returns the path written."""
+    out = root / _INDEX
+    rows = list(_iter_index_rows(root))
+    with out.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow(_COLUMNS)
+        w.writerows(rows)
+    return out
+
+
+def register(con: duckdb.DuckDBPyConnection, *, root: Path) -> None:
+    """Create the ``comments_index`` view in *con* over *root*/_index.csv."""
+    csv_path = root / _INDEX
+    if not csv_path.is_file():
+        rebuild_index(root)
+    con.execute(
+        f"CREATE OR REPLACE VIEW {_VIEW} AS "
+        f"SELECT * FROM read_csv('{csv_path}', header=true, auto_detect=true)"
+    )
+
+
+def body_path(root: Path, *, comment_id: str, docket_id: str) -> Path:
+    """Filesystem path of the combined.md for one comment."""
+    return root / docket_id / comment_id / "combined.md"
+
+
+# ── internals ─────────────────────────────────────────────────────
+
+
+def _iter_index_rows(root: Path):
+    for docket_dir in sorted(root.iterdir()):
+        if not docket_dir.is_dir():
+            continue
+        for cdir in sorted(docket_dir.iterdir()):
+            md = cdir / "combined.md"
+            if not md.is_file():
+                continue
+            try:
+                fm, _body = parse_combined(md.read_text())
+            except (ValueError, OSError):
+                continue
+            atts = fm.get("attachments") or []
+            yield (
+                fm.get("comment_id", ""),
+                fm.get("docket_id", ""),
+                fm.get("extracted_at", ""),
+                fm.get("chars", 0),
+                len(atts),
+                sum(1 for a in atts if a.get("status") == "ok"),
+                sum(1 for a in atts if a.get("status") == "ocr_needed"),
+                sum(1 for a in atts if a.get("status") == "failed"),
+            )

src/rex/comments/walker.py

@@ -0,0 +1,102 @@
+"""Walk .state/comments/ and run extract_comment over every dir.
+
+Parallelized with ThreadPoolExecutor. PyMuPDF releases the GIL during
+PDF decoding so threads (not processes) are the right primitive — keeps
+shared state simple and avoids fork overhead per attachment.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+
+from rex.comments.combine import extract_comment
+
+log = logging.getLogger(__name__)
+
+_COMBINED = "combined.md"
+
+
+def walk_and_extract(
+    root: Path,
+    *,
+    inline_body_lookup: Callable[[str], str],
+    docket: str | None = None,
+    limit: int | None = None,
+    workers: int = 8,
+    force: bool = False,
+) -> dict[str, int]:
+    """Process every comment dir under *root*.
+
+    *root* is the dir containing per-docket subdirs (typically
+    ``.state/comments/``). *inline_body_lookup* is called with each
+    ``comment_id`` and returns the inline body text from bib.sqlite (or
+    "" if missing).
+
+    Returns counts: ``{"written": N, "skipped": N, "failed": N}``.
+    """
+    candidate_dirs, skipped = _collect_dirs(
+        root, docket=docket, limit=limit, force=force
+    )
+    stats = {"written": 0, "skipped": skipped, "failed": 0}
+
+    if not candidate_dirs:
+        return stats
+
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        futures = {
+            pool.submit(_one, d, inline_body_lookup, force): d for d in candidate_dirs
+        }
+        for fut in as_completed(futures):
+            stats[fut.result()] += 1
+    return stats
+
+
+def _collect_dirs(
+    root: Path,
+    *,
+    docket: str | None,
+    limit: int | None,
+    force: bool,
+) -> tuple[list[Path], int]:
+    """Return (dirs to process, count of dirs skipped due to existing combined.md)."""
+    todo: list[Path] = []
+    skipped = 0
+    for docket_dir in sorted(root.iterdir()):
+        if not docket_dir.is_dir():
+            continue
+        if docket and docket_dir.name != docket:
+            continue
+        for cdir in sorted(docket_dir.iterdir()):
+            if not cdir.is_dir():
+                continue
+            if (cdir / _COMBINED).is_file() and not force:
+                skipped += 1
+                continue
+            todo.append(cdir)
+            if limit and len(todo) >= limit:
+                return todo, skipped
+    return todo, skipped
+
+
+def _one(
+    comment_dir: Path,
+    inline_body_lookup: Callable[[str], str],
+    force: bool,
+) -> str:
+    """Process one comment dir. Returns "written" / "skipped" / "failed"."""
+    if (comment_dir / _COMBINED).is_file() and not force:
+        return "skipped"
+    try:
+        body = inline_body_lookup(comment_dir.name)
+    except Exception as e:  # noqa: BLE001
+        log.warning("inline body lookup failed for %s: %s", comment_dir.name, e)
+        body = ""
+    try:
+        extract_comment(comment_dir, inline_body=body, force=force)
+    except Exception as e:  # noqa: BLE001
+        log.warning("extract_comment failed for %s: %s", comment_dir, e)
+        return "failed"
+    return "written"

tests/rex/comments/test_view.py

@@ -0,0 +1,58 @@
+"""DuckDB view over combined.md index."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import duckdb
+
+from rex.comments import extract_comment
+from rex.comments.view import body_path, rebuild_index, register
+
+
+def _seed_two(tmp_path: Path) -> None:
+    import fitz
+
+    for cid_suffix, text in (("0001", "Long body " * 20), ("0002", "x")):
+        cdir = tmp_path / "CMS-2024-0001" / f"CMS-2024-0001-{cid_suffix}"
+        cdir.mkdir(parents=True)
+        doc = fitz.open()
+        doc.new_page().insert_text((50, 72), text)
+        doc.save(str(cdir / "attachment_1.pdf"))
+        doc.close()
+        extract_comment(cdir, inline_body="inline body")
+
+
+def test_rebuild_index_writes_csv(tmp_path: Path):
+    _seed_two(tmp_path)
+    csv = rebuild_index(tmp_path)
+    assert csv == tmp_path / "_index.csv"
+    rows = csv.read_text().splitlines()
+    assert rows[0].startswith("comment_id,docket_id,")
+    assert len(rows) == 3  # header + 2 comments
+
+
+def test_register_view_queryable(tmp_path: Path):
+    _seed_two(tmp_path)
+    rebuild_index(tmp_path)
+
+    con = duckdb.connect(":memory:")
+    register(con, root=tmp_path)
+    rows = con.execute(
+        "SELECT comment_id, n_attachments, n_ok FROM comments_index ORDER BY comment_id"
+    ).fetchall()
+    assert len(rows) == 2
+    assert rows[0][0] == "CMS-2024-0001-0001"
+    assert rows[0][1] == 1  # n_attachments
+    assert rows[0][2] == 1  # n_ok
+    # Second row: short text → ocr_needed
+    assert rows[1][2] == 0
+
+
+def test_body_path():
+    p = body_path(
+        Path("/x/.state/comments"),
+        comment_id="CMS-2024-0001-0099",
+        docket_id="CMS-2024-0001",
+    )
+    assert p == Path("/x/.state/comments/CMS-2024-0001/CMS-2024-0001-0099/combined.md")

tests/rex/comments/test_walker.py

@@ -0,0 +1,77 @@
+"""Walker — iterate .state/comments/, look up inline body, parallelize."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import fitz
+
+from rex.comments.walker import walk_and_extract
+
+
+def _make_pdf(path: Path, text: str = "x" * 200) -> None:
+    doc = fitz.open()
+    doc.new_page().insert_text((50, 72), text)
+    doc.save(str(path))
+    doc.close()
+
+
+def _setup_two_comments(root: Path) -> tuple[Path, Path]:
+    a = root / "CMS-2024-0001" / "CMS-2024-0001-0001"
+    b = root / "CMS-2024-0001" / "CMS-2024-0001-0002"
+    a.mkdir(parents=True)
+    b.mkdir(parents=True)
+    _make_pdf(a / "attachment_1.pdf", "Comment A body content. " * 10)
+    _make_pdf(b / "attachment_1.pdf", "Comment B body content. " * 10)
+    return a, b
+
+
+def _stub_inline_body(_cid: str) -> str:
+    return "inline body for stub"
+
+
+def test_walk_and_extract_writes_all_combined(tmp_path: Path):
+    a, b = _setup_two_comments(tmp_path)
+
+    stats = walk_and_extract(tmp_path, inline_body_lookup=_stub_inline_body, workers=2)
+
+    assert (a / "combined.md").is_file()
+    assert (b / "combined.md").is_file()
+    assert stats["written"] == 2
+    assert stats["skipped"] == 0
+
+
+def test_walk_and_extract_skips_existing(tmp_path: Path):
+    a, _b = _setup_two_comments(tmp_path)
+    (a / "combined.md").write_text("---\ncomment_id: x\n---\n\npre-existing\n")
+
+    stats = walk_and_extract(tmp_path, inline_body_lookup=_stub_inline_body, workers=1)
+
+    assert stats["written"] == 1  # only the second one
+    assert stats["skipped"] == 1
+    assert "pre-existing" in (a / "combined.md").read_text()  # untouched
+
+
+def test_walk_and_extract_filters_by_docket(tmp_path: Path):
+    _setup_two_comments(tmp_path)
+    other = tmp_path / "CMS-2099-9999" / "CMS-2099-9999-0001"
+    other.mkdir(parents=True)
+    _make_pdf(other / "attachment_1.pdf", "other docket content. " * 10)
+
+    stats = walk_and_extract(
+        tmp_path,
+        inline_body_lookup=_stub_inline_body,
+        docket="CMS-2099-9999",
+        workers=1,
+    )
+
+    assert stats["written"] == 1
+    assert (other / "combined.md").is_file()
+
+
+def test_walk_and_extract_respects_limit(tmp_path: Path):
+    _setup_two_comments(tmp_path)
+    stats = walk_and_extract(
+        tmp_path, inline_body_lookup=_stub_inline_body, limit=1, workers=1
+    )
+    assert stats["written"] == 1