feat(comments): attach combined.md back to bib + fix inline-body lookup
Two changes that, together, complete the loop from comment dir to a single MD corpus per item in Zotero. The inline-body lookup in cli.comments was querying items.key for a CMS comment ID — but item keys are bib hashes (e.g. M5IR9YAE), so every lookup returned None and every previously-extracted combined.md was written with an empty inline body. Replace it with a one-shot URL → comment_id parse over the items table, materialized into a cache (164k+ rows × LIKE matching can't use any index). Wire combined.md back to bib so it flows to Zotero via `bib sync`. walker.walk_and_extract now accepts an `on_combined(comment_id, path)` callback invoked on the main thread for every dir that ends up with a combined.md — written or pre-existing. Skipped dirs get the callback too, so backfill-attaching previously-extracted MDs on a re-run works without --force. CLI exposes --attach/--no-attach (default on; auto-off under --no-bib). Idempotent: pre-loads the set of items already carrying a combined.md attachment. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -11,24 +11,32 @@ Phase 1 (this file):
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
|
||||
app = typer.Typer(no_args_is_help=True)
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_DEFAULT_ROOT = Path(".state/comments")
|
||||
|
||||
|
||||
def _bib_lookup_factory(use_bib: bool):
|
||||
"""Return a callable ``comment_id -> inline body``. Empty if --no-bib.
|
||||
def _build_bib_helpers(use_bib: bool, attach: bool):
|
||||
"""Return ``(body_lookup, attach_callback)``.
|
||||
|
||||
Opens its own sqlite connection with ``check_same_thread=False`` so
|
||||
walker worker threads can call it directly. Read-only path; SQLite's
|
||||
default thread-safe mode handles the concurrent reads.
|
||||
``body_lookup(comment_id) -> str`` is called from worker threads and
|
||||
returns the inline comment body from bib.sqlite (or "" if missing or
|
||||
--no-bib). Backed by a comment_id → (item_key, body) cache built
|
||||
once at startup — bib has 164k+ rows and the URL → comment_id parse
|
||||
can't use any index, so we materialize the whole map up front.
|
||||
|
||||
``attach_callback(comment_id, md_path) -> None`` runs on the main
|
||||
thread and registers combined.md as a bib attachment, skipping items
|
||||
that already have one. Returns None when --no-bib or --no-attach.
|
||||
"""
|
||||
if not use_bib:
|
||||
return lambda _cid: ""
|
||||
return lambda _cid: "", None
|
||||
|
||||
import sqlite3
|
||||
|
||||
@@ -37,13 +45,52 @@ def _bib_lookup_factory(use_bib: bool):
|
||||
db = str(path("db.bib"))
|
||||
con = sqlite3.connect(db, check_same_thread=False)
|
||||
|
||||
def lookup(comment_id: str) -> str:
|
||||
row = con.execute(
|
||||
"SELECT abstract FROM items WHERE key = ?", (comment_id,)
|
||||
).fetchone()
|
||||
return (row[0] if row and row[0] else "") or ""
|
||||
cache: dict[str, tuple[str, str]] = {}
|
||||
for row in con.execute(
|
||||
"SELECT key, url, abstract FROM items "
|
||||
"WHERE url LIKE 'https://www.regulations.gov/comment/%'"
|
||||
):
|
||||
cid = (row[1] or "").rsplit("/", 1)[-1]
|
||||
if cid:
|
||||
cache[cid] = (row[0], row[2] or "")
|
||||
|
||||
return lookup
|
||||
def body_lookup(comment_id: str) -> str:
|
||||
return cache.get(comment_id, ("", ""))[1]
|
||||
|
||||
if not attach:
|
||||
return body_lookup, None
|
||||
|
||||
from bib import connect
|
||||
|
||||
store = connect()
|
||||
existing: set[str] = {
|
||||
row[0]
|
||||
for row in store._con().execute(
|
||||
"SELECT i.key FROM attachments a JOIN items i ON a.item_id = i.id "
|
||||
"WHERE a.filename = 'combined.md'"
|
||||
)
|
||||
}
|
||||
|
||||
def attach_callback(comment_id: str, md_path: Path) -> None:
|
||||
info = cache.get(comment_id)
|
||||
if not info:
|
||||
return # comment not in bib
|
||||
item_key = info[0]
|
||||
if item_key in existing:
|
||||
return # already attached on a prior run
|
||||
try:
|
||||
store.attach_file(item_key, md_path, title="combined.md")
|
||||
existing.add(item_key)
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("attach combined.md failed for %s: %s", item_key, e)
|
||||
|
||||
return body_lookup, attach_callback
|
||||
|
||||
|
||||
# Back-compat alias — older code/tests may still import this name.
|
||||
def _bib_lookup_factory(use_bib: bool):
|
||||
body, _ = _build_bib_helpers(use_bib, attach=False)
|
||||
return body
|
||||
|
||||
|
||||
@app.command()
|
||||
@@ -64,6 +111,12 @@ def extract(
|
||||
"--bib/--no-bib",
|
||||
help="Look up inline body from bib.sqlite items.abstract.",
|
||||
),
|
||||
attach: bool = typer.Option(
|
||||
True,
|
||||
"--attach/--no-attach",
|
||||
help="Attach combined.md back to its bib item (so it flows to "
|
||||
"Zotero via `bib sync`). Implies --bib; no-op when --no-bib.",
|
||||
),
|
||||
) -> None:
|
||||
"""Walk comment dirs and write combined.md (PDF/DOCX → text)."""
|
||||
from rex.comments.walker import walk_and_extract
|
||||
@@ -72,7 +125,7 @@ def extract(
|
||||
typer.echo(f"root not found: {root}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
lookup = _bib_lookup_factory(use_bib)
|
||||
lookup, attach_cb = _build_bib_helpers(use_bib, attach=use_bib and attach)
|
||||
stats = walk_and_extract(
|
||||
root,
|
||||
inline_body_lookup=lookup,
|
||||
@@ -80,6 +133,7 @@ def extract(
|
||||
limit=limit or None,
|
||||
workers=workers,
|
||||
force=force,
|
||||
on_combined=attach_cb,
|
||||
)
|
||||
for k, v in stats.items():
|
||||
typer.echo(f" {k}: {v}")
|
||||
|
||||
@@ -27,6 +27,7 @@ def walk_and_extract(
|
||||
limit: int | None = None,
|
||||
workers: int = 8,
|
||||
force: bool = False,
|
||||
on_combined: Callable[[str, Path], None] | None = None,
|
||||
) -> dict[str, int]:
|
||||
"""Process every comment dir under *root*.
|
||||
|
||||
@@ -35,22 +36,35 @@ def walk_and_extract(
|
||||
``comment_id`` and returns the inline body text from bib.sqlite (or
|
||||
"" if missing).
|
||||
|
||||
*on_combined*, if given, is called on the main thread with
|
||||
``(comment_id, combined_md_path)`` for every dir that ends up with a
|
||||
combined.md — both newly-written ones and pre-existing skips. Lets
|
||||
callers attach the markdown to bib without crossing thread boundaries.
|
||||
|
||||
Returns counts: ``{"written": N, "skipped": N, "failed": N}``.
|
||||
"""
|
||||
candidate_dirs, skipped = _collect_dirs(
|
||||
candidate_dirs, skipped_dirs = _collect_dirs(
|
||||
root, docket=docket, limit=limit, force=force
|
||||
)
|
||||
stats = {"written": 0, "skipped": skipped, "failed": 0}
|
||||
|
||||
if not candidate_dirs:
|
||||
return stats
|
||||
stats = {"written": 0, "skipped": len(skipped_dirs), "failed": 0}
|
||||
|
||||
if candidate_dirs:
|
||||
with ThreadPoolExecutor(max_workers=workers) as pool:
|
||||
futures = {
|
||||
pool.submit(_one, d, inline_body_lookup, force): d for d in candidate_dirs
|
||||
pool.submit(_one, d, inline_body_lookup, force): d
|
||||
for d in candidate_dirs
|
||||
}
|
||||
for fut in as_completed(futures):
|
||||
stats[fut.result()] += 1
|
||||
status = fut.result()
|
||||
stats[status] += 1
|
||||
if on_combined and status == "written":
|
||||
cdir = futures[fut]
|
||||
on_combined(cdir.name, cdir / _COMBINED)
|
||||
|
||||
if on_combined:
|
||||
for cdir in skipped_dirs:
|
||||
on_combined(cdir.name, cdir / _COMBINED)
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
@@ -60,10 +74,16 @@ def _collect_dirs(
|
||||
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)."""
|
||||
) -> tuple[list[Path], list[Path]]:
|
||||
"""Return ``(dirs_to_process, dirs_skipped)``.
|
||||
|
||||
``dirs_skipped`` are dirs that already have combined.md and would not
|
||||
be re-processed. We return them as a list (not a count) so callers
|
||||
can still drive per-dir post-processing — e.g. attaching the existing
|
||||
combined.md to bib on a re-run.
|
||||
"""
|
||||
todo: list[Path] = []
|
||||
skipped = 0
|
||||
skipped: list[Path] = []
|
||||
for docket_dir in sorted(root.iterdir()):
|
||||
if not docket_dir.is_dir():
|
||||
continue
|
||||
@@ -73,7 +93,7 @@ def _collect_dirs(
|
||||
if not cdir.is_dir():
|
||||
continue
|
||||
if (cdir / _COMBINED).is_file() and not force:
|
||||
skipped += 1
|
||||
skipped.append(cdir)
|
||||
continue
|
||||
todo.append(cdir)
|
||||
if limit and len(todo) >= limit:
|
||||
|
||||
@@ -57,3 +57,60 @@ def test_stats_after_extract(tmp_path: Path):
|
||||
result = runner.invoke(app, ["stats", "--root", str(tmp_path)])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "1" in result.output # at least one comment counted
|
||||
|
||||
|
||||
def test_extract_attaches_combined_md_to_bib(tmp_path: Path, monkeypatch):
|
||||
"""End-to-end: extract writes combined.md AND attaches it to the bib item."""
|
||||
from bib import connect
|
||||
from bib.item import Source
|
||||
|
||||
bib_db = tmp_path / "bib.sqlite"
|
||||
monkeypatch.setattr("conf.path", lambda _: bib_db)
|
||||
|
||||
cdir = _seed(tmp_path)
|
||||
comment_id = cdir.name # CMS-2024-0001-0001
|
||||
|
||||
store = connect(str(bib_db))
|
||||
item = Source(
|
||||
title=f"Comment {comment_id}",
|
||||
url=f"https://www.regulations.gov/comment/{comment_id}",
|
||||
)
|
||||
item_key = store.upsert(item)
|
||||
store.close()
|
||||
|
||||
result = runner.invoke(
|
||||
app,
|
||||
[
|
||||
"extract",
|
||||
"--root",
|
||||
str(tmp_path),
|
||||
"--workers",
|
||||
"1",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert (cdir / "combined.md").is_file()
|
||||
|
||||
store = connect(str(bib_db))
|
||||
rows = list(
|
||||
store._con().execute(
|
||||
"SELECT a.filename FROM attachments a JOIN items i ON a.item_id=i.id "
|
||||
"WHERE i.key=? AND a.filename='combined.md'",
|
||||
(item_key,),
|
||||
)
|
||||
)
|
||||
store.close()
|
||||
assert len(rows) == 1, "combined.md should be attached exactly once"
|
||||
|
||||
# Re-running must not duplicate the attachment (idempotence)
|
||||
runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"])
|
||||
store = connect(str(bib_db))
|
||||
rows = list(
|
||||
store._con().execute(
|
||||
"SELECT COUNT(*) FROM attachments a JOIN items i ON a.item_id=i.id "
|
||||
"WHERE i.key=? AND a.filename='combined.md'",
|
||||
(item_key,),
|
||||
)
|
||||
)
|
||||
store.close()
|
||||
assert rows[0][0] == 1, "second run created a duplicate attachment"
|
||||
|
||||
@@ -75,3 +75,37 @@ def test_walk_and_extract_respects_limit(tmp_path: Path):
|
||||
tmp_path, inline_body_lookup=_stub_inline_body, limit=1, workers=1
|
||||
)
|
||||
assert stats["written"] == 1
|
||||
|
||||
|
||||
def test_on_combined_called_for_written_dirs(tmp_path: Path):
|
||||
a, b = _setup_two_comments(tmp_path)
|
||||
seen: list[tuple[str, Path]] = []
|
||||
|
||||
walk_and_extract(
|
||||
tmp_path,
|
||||
inline_body_lookup=_stub_inline_body,
|
||||
workers=2,
|
||||
on_combined=lambda cid, p: seen.append((cid, p)),
|
||||
)
|
||||
|
||||
assert sorted(cid for cid, _ in seen) == sorted([a.name, b.name])
|
||||
assert all(p.name == "combined.md" for _, p in seen)
|
||||
|
||||
|
||||
def test_on_combined_called_for_skipped_dirs(tmp_path: Path):
|
||||
"""Pre-existing combined.md still gets the callback — so previously
|
||||
extracted MDs can be backfill-attached on a later run."""
|
||||
a, b = _setup_two_comments(tmp_path)
|
||||
(a / "combined.md").write_text("---\n---\nstale\n")
|
||||
|
||||
seen: list[str] = []
|
||||
stats = walk_and_extract(
|
||||
tmp_path,
|
||||
inline_body_lookup=_stub_inline_body,
|
||||
workers=1,
|
||||
on_combined=lambda cid, _p: seen.append(cid),
|
||||
)
|
||||
|
||||
assert stats["written"] == 1
|
||||
assert stats["skipped"] == 1
|
||||
assert sorted(seen) == sorted([a.name, b.name])
|
||||
|
||||
Reference in New Issue
Block a user