feat(comments): write per-attachment .md siblings + attach to bib

Each comment dir now produces, alongside combined.md, one
``<original>.md`` sibling per attachment containing just that
attachment's extracted text + a small YAML frontmatter (status,
chars, source_file). Lets Zotero (and anything else consuming the
corpus) navigate to a single attachment's transcript instead of
always opening the aggregated combined.md.

- combine.extract_comment writes siblings during fresh extraction.
- New combine.derive_siblings_from_combined parses an existing
  combined.md and emits the missing siblings without re-decoding the
  underlying PDFs/DOCXs — the backfill path for the 23k items already
  extracted before this change.
- walker renames its callback to on_extracted(comment_id, comment_dir)
  and runs the sibling-backfill for any skipped dir before invoking
  the callback, so callers always see a complete set of MDs.
- cli.comments attach callback now glob('*.md')s the dir and attaches
  each missing one; the existing-attachments cache key changes from
  item_key to (item_key, filename) to keep idempotence at the per-file
  level.

Per-attachment metadata extraction (author/org from letterhead +
signature lines) tracked separately as #416.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
kert
2026-04-27 18:26:18 -04:00
parent dfa2590e06
commit 684fb7fa1e
6 changed files with 225 additions and 75 deletions

View File

@@ -63,26 +63,32 @@ def _build_bib_helpers(use_bib: bool, attach: bool):
from bib import connect
store = connect()
existing: set[str] = {
row[0]
# Pre-load every existing markdown attachment so re-runs are no-ops
# at the per-(item, filename) level. Both combined.md and the per-
# attachment <name>.md siblings live in this set.
existing: set[tuple[str, str]] = {
(row[0], row[1])
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'"
"SELECT i.key, a.filename FROM attachments a "
"JOIN items i ON a.item_id = i.id "
"WHERE a.filename LIKE '%.md'"
)
}
def attach_callback(comment_id: str, md_path: Path) -> None:
def attach_callback(comment_id: str, comment_dir: 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)
for md_path in sorted(comment_dir.glob("*.md")):
key = (item_key, md_path.name)
if key in existing:
continue
try:
store.attach_file(item_key, md_path, title=md_path.name)
existing.add(key)
except Exception as e: # noqa: BLE001
log.warning("attach %s failed for %s: %s", md_path.name, item_key, e)
return body_lookup, attach_callback
@@ -133,7 +139,7 @@ def extract(
limit=limit or None,
workers=workers,
force=force,
on_combined=attach_cb,
on_extracted=attach_cb,
)
for k, v in stats.items():
typer.echo(f" {k}: {v}")

View File

@@ -7,11 +7,16 @@ Public API:
from __future__ import annotations
from rex.comments.combine import extract_comment, parse_combined
from rex.comments.combine import (
derive_siblings_from_combined,
extract_comment,
parse_combined,
)
from rex.comments.extract import ExtractResult, extract_attachment
__all__ = [
"ExtractResult",
"derive_siblings_from_combined",
"extract_attachment",
"extract_comment",
"parse_combined",

View File

@@ -27,6 +27,11 @@ def extract_comment(
) -> Path:
"""Extract every attachment in *comment_dir* and write *combined.md*.
Also writes one ``<filename>.md`` sibling per attachment with that
attachment's text only — useful when downstream tools (Zotero) want
to navigate to a single attachment's transcript instead of the
aggregated combined.md.
Returns the path to combined.md. If it already exists and ``force``
is False, returns it without doing any work (resume rule).
@@ -41,13 +46,21 @@ def extract_comment(
attachments_meta: list[dict[str, Any]] = []
body_sections: list[str] = []
sibling_payloads: list[tuple[str, str]] = [] # (sibling_filename, contents)
for path in sorted(_attachment_paths(comment_dir)):
result = extract_attachment(path)
attachments_meta.append(
{"file": path.name, "status": result.status, "chars": result.chars}
)
body_sections.append(_render_section(path.name, result))
section = _render_section(path.name, result)
body_sections.append(section)
sibling_payloads.append(
(
_sibling_name(path.name),
_sibling_contents(path.name, comment_id, docket_id, result),
)
)
inline_section = (
f"## Inline comment\n\n{inline_body.strip()}\n" if inline_body.strip() else ""
@@ -68,9 +81,44 @@ def extract_comment(
tmp = out.with_suffix(out.suffix + ".tmp")
tmp.write_text(_emit(frontmatter, body_text))
os.replace(tmp, out)
for sibling_name, contents in sibling_payloads:
_write_sibling(comment_dir, sibling_name, contents)
return out
def derive_siblings_from_combined(comment_dir: Path) -> list[Path]:
"""Backfill: write per-attachment sibling MDs from an existing combined.md.
For dirs that were extracted before sibling-write existed, this
parses combined.md and emits the sibling files without re-decoding
the original PDFs/DOCXs. Idempotent — skips siblings that already
exist with non-zero size.
"""
combined = comment_dir / _FILENAME
if not combined.is_file():
return []
fm, body = parse_combined(combined.read_text())
docket_id = fm.get("docket_id", comment_dir.parent.name)
comment_id = fm.get("comment_id", comment_dir.name)
written: list[Path] = []
for filename, section_body in _split_sections(body):
if filename.lower() == "inline comment":
continue
sibling_name = _sibling_name(filename)
sibling = comment_dir / sibling_name
if sibling.is_file() and sibling.stat().st_size > 0:
continue
contents = _sibling_text_from_section(
filename, comment_id, docket_id, section_body
)
_write_sibling(comment_dir, sibling_name, contents)
written.append(sibling)
return written
def parse_combined(text: str) -> tuple[dict[str, Any], str]:
"""Split a combined.md into (frontmatter dict, body markdown)."""
if not text.startswith("---\n"):
@@ -95,17 +143,100 @@ def _attachment_paths(comment_dir: Path) -> list[Path]:
def _render_section(filename: str, result: ExtractResult) -> str:
return f"## {filename}\n\n{_section_body_for(result)}\n"
def _section_body_for(result: ExtractResult) -> str:
if result.status == "ok":
body = result.text.strip()
elif result.status == "ocr_needed":
body = f"_(ocr_needed — {result.chars} chars extracted)_"
elif result.status == "failed":
body = "_(failed — extraction error, see logs)_"
else: # unsupported
body = "_(unsupported file type)_"
return f"## {filename}\n\n{body}\n"
return result.text.strip()
if result.status == "ocr_needed":
return f"_(ocr_needed — {result.chars} chars extracted)_"
if result.status == "failed":
return "_(failed — extraction error, see logs)_"
return "_(unsupported file type)_"
def _emit(frontmatter: dict[str, Any], body: str) -> str:
yaml_text = yaml.safe_dump(frontmatter, sort_keys=False, default_flow_style=False)
return f"---\n{yaml_text}---\n\n{body}"
def _sibling_name(attachment_filename: str) -> str:
# Preserve the original extension as part of the sibling stem so
# `attachment_1.pdf` and `attachment_1.docx` (rare but possible in
# the same dir) don't collide on a shared `attachment_1.md`.
return f"{attachment_filename}.md"
def _sibling_contents(
attachment_filename: str,
comment_id: str,
docket_id: str,
result: ExtractResult,
) -> str:
fm = {
"comment_id": comment_id,
"docket_id": docket_id,
"source_file": attachment_filename,
"status": result.status,
"chars": result.chars,
"extracted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
}
return _emit(fm, _section_body_for(result) + "\n")
def _sibling_text_from_section(
attachment_filename: str,
comment_id: str,
docket_id: str,
section_body: str,
) -> str:
# Backfill path: we only have the rendered section body, not the
# original ExtractResult, so chars/status are best-effort.
body = section_body.strip()
status = "ok"
if body.startswith("_(failed"):
status = "failed"
elif body.startswith("_(ocr_needed"):
status = "ocr_needed"
elif body.startswith("_(unsupported"):
status = "unsupported"
fm = {
"comment_id": comment_id,
"docket_id": docket_id,
"source_file": attachment_filename,
"status": status,
"chars": len(body),
"extracted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
}
return _emit(fm, body + "\n")
def _write_sibling(comment_dir: Path, sibling_name: str, contents: str) -> None:
out = comment_dir / sibling_name
tmp = out.with_suffix(out.suffix + ".tmp")
tmp.write_text(contents)
os.replace(tmp, out)
def _split_sections(body: str) -> list[tuple[str, str]]:
"""Split a combined.md body into ``[(filename, section_body), ...]``.
Sections start with ``## <filename>`` lines. Anything before the
first ``## `` line is dropped.
"""
sections: list[tuple[str, str]] = []
current_name: str | None = None
current_lines: list[str] = []
for line in body.splitlines():
if line.startswith("## "):
if current_name is not None:
sections.append((current_name, "\n".join(current_lines).strip()))
current_name = line[3:].strip()
current_lines = []
else:
if current_name is not None:
current_lines.append(line)
if current_name is not None:
sections.append((current_name, "\n".join(current_lines).strip()))
return sections

View File

@@ -12,7 +12,7 @@ from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from rex.comments.combine import extract_comment
from rex.comments.combine import derive_siblings_from_combined, extract_comment
log = logging.getLogger(__name__)
@@ -27,7 +27,7 @@ def walk_and_extract(
limit: int | None = None,
workers: int = 8,
force: bool = False,
on_combined: Callable[[str, Path], None] | None = None,
on_extracted: Callable[[str, Path], None] | None = None,
) -> dict[str, int]:
"""Process every comment dir under *root*.
@@ -36,10 +36,14 @@ 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.
*on_extracted*, if given, is called on the main thread with
``(comment_id, comment_dir)`` for every dir that ends up with a
combined.md — both newly-written ones and pre-existing skips. The
callback is responsible for discovering whatever markdown files it
wants to consume in the dir (combined.md plus per-attachment
siblings). Skipped dirs that pre-date the sibling-write feature get
siblings derived from combined.md before the callback fires, so the
callback always sees a complete set.
Returns counts: ``{"written": N, "skipped": N, "failed": N}``.
"""
@@ -57,13 +61,17 @@ def walk_and_extract(
for fut in as_completed(futures):
status = fut.result()
stats[status] += 1
if on_combined and status == "written":
if on_extracted and status == "written":
cdir = futures[fut]
on_combined(cdir.name, cdir / _COMBINED)
on_extracted(cdir.name, cdir)
if on_combined:
if on_extracted:
for cdir in skipped_dirs:
on_combined(cdir.name, cdir / _COMBINED)
try:
derive_siblings_from_combined(cdir)
except Exception as e: # noqa: BLE001
log.warning("sibling backfill failed for %s: %s", cdir, e)
on_extracted(cdir.name, cdir)
return stats

View File

@@ -59,8 +59,9 @@ def test_stats_after_extract(tmp_path: Path):
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."""
def test_extract_attaches_combined_md_and_siblings_to_bib(tmp_path: Path, monkeypatch):
"""End-to-end: extract writes combined.md plus per-attachment sibling MDs,
and attaches all of them to the bib item. Idempotent on re-run."""
from bib import connect
from bib.item import Source
@@ -78,39 +79,27 @@ def test_extract_attaches_combined_md_to_bib(tmp_path: Path, monkeypatch):
item_key = store.upsert(item)
store.close()
result = runner.invoke(
app,
[
"extract",
"--root",
str(tmp_path),
"--workers",
"1",
],
)
result = runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"])
assert result.exit_code == 0, result.output
assert (cdir / "combined.md").is_file()
assert (cdir / "attachment_1.pdf.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,),
def md_attachments():
store = connect(str(bib_db))
rows = sorted(
r[0]
for r in store._con().execute(
"SELECT a.filename FROM attachments a "
"JOIN items i ON a.item_id=i.id "
"WHERE i.key=? AND a.filename LIKE '%.md'",
(item_key,),
)
)
)
store.close()
assert len(rows) == 1, "combined.md should be attached exactly once"
store.close()
return rows
# Re-running must not duplicate the attachment (idempotence)
assert md_attachments() == ["attachment_1.pdf.md", "combined.md"]
# Re-running must not duplicate attachments
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"
assert md_attachments() == ["attachment_1.pdf.md", "combined.md"]

View File

@@ -77,7 +77,7 @@ def test_walk_and_extract_respects_limit(tmp_path: Path):
assert stats["written"] == 1
def test_on_combined_called_for_written_dirs(tmp_path: Path):
def test_on_extracted_called_for_written_dirs(tmp_path: Path):
a, b = _setup_two_comments(tmp_path)
seen: list[tuple[str, Path]] = []
@@ -85,27 +85,38 @@ def test_on_combined_called_for_written_dirs(tmp_path: Path):
tmp_path,
inline_body_lookup=_stub_inline_body,
workers=2,
on_combined=lambda cid, p: seen.append((cid, p)),
on_extracted=lambda cid, d: seen.append((cid, d)),
)
assert sorted(cid for cid, _ in seen) == sorted([a.name, b.name])
assert all(p.name == "combined.md" for _, p in seen)
# callback gets the comment dir; sibling + combined.md are inside it
for _cid, d in seen:
assert (d / "combined.md").is_file()
assert (d / "attachment_1.pdf.md").is_file()
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."""
def test_on_extracted_called_for_skipped_dirs(tmp_path: Path):
"""Pre-existing combined.md still gets the callback, and any missing
sibling MDs are derived from it before the callback fires — so
previously-extracted dirs end up with the same set of MDs as
freshly-written ones."""
a, b = _setup_two_comments(tmp_path)
(a / "combined.md").write_text("---\n---\nstale\n")
# Pre-write a stale combined.md with one section so derive_siblings
# has something to split.
(a / "combined.md").write_text(
"---\ncomment_id: x\ndocket_id: y\n---\n\n## attachment_1.pdf\n\nstale body\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),
on_extracted=lambda cid, _d: seen.append(cid),
)
assert stats["written"] == 1
assert stats["skipped"] == 1
assert stats["written"] == 1 # b
assert stats["skipped"] == 1 # a
assert sorted(seen) == sorted([a.name, b.name])
# a's sibling was backfilled from its (stale) combined.md
assert (a / "attachment_1.pdf.md").is_file()