feat(comments): palliative-mention sweep — topic:palliative tags + extra_json snippets (refs #660)

dev/scripts/tag_palliative_comments.py

@@ -0,0 +1,200 @@
+"""Tag PFS-rulemaking comments that reference palliative care (#660).
+
+Sweeps every regulations.gov comment item in the bib store. Per item the
+text is the extracted body (``.state/comments/<docket>/<comment_id>/combined.md``)
+when present, else the enriched abstract. Items matching the ``palliativ``
+stem get:
+
+- tag ``topic:palliative``
+- ``extra_json`` keys ``palliative_mentions`` / ``palliative_snippets``
+
+Per-docket sweep stats (total / with_text / hits) land in ``bib_meta``
+under key ``palliative_sweep`` so the notebook can disclose denominators
+without access to ``.state``. Idempotent: re-running refreshes extras and
+removes the tag from items that no longer match.
+
+Usage:
+    uv run python dev/scripts/tag_palliative_comments.py [--dry-run]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from collections import defaultdict
+from datetime import datetime, timezone
+from pathlib import Path
+
+_PATTERN = re.compile(r"palliativ\w*", re.IGNORECASE)
+_COMMENT_URL_PREFIX = "https://www.regulations.gov/comment/"
+_TAG = "topic:palliative"
+_META_KEY = "palliative_sweep"
+_SNIPPET_MAX_CHARS = 320
+
+
+def find_mentions(text: str, max_snippets: int = 3) -> tuple[int, list[str]]:
+    """Count ``palliativ*`` matches and pull up to *max_snippets* sentences."""
+    matches = list(_PATTERN.finditer(text))
+    if not matches:
+        return 0, []
+    snippets: list[str] = []
+    seen_spans: list[tuple[int, int]] = []
+    for m in matches:
+        if len(snippets) >= max_snippets:
+            break
+        if any(s <= m.start() < e for s, e in seen_spans):
+            continue  # sentence already captured
+        # sentence-ish window: back to the previous ./!/?/newline, forward
+        # to the next
+        start = max(
+            text.rfind(". ", 0, m.start()),
+            text.rfind("! ", 0, m.start()),
+            text.rfind("? ", 0, m.start()),
+            text.rfind("\n", 0, m.start()),
+        )
+        start = start + 2 if start > 0 else 0
+        ends = [
+            text.find(". ", m.end()),
+            text.find("! ", m.end()),
+            text.find("? ", m.end()),
+            text.find("\n", m.end()),
+        ]
+        ends = [e for e in ends if e != -1]
+        end = min(ends) + 1 if ends else len(text)
+        seen_spans.append((start, end))
+        snippet = " ".join(text[start:end].split())
+        if len(snippet) > _SNIPPET_MAX_CHARS:
+            cut = snippet.lower().find("palliativ")
+            lo = max(0, cut - _SNIPPET_MAX_CHARS // 2)
+            snippet = (
+                ("…" if lo else "") + snippet[lo : lo + _SNIPPET_MAX_CHARS - 2] + "…"
+            )
+        snippets.append(snippet)
+    return len(matches), snippets
+
+
+def _comment_rows(store) -> list[tuple[str, str, str, str]]:
+    """(key, url, docket, comment_id) for every comment item."""
+    rows = (
+        store._con()
+        .execute(
+            "SELECT key, url FROM items WHERE url LIKE ?",
+            (f"{_COMMENT_URL_PREFIX}%",),
+        )
+        .fetchall()
+    )
+    out = []
+    for key, url in rows:
+        comment_id = url[len(_COMMENT_URL_PREFIX) :]
+        docket = comment_id.rsplit("-", 1)[0]
+        out.append((key, url, docket, comment_id))
+    return out
+
+
+def _load_extra(store, key: str) -> dict:
+    row = (
+        store._con()
+        .execute("SELECT extra_json FROM items WHERE key = ?", (key,))
+        .fetchone()
+    )
+    return json.loads(row[0]) if row and row[0] else {}
+
+
+def _merge_extra(store, key: str, mentions: int, snippets: list[str]) -> None:
+    extra = _load_extra(store, key)
+    extra["palliative_mentions"] = mentions
+    extra["palliative_snippets"] = snippets
+    store.update(key, extra_json=json.dumps(extra))
+
+
+def _clear_extra(store, key: str) -> None:
+    extra = _load_extra(store, key)
+    if "palliative_mentions" in extra or "palliative_snippets" in extra:
+        extra.pop("palliative_mentions", None)
+        extra.pop("palliative_snippets", None)
+        store.update(key, extra_json=json.dumps(extra))
+
+
+def sweep(store, root: Path, dry_run: bool = False) -> dict:
+    from rex.comments.combine import parse_combined
+
+    tagged_before = {
+        r[0]
+        for r in store._con().execute(
+            "SELECT i.key FROM items i JOIN item_tags it ON it.item_id = i.id "
+            "JOIN tags t ON t.id = it.tag_id WHERE t.name = ?",
+            (_TAG,),
+        )
+    }
+    stats: dict[str, dict[str, int]] = defaultdict(
+        lambda: {"total": 0, "with_text": 0, "hits": 0}
+    )
+    hits: set[str] = set()
+    for key, _url, docket, comment_id in _comment_rows(store):
+        stats[docket]["total"] += 1
+        combined = root / docket / comment_id / "combined.md"
+        text = ""
+        if combined.exists():
+            _, body = parse_combined(combined.read_text())
+            text = body
+        if not text.strip():
+            text = store.get(key).abstract or ""
+        if not text.strip():
+            continue
+        stats[docket]["with_text"] += 1
+        count, snippets = find_mentions(text)
+        if not count:
+            continue
+        stats[docket]["hits"] += 1
+        hits.add(key)
+        if not dry_run:
+            store.add_tag(key, _TAG)
+            _merge_extra(store, key, count, snippets)
+
+    stale = tagged_before - hits
+    if not dry_run:
+        for key in stale:
+            store.remove_tag(key, _TAG)
+            _clear_extra(store, key)
+        meta = {
+            "dockets": dict(stats),
+            "swept_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        }
+        store._con().execute(
+            "INSERT INTO bib_meta (key, value) VALUES (?, ?) "
+            "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+            (_META_KEY, json.dumps(meta)),
+        )
+        store._con().commit()
+    return {"stats": dict(stats), "hits": len(hits), "untagged": len(stale)}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", default=Path(".state/comments"), type=Path)
+    parser.add_argument("--db", default="", help="bib.sqlite path (default: cfg)")
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    if args.db:
+        from bib.store import Store
+
+        store = Store(args.db)
+    else:
+        from conf.connect import bib
+
+        store = bib()
+    result = sweep(store, args.root, dry_run=args.dry_run)
+    for docket in sorted(result["stats"]):
+        s = result["stats"][docket]
+        print(
+            f"  {docket}: {s['hits']} hits / {s['with_text']} with text "
+            f"/ {s['total']} total"
+        )
+    print(f"tagged={result['hits']} untagged_stale={result['untagged']}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/dev/test_tag_palliative_comments.py

@@ -0,0 +1,53 @@
+"""Tests for dev/scripts/tag_palliative_comments.py — mention extraction."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+_SCRIPT = (
+    Path(__file__).resolve().parents[2]
+    / "dev"
+    / "scripts"
+    / "tag_palliative_comments.py"
+)
+_spec = importlib.util.spec_from_file_location("_tag_palliative", _SCRIPT)
+assert _spec and _spec.loader
+tpc = importlib.util.module_from_spec(_spec)
+sys.modules["_tag_palliative"] = tpc
+_spec.loader.exec_module(tpc)
+
+
+class TestFindMentions:
+    def test_no_match(self):
+        assert tpc.find_mentions("Medicare payment policy comment.") == (0, [])
+
+    def test_counts_and_case(self):
+        text = (
+            "We support Palliative care. Our palliative program grew. "
+            "Palliative-trained nurses matter."
+        )
+        count, snippets = tpc.find_mentions(text)
+        assert count == 3
+        assert len(snippets) >= 1
+        assert all("alliativ" in s for s in snippets)
+
+    def test_snippets_are_sentences_with_context(self):
+        text = (
+            "Intro sentence. CMS should pay for palliative consultations "
+            "in the home. Closing sentence."
+        )
+        _, snippets = tpc.find_mentions(text)
+        assert snippets == ["CMS should pay for palliative consultations in the home."]
+
+    def test_snippet_cap(self):
+        text = " ".join(f"Point {i} on palliative care." for i in range(10))
+        count, snippets = tpc.find_mentions(text, max_snippets=3)
+        assert count == 10
+        assert len(snippets) == 3
+
+    def test_long_sentence_trimmed(self):
+        text = "x" * 400 + " palliative " + "y" * 400
+        _, snippets = tpc.find_mentions(text)
+        assert snippets and len(snippets[0]) <= 320