homelab/stack
1
0
Fork 0
Code Issues 169 Pull Requests 1 Actions Packages Projects 1 Releases 1 Wiki Activity

fix(bib): fr-grab --force keeps pincites, pin tags and note links

Browse Source 
Replacing an anchor map can renumber every paragraph: the CY2027 PFS
final rule's public-inspection capture diverged from the printed
issue's ids at p-5, and a plain re-grab would have pointed 113
pincites, 99 pin tags and 128 note links at the wrong text. grab now
maps each referenced paragraph to the new one holding its text (word
4-gram containment, repeated text placed by its nearest neighbour),
rewrites pincites (merging collapses), pin: tags, fr_links and note
links with their URL and FR page cite in the anchor-swap transaction,
and refuses before writing when a reference has no confident home.
fr-grab --map p-OLD=p-NEW places the rest. On the real CY2027 remap it
reproduces all 99 hand-verified ids with three overrides.
This commit is contained in:
kert
2026-11-11 13:18:01 -05:00
parent 1c3f83e6d5
commit 254f16bc81
3 changed files with 411 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
src/bib/frlink.py
View File

@@ -129,6 +129,213 @@ def parse_anchors(html: str) -> list[Anchor]:
return anchors
# ── Remap references when an anchor map is replaced ─────────────────
class UnresolvedAnchors(ValueError):
"""A pinned paragraph has no confident home in the new anchor map."""
# A note link to a paragraph, with an optional page cite after it:
# <a href="...#p-56">p-56</a> (PI PDF p. 7) or ... (91 FR 70993)
_NOTE_LINK_RE = re.compile(
r'(href=")([^"#]*)#p-(\d+)(">)p-(\d+)(</a>)'
r"(\s*\((?:PI PDF p\. \d+|\d+ FR \d+)\))?"
)
def _match_words(text: str) -> list[str]:
# The official HTML carries inline "( printed page 71104)" markers and
# footnote refs "[40]" that a public-inspection capture does not.
text = re.sub(r"\(\s*printed page \d+\s*\)", " ", text)
text = re.sub(r"\[\d+\]", " ", text)
return re.sub(r"[^a-z0-9]+", " ", text.lower()).split()
def _shingles(text: str, k: int = 4) -> set[tuple[str, ...]]:
w = _match_words(text)
k = min(k, max(1, len(w)))
return {tuple(w[i : i + k]) for i in range(max(1, len(w) - k + 1))}
def map_paragraphs(
old: dict[int, str],
new: list[Anchor],
*,
overrides: dict[int, int] | None = None,
min_containment: float = 0.95,
) -> dict[int, int]:
"""Map each old paragraph id to the new id holding its text.
An old paragraph maps where at least ``min_containment`` of its word
4-grams appear in one new paragraph (so a page-break fragment finds
the whole paragraph). Text CMS repeats (amendment lists restated in
the final summary) has several homes; the one nearest the position
its closest unambiguous neighbour predicts wins. ``overrides`` place
what text cannot (a heading, a merged bullet). Raises
:class:`UnresolvedAnchors` naming every id with no home.
"""
overrides = dict(overrides or {})
new_ids = {a.p_id for a in new}
bad = sorted(v for v in overrides.values() if v not in new_ids)
if bad:
raise UnresolvedAnchors(f"overrides point at missing paragraphs: {bad}")
sh_new = [(a.p_id, _shingles(a.text)) for a in new]
index: dict[tuple[str, ...], list[int]] = {}
for i, (_, sh) in enumerate(sh_new):
for g in sh:
index.setdefault(g, []).append(i)
mapping = {p: overrides[p] for p in old if p in overrides}
ambiguous: dict[int, list[int]] = {}
unresolved: list[str] = []
for p in sorted(set(old) - set(overrides)):
sh = _shingles(old[p])
hits: dict[int, int] = {}
for g in sh:
for i in index.get(g, ()):
hits[i] = hits.get(i, 0) + 1
cands = [
sh_new[i][0] for i, h in hits.items() if h / len(sh) >= min_containment
]
if len(cands) == 1:
mapping[p] = cands[0]
elif cands:
ambiguous[p] = cands
else:
best = max(hits.values(), default=0) / len(sh)
unresolved.append(f"p-{p} (best {best:.2f}: {old[p][:60]!r})")
if unresolved:
raise UnresolvedAnchors(
"no confident home in the new anchor map for "
+ "; ".join(unresolved)
+ " — pass overrides={old: new}"
)
for p, cands in ambiguous.items():
if not mapping:
raise UnresolvedAnchors(
f"p-{p}: repeated text {sorted(cands)}, no anchor to place it by"
)
near = min(mapping, key=lambda q: abs(q - p))
predicted = mapping[near] + (p - near)
mapping[p] = min(cands, key=lambda c: abs(c - predicted))
return mapping
def _referenced_ids(con: Any, item_key: str) -> set[int]:
"""Paragraph ids that pincites, pin: tags, fr_links and the item's
notes point at."""
ids = {
int(r[0][2:])
for r in con.execute(
"SELECT locator FROM pincites WHERE item_key = ? AND locator LIKE 'p-%'",
(item_key,),
)
}
ids |= {
int(r[0].rsplit("/p-", 1)[1])
for r in con.execute(
"SELECT name FROM tags WHERE name LIKE ?", (f"pin:{item_key}/p-%",)
)
if r[0].rsplit("/p-", 1)[1].isdigit()
}
ids |= {
r[0]
for r in con.execute(
"SELECT p_id FROM fr_links WHERE item_key = ? AND p_id IS NOT NULL",
(item_key,),
)
}
for (content,) in con.execute(
"SELECT n.content FROM notes n JOIN items i ON i.id = n.item_id WHERE i.key = ?",
(item_key,),
):
ids |= {int(m.group(3)) for m in _NOTE_LINK_RE.finditer(content or "")}
return ids
def _rewrite_references(
con: Any,
item_key: str,
mapping: dict[int, int],
page_of: dict[int, int],
*,
old_url: str,
new_url: str,
volume: int,
) -> None:
"""Point every reference at its new paragraph (inside the caller's
transaction). Pincites that collapse onto one paragraph merge."""
rows = con.execute(
"SELECT id, fn_path, locator, note FROM pincites "
"WHERE item_key = ? AND locator LIKE 'p-%' ORDER BY id",
(item_key,),
).fetchall()
groups: dict[tuple[str, str], list[tuple[int, str]]] = {}
for pid, fn, loc, note in rows:
new_loc = f"p-{mapping.get(int(loc[2:]), int(loc[2:]))}"
groups.setdefault((fn, new_loc), []).append((pid, note))
for pid, *_ in rows: # free the unique (fn_path, item_key, locator) slots
con.execute("UPDATE pincites SET locator = ? WHERE id = ?", (f"~{pid}", pid))
for (_, loc), members in groups.items():
keep, *drop = (pid for pid, _ in members)
note = " ".join(dict.fromkeys(n for _, n in members if n))
con.execute(
"UPDATE pincites SET locator = ?, note = ? WHERE id = ?", (loc, note, keep)
)
for pid in drop:
con.execute("DELETE FROM pincites WHERE id = ?", (pid,))
for tag_id, name in con.execute(
"SELECT id, name FROM tags WHERE name LIKE ?", (f"pin:{item_key}/p-%",)
).fetchall():
old_p = name.rsplit("/p-", 1)[1]
if not old_p.isdigit() or int(old_p) not in mapping:
continue
new_name = f"pin:{item_key}/p-{mapping[int(old_p)]}"
if new_name == name:
continue
con.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (new_name,))
new_id = con.execute(
"SELECT id FROM tags WHERE name = ?", (new_name,)
).fetchone()[0]
con.execute(
"INSERT OR IGNORE INTO item_tags (item_id, tag_id) "
"SELECT item_id, ? FROM item_tags WHERE tag_id = ?",
(new_id, tag_id),
)
con.execute("DELETE FROM item_tags WHERE tag_id = ?", (tag_id,))
def link(m: re.Match) -> str:
n = mapping.get(int(m.group(3)), int(m.group(3)))
base = new_url if m.group(2) == old_url else m.group(2)
cite = (
f" ({volume} FR {page_of[n]})"
if m.group(7) and n in page_of
else (m.group(7) or "")
)
return f"{m.group(1)}{base}#p-{n}{m.group(4)}p-{n}{m.group(6)}{cite}"
for nid, content in con.execute(
"SELECT n.id, n.content FROM notes n JOIN items i ON i.id = n.item_id WHERE i.key = ?",
(item_key,),
).fetchall():
out = _NOTE_LINK_RE.sub(link, content or "")
if out != content:
con.execute("UPDATE notes SET content = ? WHERE id = ?", (out, nid))
for lid, p, url in con.execute(
"SELECT id, p_id, url FROM fr_links WHERE item_key = ? AND p_id IS NOT NULL",
(item_key,),
).fetchall():
n = mapping.get(p, p)
new_link = re.sub(r"#p-\d+$", f"#p-{n}", url.replace(old_url, new_url, 1))
con.execute(
"UPDATE fr_links SET p_id = ?, page = ?, url = ? WHERE id = ?",
(n, page_of.get(n, 0), new_link, lid),
)
# ── Grab ────────────────────────────────────────────────────────────
@@ -164,6 +371,7 @@ def grab(
*,
force: bool = False,
fetch: Callable[[str], tuple[dict, str]] | None = None,
overrides: dict[int, int] | None = None,
) -> dict[str, Any]:
"""Grab and persist ``item_key``'s FR anchor map.
@@ -171,6 +379,13 @@ def grab(
(injectable for tests). Replaces the item's ``fr_anchors``
partition and ``fr_anchor_docs`` row. Skips (returns
``{"skipped": True}``) when already grabbed unless ``force``.
A replaced map can renumber paragraphs (a public-inspection capture
vs the printed issue), so every pincite, ``pin:`` tag, ``fr_links``
row and note link on the item is moved to the paragraph holding the
same text (:func:`map_paragraphs`; ``overrides`` place the rest) in
the same transaction. If any reference has no confident home,
:class:`UnresolvedAnchors` is raised before anything is written.
"""
con = store._con() # noqa: SLF001 — same pattern as bib/oig.py
existing = con.execute(
@@ -184,6 +399,25 @@ def grab(
anchors = parse_anchors(body)
pages = {a.page for a in anchors}
mapping: dict[int, int] = {}
old_url = ""
if existing:
refs = _referenced_ids(con, item_key)
old_text = {
r[0]: r[1]
for r in con.execute(
"SELECT p_id, text FROM fr_anchors WHERE item_key = ?", (item_key,)
)
}
refs &= set(old_text)
if refs:
mapping = map_paragraphs(
{p: old_text[p] for p in refs}, anchors, overrides=overrides
)
old_url = con.execute(
"SELECT html_url FROM fr_anchor_docs WHERE item_key = ?", (item_key,)
).fetchone()[0]
con.execute("DELETE FROM fr_anchors WHERE item_key = ?", (item_key,))
con.executemany(
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) "
@@ -208,6 +442,16 @@ def grab(
len(pages),
),
)
if mapping or (old_url and old_url != meta["html_url"]):
_rewrite_references(
con,
item_key,
mapping,
{a.p_id: a.page for a in anchors},
old_url=old_url,
new_url=meta["html_url"],
volume=int(meta["volume"]),
)
con.commit()
return {
"item_key": item_key,
@@ -215,6 +459,7 @@ def grab(
"n_paragraphs": len(anchors),
"n_pages": len(pages),
"skipped": False,
"remapped": {k: v for k, v in mapping.items() if k != v},
}

28
src/cli/bib.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import re
from collections.abc import Callable
from pathlib import Path
@@ -1126,20 +1127,41 @@ def fr_grab(
force: bool = typer.Option(
False, "--force", help="Re-grab even if an anchor map already exists."
),
remap: list[str] = typer.Option(
[],
"--map",
help="Place a pinned paragraph text matching cannot: p-OLD=p-NEW "
"(repeatable; --key only).",
),
) -> None:
"""Grab federalregister.gov paragraph/page anchor maps into bib (#634).
One body-HTML fetch per rule; anchors persist in ``fr_anchors`` /
``fr_anchor_docs`` for offline jump-link resolution.
``fr_anchor_docs`` for offline jump-link resolution. A --force
re-grab moves the item's pincites, pin: tags, fr_links and note
links to the paragraphs holding the same text, and refuses (writing
nothing) when one has no confident home — add --map for those.
"""
from bib import connect, frlink
store = connect()
if bool(key) == all_rules:
raise typer.BadParameter("pass exactly one of --key or --all")
overrides: dict[int, int] = {}
for spec in remap:
m = re.fullmatch(r"p-(\d+)=p-(\d+)", spec.strip())
if not m:
raise typer.BadParameter(f"--map wants p-OLD=p-NEW, got {spec!r}")
overrides[int(m.group(1))] = int(m.group(2))
if key:
results = [frlink.grab(store, key, force=force)]
try:
results = [frlink.grab(store, key, force=force, overrides=overrides)]
except frlink.UnresolvedAnchors as e:
typer.echo(f" {key}: refused, nothing written — {e}", err=True)
raise typer.Exit(1) from e
else:
if overrides:
raise typer.BadParameter("--map needs --key")
results = frlink.backfill(store, force=force)
for r in results:
if r.get("skipped"):
@@ -1147,9 +1169,11 @@ def fr_grab(
elif "error" in r:
typer.echo(f" {r['item_key']}: ERROR {r['error']}")
else:
moved = r.get("remapped") or {}
typer.echo(
f" {r['item_key']}: {r['document_number']} — "
f"{r['n_paragraphs']} paragraphs / {r['n_pages']} pages"
+ (f"; {len(moved)} pinned paragraph ids moved" if moved else "")
)

140
tests/bib/test_frlink.py
View File

@@ -447,3 +447,143 @@ class TestRuleKind:
frlink.rule_kind(s, "NOPE0000", title="CY 2005 PFS Final Rule") == "final"
)
assert frlink.rule_kind(s, key, title="") == "proposed"
# ── Force regrab keeps references (PI capture → official anchors) ───
_A = "Budget neutrality requires that increases or decreases in relative value units stay within limits."
_B = "The qualifying conversion factor receives a three quarters of a percent update each year."
_C = "Home or residence visits count toward assignment with or without the new complexity modifiers."
_D = "We received no comments on this section of the proposal and finalize it as proposed today."
PI_HTML = f"""<html><body>
<p id="p-1" data-page="7">{_A}</p>
<p id="p-2" data-page="7">{_B}</p>
<p id="p-3" data-page="8">{_C}</p>
<p id="p-4" data-page="9">{_D}</p>
</body></html>"""
# The official issue adds a front-matter paragraph, so every id shifts.
OFFICIAL_HTML = f"""<html><body>
<p id="p-1" data-page="70992">Billing Code 4120-01-P front matter for this document.</p>
<p id="p-2" data-page="70992">{_A}</p>
<p id="p-3" data-page="70993">{_B}</p>
<p id="p-4" data-page="70994">{_C} ( printed page 70995) [40]</p>
<p id="p-5" data-page="70995">{_D}</p>
</body></html>"""
PI_META = {**DOC_META, "html_url": "https://example.test/public-inspection/doc"}
OFFICIAL_META = {
**DOC_META,
"html_url": "https://example.test/documents/doc",
"start_page": 70992,
"end_page": 70995,
}
def _pinned_store() -> tuple[Store, str]:
s, key = _store_with_rule()
frlink.grab(s, key, fetch=lambda _d: (PI_META, PI_HTML))
s.upsert_pincite(f"notes.{key}.x", key, "p-3", "fr_anchor", "home visits count")
s.add_tag(key, f"pin:{key}/p-3")
item_id = s._con().execute("SELECT id FROM items WHERE key=?", (key,)).fetchone()[0] # noqa: SLF001
s._con().execute( # noqa: SLF001
"INSERT INTO notes (item_id, title, content) VALUES (?, 'n', ?)",
(
item_id,
'<p>See <a href="https://example.test/public-inspection/doc#p-3">p-3</a> (PI PDF p. 8).</p>',
),
)
s._con().commit() # noqa: SLF001
return s, key
class TestRegrabKeepsReferences:
def test_pincite_tag_and_note_follow_the_paragraph_text(self) -> None:
s, key = _pinned_store()
out = frlink.grab(
s, key, force=True, fetch=lambda _d: (OFFICIAL_META, OFFICIAL_HTML)
)
assert out["remapped"] == {3: 4}
con = s._con() # noqa: SLF001
assert [
r[0]
for r in con.execute(
"SELECT locator FROM pincites WHERE item_key=?", (key,)
)
] == ["p-4"]
assert (
f"pin:{key}/p-4" in s.get(key).tags
and f"pin:{key}/p-3" not in s.get(key).tags
)
note = con.execute("SELECT content FROM notes").fetchone()[0]
assert (
'href="https://example.test/documents/doc#p-4">p-4</a> (91 FR 70994)'
in note
)
assert "public-inspection" not in note and "PI PDF" not in note
def test_unplaceable_reference_refuses_and_writes_nothing(self) -> None:
s, key = _pinned_store()
moved = OFFICIAL_HTML.replace(
_C, "An entirely different paragraph about something else here today."
)
with pytest.raises(frlink.UnresolvedAnchors, match="p-3"):
frlink.grab(s, key, force=True, fetch=lambda _d: (OFFICIAL_META, moved))
con = s._con() # noqa: SLF001
assert con.execute("SELECT locator FROM pincites").fetchone()[0] == "p-3"
assert (
con.execute("SELECT html_url FROM fr_anchor_docs").fetchone()[0]
== PI_META["html_url"]
)
def test_override_places_what_text_cannot(self) -> None:
s, key = _pinned_store()
moved = OFFICIAL_HTML.replace(
_C, "An entirely different paragraph about something else here today."
)
out = frlink.grab(
s,
key,
force=True,
fetch=lambda _d: (OFFICIAL_META, moved),
overrides={3: 5},
)
assert out["remapped"] == {3: 5}
assert s._con().execute("SELECT locator FROM pincites").fetchone()[0] == "p-5" # noqa: SLF001
class TestFrGrabCli:
def test_refusal_exits_nonzero_and_names_the_paragraph(self, monkeypatch) -> None:
from typer.testing import CliRunner
from cli.bib import app
s, key = _pinned_store()
moved = OFFICIAL_HTML.replace(
_C, "An entirely different paragraph about something else here today."
)
real = frlink.grab
monkeypatch.setattr("bib.connect", lambda: s)
monkeypatch.setattr(
frlink,
"grab",
lambda st, k, **kw: real(
st, k, fetch=lambda _d: (OFFICIAL_META, moved), **kw
),
)
res = CliRunner().invoke(app, ["fr-grab", "--key", key, "--force"])
assert res.exit_code == 1 and "p-3" in res.output
res = CliRunner().invoke(
app, ["fr-grab", "--key", key, "--force", "--map", "p-3=p-5"]
)
assert res.exit_code == 0, res.output
assert "1 pinned paragraph ids moved" in res.output
def test_bad_map_spec_is_rejected(self) -> None:
from typer.testing import CliRunner
from cli.bib import app
res = CliRunner().invoke(app, ["fr-grab", "--key", "X", "--map", "3=5"])
assert res.exit_code != 0