feat(bib): cfrlink context + placement over regulation items + cfr-jump CLI (closes #640)
This commit is contained in:
@@ -25,7 +25,8 @@ from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
from bib.frlink import JumpLink
|
||||
from bib.store import Store
|
||||
|
||||
_BASE = "https://www.ecfr.gov"
|
||||
|
||||
@@ -131,3 +132,69 @@ def cite_of(link: str) -> str:
|
||||
date=date or "",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
# ── Bib context + placement ─────────────────────────────────────────
|
||||
|
||||
|
||||
def item_for(cite: CfrCite, store: Store) -> str:
|
||||
"""The bib regulation item a cite belongs to — the exact-section
|
||||
item when the library has one, else the part item, else ``""``.
|
||||
URL construction never needs this; placement does."""
|
||||
sec_suffix = f"/title-{cite.title}/section-{cite.section}" if cite.section else ""
|
||||
part_suffixes = (
|
||||
f"/title-{cite.title}/part-{cite.part}",
|
||||
f"/part-{cite.part}",
|
||||
)
|
||||
part_hit = ""
|
||||
for item in store.list_items(item_type="regulation"):
|
||||
u = (item.url or "").split("#", 1)[0].rstrip("/")
|
||||
if "ecfr.gov" not in u:
|
||||
continue
|
||||
if (
|
||||
sec_suffix
|
||||
and u.endswith(f"section-{cite.section}")
|
||||
and f"title-{cite.title}/" in u
|
||||
):
|
||||
return item.key
|
||||
if not part_hit and u.endswith(part_suffixes[0]):
|
||||
part_hit = item.key
|
||||
elif (
|
||||
not part_hit
|
||||
and f"title-{cite.title}/" in u
|
||||
and u.endswith(part_suffixes[1])
|
||||
):
|
||||
part_hit = item.key
|
||||
return part_hit
|
||||
|
||||
|
||||
def place(store: Store, ref: str, *, label: str = "") -> "JumpLink":
|
||||
"""Resolve a CFR cite, attach it to its regulation item in
|
||||
``fr_links`` (page=0 marks non-FR rows), and return the link.
|
||||
Raises when the library has no matching regulation item — placed
|
||||
links need a Zotero parent."""
|
||||
from bib.frlink import JumpLink
|
||||
|
||||
cite = parse_cite(ref)
|
||||
key = item_for(cite, store)
|
||||
if not key:
|
||||
raise ValueError(
|
||||
f"{ref!r}: no bib regulation item covers {canonical(cite)} — "
|
||||
f"placement needs a library parent (part or section item)"
|
||||
)
|
||||
link_url = url(cite)
|
||||
con = store._con() # noqa: SLF001 — same pattern as bib.frlink.place
|
||||
con.execute(
|
||||
"INSERT OR IGNORE INTO fr_links (item_key, p_id, page, label, url) "
|
||||
"VALUES (?, NULL, 0, ?, ?)",
|
||||
(key, label or canonical(cite), link_url),
|
||||
)
|
||||
con.commit()
|
||||
return JumpLink(url=link_url, item_key=key, page=0)
|
||||
|
||||
|
||||
def md_link(ref: str, *, text: str = "") -> str:
|
||||
"""Markdown ``[text](url)`` for a CFR cite — store-free (the
|
||||
transform is deterministic)."""
|
||||
cite = parse_cite(ref)
|
||||
return f"[{text or canonical(cite)}]({url(cite)})"
|
||||
|
||||
@@ -727,3 +727,38 @@ def fr_jump(
|
||||
typer.echo(frlink.md_link(ref, store=store, item_key=key) if md else link.url)
|
||||
if link.snippet:
|
||||
typer.echo(f' ¶ "{link.snippet[:100]}..."')
|
||||
|
||||
|
||||
@app.command(name="cfr-jump")
|
||||
def cfr_jump(
|
||||
ref: str = typer.Argument(
|
||||
"",
|
||||
help='CFR cite: "42 CFR 425.400(c)(1)(iv)" or "42 CFR Part 425".',
|
||||
),
|
||||
from_url: str = typer.Option(
|
||||
"", "--url", help="Reverse mode: print the cite for an eCFR URL."
|
||||
),
|
||||
md: bool = typer.Option(False, "--md", help="Print a markdown link."),
|
||||
do_place: bool = typer.Option(
|
||||
False,
|
||||
"--place",
|
||||
help="Attach the link to its bib regulation item (fr_links) so the "
|
||||
"next zotero sync carries it as a child link attachment.",
|
||||
),
|
||||
label: str = typer.Option("", "--label", help="Label for the placed link."),
|
||||
) -> None:
|
||||
"""Bidirectional CFR cite ↔ eCFR URL (#640); optionally place the link."""
|
||||
from bib import cfrlink, connect
|
||||
|
||||
if bool(ref) == bool(from_url):
|
||||
raise typer.BadParameter("pass exactly one of REF or --url")
|
||||
if from_url:
|
||||
typer.echo(cfrlink.cite_of(from_url))
|
||||
return
|
||||
if do_place:
|
||||
link = cfrlink.place(connect(), ref, label=label)
|
||||
typer.echo(f"placed on {link.item_key}: {link.url}")
|
||||
elif md:
|
||||
typer.echo(cfrlink.md_link(ref))
|
||||
else:
|
||||
typer.echo(cfrlink.url(cfrlink.parse_cite(ref)))
|
||||
|
||||
@@ -128,3 +128,88 @@ class TestRoundTrip:
|
||||
@pytest.mark.parametrize("u", [SEC_URL, FRAG_URL, PART_URL])
|
||||
def test_url_cite_url_is_short_form(self, u: str) -> None:
|
||||
assert cfrlink.url(cfrlink.parse_cite(cfrlink.cite_of(u))) == u
|
||||
|
||||
|
||||
# ── item_for / place (#640) ─────────────────────────────────────────
|
||||
|
||||
|
||||
def _reg_store():
|
||||
from bib.item import Regulation
|
||||
from bib.store import Store
|
||||
|
||||
s = Store(":memory:")
|
||||
part_key = s.create(
|
||||
Regulation(
|
||||
title="42 CFR Part 425",
|
||||
url="https://www.ecfr.gov/current/title-42/part-425",
|
||||
)
|
||||
)
|
||||
sec_key = s.create(
|
||||
Regulation(
|
||||
title="42 CFR Part 425 § 425.400",
|
||||
url="https://www.ecfr.gov/current/title-42/part-425/section-425.400",
|
||||
)
|
||||
)
|
||||
s.create(
|
||||
Regulation(
|
||||
title="42 CFR Part 414",
|
||||
url="https://www.ecfr.gov/current/title-42/part-414",
|
||||
)
|
||||
)
|
||||
return s, part_key, sec_key
|
||||
|
||||
|
||||
class TestItemFor:
|
||||
def test_exact_section_beats_part(self) -> None:
|
||||
s, _part, sec = _reg_store()
|
||||
assert cfrlink.item_for(cfrlink.parse_cite("42 CFR 425.400(c)"), s) == sec
|
||||
s.close()
|
||||
|
||||
def test_part_fallback(self) -> None:
|
||||
s, part, _sec = _reg_store()
|
||||
assert cfrlink.item_for(cfrlink.parse_cite("42 CFR 425.402"), s) == part
|
||||
s.close()
|
||||
|
||||
def test_miss_returns_empty(self) -> None:
|
||||
s, _part, _sec = _reg_store()
|
||||
assert cfrlink.item_for(cfrlink.parse_cite("45 CFR 155.20"), s) == ""
|
||||
s.close()
|
||||
|
||||
|
||||
class TestCfrPlace:
|
||||
def test_place_on_section_item(self) -> None:
|
||||
s, _part, sec = _reg_store()
|
||||
link = cfrlink.place(s, "42 CFR 425.400(c)(1)(iv)", label="attribution ¶")
|
||||
assert link.item_key == sec
|
||||
con = s._con() # noqa: SLF001
|
||||
row = con.execute(
|
||||
"SELECT item_key, p_id, page, label, url FROM fr_links"
|
||||
).fetchone()
|
||||
assert tuple(row) == (
|
||||
sec,
|
||||
None,
|
||||
0,
|
||||
"attribution ¶",
|
||||
FRAG_URL,
|
||||
)
|
||||
cfrlink.place(s, "42 CFR 425.400(c)(1)(iv)") # same URL → no dup
|
||||
n = con.execute("SELECT count(*) FROM fr_links").fetchone()[0]
|
||||
assert n == 1
|
||||
s.close()
|
||||
|
||||
def test_place_without_matching_item_raises(self) -> None:
|
||||
s, _part, _sec = _reg_store()
|
||||
with pytest.raises(ValueError, match="regulation item"):
|
||||
cfrlink.place(s, "45 CFR 155.20")
|
||||
s.close()
|
||||
|
||||
|
||||
class TestMdLink:
|
||||
def test_store_free_markdown(self) -> None:
|
||||
assert (
|
||||
cfrlink.md_link("42 CFR 425.400(c)(1)(iv)")
|
||||
== f"[42 CFR 425.400(c)(1)(iv)]({FRAG_URL})"
|
||||
)
|
||||
assert cfrlink.md_link("42 CFR Part 425", text="the MSSP part") == (
|
||||
f"[the MSSP part]({PART_URL})"
|
||||
)
|
||||
|
||||
@@ -621,3 +621,40 @@ class TestSyncFrLinks:
|
||||
assert n == 1
|
||||
con.close()
|
||||
store.close()
|
||||
|
||||
|
||||
class TestSyncCfrLinks:
|
||||
def test_cfr_placed_row_syncs_as_link_child(self, tmp_path) -> None:
|
||||
"""#640: a CFR-placed fr_links row (page=0, p_id NULL) rides the
|
||||
same linkMode=3 path as FR links."""
|
||||
from bib import cfrlink
|
||||
from bib.store import Store
|
||||
|
||||
db_path = str(tmp_path / "zotero.sqlite")
|
||||
_make_zotero_db(db_path).close()
|
||||
|
||||
store = Store(str(tmp_path / "bib.sqlite"))
|
||||
key = store.create(
|
||||
Regulation(
|
||||
key="REGCFR42",
|
||||
title="42 CFR Part 425",
|
||||
url="https://www.ecfr.gov/current/title-42/part-425",
|
||||
)
|
||||
)
|
||||
cfrlink.place(store, "42 CFR 425.402", label="assignment methodology")
|
||||
|
||||
stats = push_to_zotero([store.get(key)], store=store, zotero_db=db_path)
|
||||
assert stats["fr_links"] == 1
|
||||
|
||||
con = sqlite3.connect(db_path)
|
||||
con.row_factory = sqlite3.Row
|
||||
url = con.execute(
|
||||
"""SELECT idv.value FROM itemAttachments ia
|
||||
JOIN itemData d ON d.itemID = ia.itemID
|
||||
JOIN itemDataValues idv ON idv.valueID = d.valueID
|
||||
JOIN fields f ON f.fieldID = d.fieldID
|
||||
WHERE ia.linkMode = 3 AND f.fieldName = 'url'"""
|
||||
).fetchone()[0]
|
||||
assert url == "https://www.ecfr.gov/current/title-42/section-425.402"
|
||||
con.close()
|
||||
store.close()
|
||||
|
||||
Reference in New Issue
Block a user