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:
@@ -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}")
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user