Merge pull request 'feat: pydoc pincite — structured Zotero citations in docstrings' (#293) from feat/pydoc-pincite into main

dev/scripts/fetch_fr_attachments.py

@@ -0,0 +1,185 @@
+"""Fetch Federal Register PDF and TXT for all rules in bib.sqlite.
+
+For each rule item with a ``document_number``, queries the Federal
+Register API for the PDF and raw text URLs, downloads both, and
+attaches them to the item in the bib store.
+
+Usage::
+
+    uv run python dev/scripts/fetch_fr_attachments.py            # fetch all
+    uv run python dev/scripts/fetch_fr_attachments.py --dry-run   # preview
+    uv run python dev/scripts/fetch_fr_attachments.py --key KEY   # single item
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+import time
+from pathlib import Path
+
+import httpx
+
+from conf import connect
+
+FR_API = "https://www.federalregister.gov/api/v1/documents"
+
+
+def _extract_doc_number(url: str) -> str | None:
+    """Extract FR document number from a federalregister.gov URL."""
+    m = re.search(r"/documents/\d{4}/\d{2}/\d{2}/([^/]+)/", url)
+    return m.group(1) if m else None
+
+
+def _fetch_fr_metadata(doc_number: str) -> dict | None:
+    """Query the FR API for a document's metadata."""
+    try:
+        resp = httpx.get(
+            f"{FR_API}/{doc_number}.json",
+            timeout=30,
+            follow_redirects=True,
+        )
+        if resp.status_code != 200:
+            return None
+        data = resp.json()
+        # Verify this is actually a CMS rule (not a collision)
+        agencies = [a.get("id") for a in data.get("agencies", [])]
+        if "centers-for-medicare-medicaid-services" not in agencies and data.get(
+            "agency_names"
+        ):
+            # Might be a doc number collision — skip
+            return None
+        return data
+    except Exception:
+        return None
+
+
+def _download_file(url: str, dest: Path) -> bool:
+    """Download a file to a local path."""
+    try:
+        resp = httpx.get(url, timeout=120, follow_redirects=True)
+        if resp.status_code == 200 and len(resp.content) > 100:
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            dest.write_bytes(resp.content)
+            return True
+    except Exception:
+        pass
+    return False
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--key", default="", help="Process a single item key")
+    args = parser.parse_args()
+
+    store = connect.bib()
+    rules = store.list_items(item_type="rule")
+
+    if args.key:
+        rules = [r for r in rules if r.key == args.key]
+
+    # Get existing attachment filenames to avoid re-downloading
+    con = store._con()
+    existing_atts = set()
+    for row in con.execute("SELECT a.filename FROM attachments a").fetchall():
+        existing_atts.add(row[0])
+
+    download_dir = Path("data/fr_downloads")
+    stats = {"checked": 0, "downloaded": 0, "skipped": 0, "errors": 0}
+
+    for rule in rules:
+        stats["checked"] += 1
+
+        # Get document number from the item or parse from URL
+        doc_num = rule.document_number or _extract_doc_number(rule.url or "")
+        if not doc_num:
+            print(f"  SKIP {rule.key}: no document number")
+            stats["skipped"] += 1
+            continue
+
+        # Check if we already have PDF/TXT for this rule
+        pdf_name = f"{doc_num}.pdf"
+        txt_name = f"{doc_num}.txt"
+        has_pdf = pdf_name in existing_atts
+        has_txt = txt_name in existing_atts
+
+        if has_pdf and has_txt:
+            stats["skipped"] += 1
+            continue
+
+        print(f"  {rule.key}: {rule.title[:60]}  doc={doc_num}")
+
+        # Construct URLs directly (FR API can be slow/unreliable)
+        # govinfo.gov PDF: https://www.govinfo.gov/content/pkg/FR-YYYY-MM-DD/pdf/DOC.pdf
+        # We can also try the FR API for the exact URL
+        meta = _fetch_fr_metadata(doc_num)
+        time.sleep(0.5)  # Rate limit
+
+        if meta is None:
+            # Try constructing PDF URL from the rule's FR URL
+            # The FR URL has the date: /documents/YYYY/MM/DD/doc_number/
+            date_match = re.search(
+                r"/documents/(\d{4})/(\d{2})/(\d{2})/", rule.url or ""
+            )
+            if date_match:
+                y, m, d = date_match.groups()
+                pdf_url = f"https://www.govinfo.gov/content/pkg/FR-{y}-{m}-{d}/pdf/{doc_num}.pdf"
+                txt_url = f"https://www.federalregister.gov/documents/full_text/text/{y}/{m}/{d}/{doc_num}.txt"
+            else:
+                print("    ERROR: can't determine date from URL")
+                stats["errors"] += 1
+                continue
+        else:
+            pdf_url = meta.get("pdf_url", "")
+            txt_url = meta.get("raw_text_url", "")
+            # Construct txt URL if missing
+            if not txt_url and meta.get("publication_date"):
+                parts = meta["publication_date"].split("-")
+                if len(parts) == 3:
+                    txt_url = (
+                        f"https://www.federalregister.gov/documents/full_text/text/"
+                        f"{parts[0]}/{parts[1]}/{parts[2]}/{doc_num}.txt"
+                    )
+
+        if args.dry_run:
+            if not has_pdf:
+                print(f"    [dry-run] would download PDF: {pdf_url}")
+            if not has_txt:
+                print(f"    [dry-run] would download TXT: {txt_url}")
+            continue
+
+        # Download PDF
+        if not has_pdf and pdf_url:
+            pdf_path = download_dir / pdf_name
+            if _download_file(pdf_url, pdf_path):
+                store.attach_file(rule.key, pdf_path, title=f"FR {doc_num} (PDF)")
+                print(f"    PDF: {pdf_path.stat().st_size:,} bytes")
+                stats["downloaded"] += 1
+            else:
+                print("    PDF: download failed")
+                stats["errors"] += 1
+            time.sleep(1)
+
+        # Download TXT
+        if not has_txt and txt_url:
+            txt_path = download_dir / txt_name
+            if _download_file(txt_url, txt_path):
+                store.attach_file(rule.key, txt_path, title=f"FR {doc_num} (TXT)")
+                print(f"    TXT: {txt_path.stat().st_size:,} bytes")
+                stats["downloaded"] += 1
+            else:
+                print("    TXT: download failed")
+                stats["errors"] += 1
+            time.sleep(1)
+
+    print(
+        f"\nDone: checked={stats['checked']} downloaded={stats['downloaded']} "
+        f"skipped={stats['skipped']} errors={stats['errors']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/bib/__init__.py

@@ -62,6 +62,11 @@ from .item import Manual as Manual
 from .item import Regulation as Regulation
 from .item import Rule as Rule
 from .item import Source as Source
+from .pincite import CitationGraph as CitationGraph
+from .pincite import Pincite as Pincite
+from .pincite import build_citation_graph as build_citation_graph
+from .pincite import extract_all_pincites as extract_all_pincites
+from .pincite import parse_pincites as parse_pincites
 from .spider import crawl as crawl
 from .spider import crawl_all as crawl_all
 from .store import Store as Store

src/bib/pincite.py

@@ -0,0 +1,574 @@
+"""Pinpoint citations — structured Zotero references in docstrings.
+
+Embeds machine-parseable ``:pincite:`` directives in function docstrings
+that link to specific pages, sections, or paragraphs of bibliography
+items.  Enables a knowledge graph from regulatory source to function
+to table to column.
+
+Directive format::
+
+    :pincite:`JX46GQ9L p.14` — NDC validation rules
+    :pincite:`9ASETLJ4 §3.2` — exclusion criteria
+
+Components:
+
+- **Parse**: regex extraction of ``:pincite:`` directives from docstrings
+- **Store**: upsert/query pincite records in ``bib.sqlite``
+- **Graph**: build a citation knowledge graph (nodes + edges)
+- **Inject**: write pincite blocks back into source files
+- **Check**: validate all cited item keys exist
+
+Usage::
+
+    from bib.pincite import parse_pincites, build_citation_graph
+
+    pincites = parse_pincites(fn.__doc__, "aco.express.pharmacy.pharmacy_claims")
+    graph = build_citation_graph(store)
+    print(graph.to_mermaid())
+"""
+
+from __future__ import annotations
+
+import ast
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+
+from bib.tag import Tag
+
+if TYPE_CHECKING:
+    from bib.store import Store
+
+
+# ── Regex ────────────────────────────────────────────────────────────
+
+# Zotero key charset: 23456789ABCDEFGHIJKLMNPQRSTUVWXYZ (no 0, 1, O, L)
+_PINCITE_RE = re.compile(
+    r":pincite:`"
+    r"([23456789A-HJ-NP-Z]{8})"  # group 1: item key
+    r"(?:\s+(.+?))?"  # group 2: locator (optional)
+    r"`"
+    r"(?:\s*(?:--|—)\s*(.+))?"  # group 3: note (optional)
+)
+
+
+# ── Data model ───────────────────────────────────────────────────────
+
+
+class Pincite(BaseModel):
+    """A pinpoint citation linking a function to a bibliography item."""
+
+    fn_path: str
+    item_key: str
+    locator: str = ""
+    locator_type: str = ""
+    note: str = ""
+
+
+@dataclass
+class CitationNode:
+    """A node in the citation knowledge graph."""
+
+    id: str
+    kind: str  # "function" or "item"
+    label: str
+    metadata: dict = field(default_factory=dict)
+
+
+@dataclass
+class CitationEdge:
+    """An edge linking a function to a cited item."""
+
+    source: str  # fn_path
+    target: str  # item_key
+    locator: str = ""
+    locator_type: str = ""
+    note: str = ""
+
+
+@dataclass
+class CitationGraph:
+    """Knowledge graph of function → item citations."""
+
+    nodes: list[CitationNode] = field(default_factory=list)
+    edges: list[CitationEdge] = field(default_factory=list)
+
+    def functions_citing(self, item_key: str) -> list[str]:
+        """Return fn_paths that cite a given item."""
+        return [e.source for e in self.edges if e.target == item_key]
+
+    def items_cited_by(self, fn_path: str) -> list[str]:
+        """Return item_keys cited by a given function."""
+        return [e.target for e in self.edges if e.source == fn_path]
+
+    def to_networkx_dict(self) -> dict:
+        """Export as networkx-compatible node-link dict."""
+        return {
+            "directed": True,
+            "multigraph": False,
+            "nodes": [
+                {"id": n.id, "kind": n.kind, "label": n.label, **n.metadata}
+                for n in self.nodes
+            ],
+            "links": [
+                {
+                    "source": e.source,
+                    "target": e.target,
+                    "locator": e.locator,
+                    "note": e.note,
+                }
+                for e in self.edges
+            ],
+        }
+
+    def to_mermaid(self) -> str:
+        """Render as Mermaid flowchart."""
+        lines = ["graph LR"]
+        for n in self.nodes:
+            if n.kind == "function":
+                lines.append(f'    {_mermaid_id(n.id)}["{n.label}"]')
+            else:
+                lines.append(f'    {_mermaid_id(n.id)}("{n.label}")')
+        for e in self.edges:
+            label = e.locator or e.note or "cites"
+            lines.append(
+                f"    {_mermaid_id(e.source)} -->|{label}| {_mermaid_id(e.target)}"
+            )
+        return "\n".join(lines)
+
+
+def _mermaid_id(s: str) -> str:
+    """Sanitise a string for use as a Mermaid node ID."""
+    return re.sub(r"[^a-zA-Z0-9_]", "_", s)
+
+
+# ── Parse layer ──────────────────────────────────────────────────────
+
+
+def _classify_locator(locator: str) -> str:
+    """Classify a locator string into a type category."""
+    if not locator:
+        return ""
+    loc = locator.strip()
+    if loc.lower().startswith(("p.", "pp.")):
+        return "page"
+    if "CFR" in loc or "cfr" in loc:
+        return "cfr"
+    if re.match(r"\d+\s+FR\s+\d+", loc, re.IGNORECASE):
+        return "fr"
+    if loc.lower().startswith(("ch.", "ch ")):
+        return "chapter"
+    if loc.startswith("¶"):
+        return "paragraph"
+    if loc.startswith("§") or re.match(r"\d+[\d.]+", loc):
+        return "section"
+    return "other"
+
+
+def parse_pincites(docstring: str, fn_path: str) -> list[Pincite]:
+    """Extract ``:pincite:`` directives from a docstring.
+
+    Parameters
+    ----------
+    docstring : str
+        The raw docstring text.
+    fn_path : str
+        Fully qualified function path for attribution.
+
+    Returns
+    -------
+    list[Pincite]
+        Parsed pincite records, deduplicated by (key, locator).
+    """
+    if not docstring:
+        return []
+
+    results: list[Pincite] = []
+    seen: set[tuple[str, str]] = set()
+
+    for m in _PINCITE_RE.finditer(docstring):
+        key = m.group(1)
+        locator = (m.group(2) or "").strip()
+        note = (m.group(3) or "").strip()
+        pair = (key, locator)
+        if pair in seen:
+            continue
+        seen.add(pair)
+        results.append(
+            Pincite(
+                fn_path=fn_path,
+                item_key=key,
+                locator=locator,
+                locator_type=_classify_locator(locator),
+                note=note,
+            )
+        )
+    return results
+
+
+# ── AST discovery ────────────────────────────────────────────────────
+
+
+def _is_narwhalify(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
+    """Check if a function is decorated with @nw.narwhalify."""
+    for dec in node.decorator_list:
+        if isinstance(dec, ast.Name) and dec.id == "narwhalify":
+            return True
+        if isinstance(dec, ast.Attribute) and dec.attr == "narwhalify":
+            return True
+        if isinstance(dec, ast.Call):
+            func = dec.func
+            if isinstance(func, ast.Name) and func.id == "narwhalify":
+                return True
+            if isinstance(func, ast.Attribute) and func.attr == "narwhalify":
+                return True
+    return False
+
+
+def _path_to_module(path: Path, src_root: Path) -> str:
+    """Convert file path to dotted module name."""
+    rel = path.relative_to(src_root)
+    parts = list(rel.with_suffix("").parts)
+    if parts[-1] == "__init__":
+        parts = parts[:-1]
+    return ".".join(parts)
+
+
+def extract_all_pincites(src_root: Path | None = None) -> list[Pincite]:
+    """Walk every ``@nw.narwhalify`` function and extract pincites.
+
+    Parameters
+    ----------
+    src_root : Path, optional
+        Root of the source tree. Defaults to ``src/`` relative to
+        the project root.
+
+    Returns
+    -------
+    list[Pincite]
+        All pincites found across all narwhalify docstrings.
+    """
+    if src_root is None:
+        src_root = Path(__file__).parent.parent  # src/
+
+    results: list[Pincite] = []
+    for py_file in sorted(src_root.rglob("*.py")):
+        if "__pycache__" in py_file.parts or ".egg-info" in str(py_file):
+            continue
+        try:
+            source = py_file.read_text(encoding="utf-8")
+            tree = ast.parse(source, filename=str(py_file))
+        except SyntaxError:
+            continue
+
+        module = _path_to_module(py_file, src_root)
+        for node in ast.walk(tree):
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                continue
+            if not _is_narwhalify(node):
+                continue
+            doc = ast.get_docstring(node) or ""
+            if not doc:
+                continue
+            fn_path = f"{module}.{node.name}"
+            results.extend(parse_pincites(doc, fn_path))
+
+    return results
+
+
+# ── Store operations ─────────────────────────────────────────────────
+
+
+def upsert_pincites(store: Store, pincites: list[Pincite]) -> int:
+    """Insert or update pincite records and sync tags.
+
+    Returns the number of rows affected.
+    """
+    con = store._con()
+    # Ensure pincites table exists
+    con.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS pincites (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            fn_path TEXT NOT NULL,
+            item_key TEXT NOT NULL,
+            locator TEXT NOT NULL DEFAULT '',
+            locator_type TEXT NOT NULL DEFAULT '',
+            note TEXT NOT NULL DEFAULT '',
+            created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')),
+            FOREIGN KEY (item_key) REFERENCES items(key) ON DELETE CASCADE
+        );
+        CREATE UNIQUE INDEX IF NOT EXISTS idx_pincites_unique
+            ON pincites(fn_path, item_key, locator);
+        CREATE INDEX IF NOT EXISTS idx_pincites_fn ON pincites(fn_path);
+        CREATE INDEX IF NOT EXISTS idx_pincites_item ON pincites(item_key);
+        """
+    )
+
+    count = 0
+    for p in pincites:
+        # Skip if item doesn't exist
+        row = con.execute(
+            "SELECT key FROM items WHERE key = ?", (p.item_key,)
+        ).fetchone()
+        if row is None:
+            continue
+
+        con.execute(
+            """
+            INSERT INTO pincites (fn_path, item_key, locator, locator_type, note)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT (fn_path, item_key, locator)
+            DO UPDATE SET locator_type = excluded.locator_type,
+                          note = excluded.note
+            """,
+            (p.fn_path, p.item_key, p.locator, p.locator_type, p.note),
+        )
+        count += 1
+
+        # Sync tags on the item
+        pin_tag = Tag.pin(p.item_key, p.locator).label
+        fn_tag = Tag.fn(p.fn_path.split(".")[-1]).label
+        for tag_label in (pin_tag, fn_tag):
+            try:
+                store.add_tag(p.item_key, tag_label)
+            except (KeyError, Exception):
+                pass
+
+    con.commit()
+    return count
+
+
+def list_pincites(
+    store: Store,
+    *,
+    fn_path: str = "",
+    item_key: str = "",
+) -> list[Pincite]:
+    """Query pincites with optional filters."""
+    con = store._con()
+
+    # Table might not exist yet
+    try:
+        con.execute("SELECT 1 FROM pincites LIMIT 0")
+    except Exception:
+        return []
+
+    sql = (
+        "SELECT fn_path, item_key, locator, locator_type, note FROM pincites WHERE 1=1"
+    )
+    params: list[str] = []
+    if fn_path:
+        sql += " AND fn_path = ?"
+        params.append(fn_path)
+    if item_key:
+        sql += " AND item_key = ?"
+        params.append(item_key)
+    sql += " ORDER BY fn_path, item_key"
+
+    rows = con.execute(sql, params).fetchall()
+    return [
+        Pincite(
+            fn_path=r["fn_path"],
+            item_key=r["item_key"],
+            locator=r["locator"],
+            locator_type=r["locator_type"],
+            note=r["note"],
+        )
+        for r in rows
+    ]
+
+
+def check_pincite_keys(
+    pincites: list[Pincite], store: Store
+) -> list[tuple[Pincite, str]]:
+    """Verify all pincite item_keys exist in bib.sqlite."""
+    con = store._con()
+    errors: list[tuple[Pincite, str]] = []
+    for p in pincites:
+        row = con.execute(
+            "SELECT key FROM items WHERE key = ?", (p.item_key,)
+        ).fetchone()
+        if row is None:
+            errors.append((p, f"item key {p.item_key!r} not found in bib.sqlite"))
+    return errors
+
+
+# ── Knowledge graph ──────────────────────────────────────────────────
+
+
+def build_citation_graph(store: Store) -> CitationGraph:
+    """Build the citation knowledge graph from the pincites table.
+
+    Nodes: every function with pincites, every cited item.
+    Edges: one per pincite row.
+    """
+    pincites = list_pincites(store)
+
+    fn_ids: set[str] = set()
+    item_ids: set[str] = set()
+    nodes: list[CitationNode] = []
+    edges: list[CitationEdge] = []
+
+    for p in pincites:
+        if p.fn_path not in fn_ids:
+            fn_ids.add(p.fn_path)
+            short = p.fn_path.rsplit(".", 1)[-1]
+            nodes.append(CitationNode(id=p.fn_path, kind="function", label=short))
+        if p.item_key not in item_ids:
+            item_ids.add(p.item_key)
+            # Try to get title from store
+            try:
+                item = store.get(p.item_key)
+                label = item.title[:60] if item.title else p.item_key
+            except Exception:
+                label = p.item_key
+            nodes.append(CitationNode(id=p.item_key, kind="item", label=label))
+
+        edges.append(
+            CitationEdge(
+                source=p.fn_path,
+                target=p.item_key,
+                locator=p.locator,
+                locator_type=p.locator_type,
+                note=p.note,
+            )
+        )
+
+    return CitationGraph(nodes=nodes, edges=edges)
+
+
+# ── Injection ────────────────────────────────────────────────────────
+
+
+def format_pincite_block(pincites: list[Pincite]) -> str:
+    """Format pincites as a docstring References section.
+
+    Returns
+    -------
+    str
+        Formatted block, e.g.::
+
+            References
+            ~~~~~~~~~~
+            :pincite:`JX46GQ9L p.14` -- NDC validation rules
+    """
+    if not pincites:
+        return ""
+    lines = ["References", "~~~~~~~~~~"]
+    for p in pincites:
+        parts = [f":pincite:`{p.item_key}"]
+        if p.locator:
+            parts.append(f" {p.locator}")
+        parts.append("`")
+        if p.note:
+            parts.append(f" -- {p.note}")
+        lines.append("".join(parts))
+    return "\n".join(lines)
+
+
+def inject_pincites_into_source(
+    source_path: Path,
+    fn_name: str,
+    pincite_block: str,
+    *,
+    dry_run: bool = False,
+) -> str | None:
+    """Insert or update a References section in a function's docstring.
+
+    Uses AST to locate the function, then text manipulation to
+    insert the pincite block. Non-destructive: preserves all existing
+    docstring content outside the References block.
+
+    Parameters
+    ----------
+    source_path : Path
+        Python source file.
+    fn_name : str
+        Function name to target (unqualified).
+    pincite_block : str
+        Formatted pincite block to insert.
+    dry_run : bool
+        If True, return modified source without writing.
+
+    Returns
+    -------
+    str | None
+        Modified source text if changes were made, None otherwise.
+    """
+    source = source_path.read_text(encoding="utf-8")
+    try:
+        tree = ast.parse(source, filename=str(source_path))
+    except SyntaxError:
+        return None
+
+    # Find the target function
+    target = None
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            if node.name == fn_name:
+                target = node
+                break
+
+    if target is None:
+        return None
+
+    doc_node = target.body[0] if target.body else None
+    if not (
+        isinstance(doc_node, ast.Expr)
+        and isinstance(doc_node.value, ast.Constant)
+        and isinstance(doc_node.value.value, str)
+    ):
+        return None  # No docstring
+
+    lines = source.splitlines(keepends=True)
+
+    # Detect indentation from the docstring body
+    doc_start = doc_node.lineno - 1  # 0-indexed
+    indent = "    " * ((doc_node.col_offset // 4) + 1)
+
+    # Find the closing triple-quote line
+    doc_end = doc_node.end_lineno - 1  # 0-indexed
+    closing_line = lines[doc_end]
+
+    # Check for existing References section
+    ref_start = None
+    for i in range(doc_start, doc_end + 1):
+        stripped = lines[i].strip()
+        if stripped in ("References", "References:", "Sources", "Sources:"):
+            ref_start = i
+            break
+        if stripped.startswith("~~") and ref_start is not None:
+            continue
+
+    # Build indented block
+    indented_lines = []
+    for line in pincite_block.splitlines():
+        if line:
+            indented_lines.append(f"{indent}{line}\n")
+        else:
+            indented_lines.append("\n")
+
+    if ref_start is not None:
+        # Replace from References header to end of docstring (before closing """)
+        new_lines = lines[:ref_start] + indented_lines + [closing_line]
+        if doc_end + 1 < len(lines):
+            new_lines += lines[doc_end + 1 :]
+    else:
+        # Append before closing triple-quote
+        # Insert a blank line separator + the block
+        insert_lines = ["\n"] + indented_lines
+        new_lines = lines[:doc_end] + insert_lines + lines[doc_end:]
+
+    new_source = "".join(new_lines)
+
+    if new_source == source:
+        return None
+
+    if not dry_run:
+        source_path.write_text(new_source, encoding="utf-8")
+
+    return new_source

src/bib/schema.sql

@@ -77,3 +77,19 @@ CREATE TABLE IF NOT EXISTS notes (
     content    TEXT NOT NULL DEFAULT '',
     created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
 );
+
+CREATE TABLE IF NOT EXISTS pincites (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    fn_path      TEXT NOT NULL,
+    item_key     TEXT NOT NULL,
+    locator      TEXT NOT NULL DEFAULT '',
+    locator_type TEXT NOT NULL DEFAULT '',
+    note         TEXT NOT NULL DEFAULT '',
+    created_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')),
+    FOREIGN KEY (item_key) REFERENCES items(key) ON DELETE CASCADE
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_pincites_unique
+    ON pincites(fn_path, item_key, locator);
+CREATE INDEX IF NOT EXISTS idx_pincites_fn ON pincites(fn_path);
+CREATE INDEX IF NOT EXISTS idx_pincites_item ON pincites(item_key);

src/bib/store.py

@@ -527,3 +527,61 @@ class Store:
 
         items = [self.get(k) for k in keys]
         return format_bibliography(items, style=style)
+
+    # ── Pincites ─────────────────────────────────────────────────
+
+    def upsert_pincite(
+        self,
+        fn_path: str,
+        item_key: str,
+        locator: str = "",
+        locator_type: str = "",
+        note: str = "",
+    ) -> int:
+        """Insert or update a pincite record. Returns the row id."""
+        from bib.pincite import Pincite, upsert_pincites
+
+        return upsert_pincites(
+            self,
+            [
+                Pincite(
+                    fn_path=fn_path,
+                    item_key=item_key,
+                    locator=locator,
+                    locator_type=locator_type,
+                    note=note,
+                )
+            ],
+        )
+
+    def list_pincites(
+        self,
+        *,
+        fn_path: str = "",
+        item_key: str = "",
+    ) -> list[dict[str, str]]:
+        """Query pincites with optional filters."""
+        from bib.pincite import list_pincites as _list
+
+        return [p.model_dump() for p in _list(self, fn_path=fn_path, item_key=item_key)]
+
+    def delete_pincites(self, *, fn_path: str = "", item_key: str = "") -> int:
+        """Delete pincites matching filters. Returns count deleted."""
+        con = self._con()
+        try:
+            con.execute("SELECT 1 FROM pincites LIMIT 0")
+        except Exception:
+            return 0
+
+        sql = "DELETE FROM pincites WHERE 1=1"
+        params: list[str] = []
+        if fn_path:
+            sql += " AND fn_path = ?"
+            params.append(fn_path)
+        if item_key:
+            sql += " AND item_key = ?"
+            params.append(item_key)
+
+        cur = con.execute(sql, params)
+        con.commit()
+        return cur.rowcount

src/bib/tag.py

@@ -218,6 +218,26 @@ class Tag(BaseModel):
         """
         return cls(namespace="col", value=f"{column_ref}={description}")
 
+    @classmethod
+    def pin(cls, item_key: str, locator: str = "") -> Tag:
+        """Tag linking a bibliography item to a pinpoint citation.
+
+        Parameters
+        ----------
+        item_key : str
+            8-char Zotero-compatible item key.
+        locator : str
+            Locator string (page, section, etc.), or empty for
+            item-level citation.
+
+        Examples::
+
+            Tag.pin("JX46GQ9L", "p.14")   # pin:JX46GQ9L/p.14
+            Tag.pin("ABC12345")            # pin:ABC12345
+        """
+        value = f"{item_key}/{locator}" if locator else item_key
+        return cls(namespace="pin", value=value)
+
     @classmethod
     def fn(cls, qualified_name: str) -> Tag:
         """Tag linking a bibliography item to an express function.

tests/bib/test_pincite.py

@@ -0,0 +1,339 @@
+"""Tests for bib.pincite — pinpoint citation system."""
+
+from __future__ import annotations
+
+import textwrap
+
+import pytest
+
+from bib.pincite import (
+    Pincite,
+    _classify_locator,
+    build_citation_graph,
+    check_pincite_keys,
+    format_pincite_block,
+    inject_pincites_into_source,
+    list_pincites,
+    parse_pincites,
+    upsert_pincites,
+)
+from bib.store import Store
+
+# ── Fixtures ─────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def store(tmp_path):
+    """In-memory-like store using a temp database."""
+    db = tmp_path / "test_bib.sqlite"
+    s = Store(database=str(db))
+    return s
+
+
+@pytest.fixture
+def store_with_items(store):
+    """Store with two items for pincite testing."""
+    from bib.item import Source
+
+    store.create(Source(key="JX46GQ9L", title="Tag vocabulary item"))
+    store.create(Source(key="9ASETLJ4", title="Nature citation standards"))
+    return store
+
+
+# ── Parse ────────────────────────────────────────────────────────────
+
+
+class TestParsePincites:
+    def test_basic(self):
+        doc = ":pincite:`JX46GQ9L p.14` -- NDC validation rules"
+        result = parse_pincites(doc, "mod.fn")
+        assert len(result) == 1
+        assert result[0].item_key == "JX46GQ9L"
+        assert result[0].locator == "p.14"
+        assert result[0].note == "NDC validation rules"
+        assert result[0].fn_path == "mod.fn"
+
+    def test_no_locator(self):
+        doc = ":pincite:`JX46GQ9L`"
+        result = parse_pincites(doc, "mod.fn")
+        assert len(result) == 1
+        assert result[0].locator == ""
+
+    def test_no_note(self):
+        doc = ":pincite:`JX46GQ9L p.14`"
+        result = parse_pincites(doc, "mod.fn")
+        assert len(result) == 1
+        assert result[0].note == ""
+
+    def test_section_locator(self):
+        doc = ":pincite:`9ASETLJ4 §3.2` — exclusion criteria"
+        result = parse_pincites(doc, "mod.fn")
+        assert result[0].locator == "§3.2"
+        assert result[0].locator_type == "section"
+        assert result[0].note == "exclusion criteria"
+
+    def test_cfr_locator(self):
+        doc = ":pincite:`ABC23456 42 CFR §414.22` -- payment rates"
+        result = parse_pincites(doc, "mod.fn")
+        assert result[0].locator == "42 CFR §414.22"
+        assert result[0].locator_type == "cfr"
+
+    def test_fr_locator(self):
+        doc = ":pincite:`DEF78923 90 FR 86252` -- final rule"
+        result = parse_pincites(doc, "mod.fn")
+        assert result[0].locator == "90 FR 86252"
+        assert result[0].locator_type == "fr"
+
+    def test_multiple(self):
+        doc = textwrap.dedent("""
+            :pincite:`JX46GQ9L p.14` -- first
+            :pincite:`9ASETLJ4 §3.2` -- second
+            :pincite:`ABC23456 Ch. 12` -- third
+        """)
+        result = parse_pincites(doc, "mod.fn")
+        assert len(result) == 3
+
+    def test_deduplication(self):
+        doc = ":pincite:`JX46GQ9L p.14` -- first\n:pincite:`JX46GQ9L p.14` -- dupe"
+        result = parse_pincites(doc, "mod.fn")
+        assert len(result) == 1
+
+    def test_empty_docstring(self):
+        assert parse_pincites("", "mod.fn") == []
+        assert parse_pincites(None, "mod.fn") == []
+
+    def test_no_pincites(self):
+        doc = "Just a plain docstring with no citations."
+        assert parse_pincites(doc, "mod.fn") == []
+
+    def test_invalid_key_charset(self):
+        # 0, 1, O, L are not valid Zotero key chars
+        doc = ":pincite:`00000000 p.1` -- bad key"
+        assert parse_pincites(doc, "mod.fn") == []
+
+    def test_em_dash(self):
+        doc = ":pincite:`JX46GQ9L p.14` — em dash note"
+        result = parse_pincites(doc, "mod.fn")
+        assert result[0].note == "em dash note"
+
+
+# ── Classify locator ─────────────────────────────────────────────────
+
+
+class TestClassifyLocator:
+    def test_page(self):
+        assert _classify_locator("p.14") == "page"
+        assert _classify_locator("pp.8-12") == "page"
+
+    def test_section(self):
+        assert _classify_locator("§3.2") == "section"
+
+    def test_cfr(self):
+        assert _classify_locator("42 CFR §414.22") == "cfr"
+
+    def test_fr(self):
+        assert _classify_locator("90 FR 86252") == "fr"
+
+    def test_chapter(self):
+        assert _classify_locator("Ch. 12") == "chapter"
+
+    def test_paragraph(self):
+        assert _classify_locator("¶4") == "paragraph"
+
+    def test_empty(self):
+        assert _classify_locator("") == ""
+
+
+# ── Store operations ─────────────────────────────────────────────────
+
+
+class TestUpsertPincites:
+    def test_insert(self, store_with_items):
+        pincites = [
+            Pincite(fn_path="mod.fn", item_key="JX46GQ9L", locator="p.14", note="test"),
+        ]
+        count = upsert_pincites(store_with_items, pincites)
+        assert count == 1
+
+        result = list_pincites(store_with_items, fn_path="mod.fn")
+        assert len(result) == 1
+        assert result[0].item_key == "JX46GQ9L"
+
+    def test_upsert_idempotent(self, store_with_items):
+        p = Pincite(fn_path="mod.fn", item_key="JX46GQ9L", locator="p.14")
+        upsert_pincites(store_with_items, [p])
+        upsert_pincites(store_with_items, [p])
+        result = list_pincites(store_with_items, fn_path="mod.fn")
+        assert len(result) == 1
+
+    def test_skips_missing_item(self, store_with_items):
+        p = Pincite(fn_path="mod.fn", item_key="ZZZZZZZZ", locator="p.1")
+        count = upsert_pincites(store_with_items, [p])
+        assert count == 0
+
+    def test_tags_added(self, store_with_items):
+        p = Pincite(fn_path="mod.fn", item_key="JX46GQ9L", locator="p.14")
+        upsert_pincites(store_with_items, [p])
+        item = store_with_items.get("JX46GQ9L")
+        assert any("pin:" in t for t in item.tags)
+
+
+class TestListPincites:
+    def test_by_fn(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="a.b", item_key="JX46GQ9L", locator="p.1"),
+                Pincite(fn_path="c.d", item_key="9ASETLJ4", locator="§2"),
+            ],
+        )
+        result = list_pincites(store_with_items, fn_path="a.b")
+        assert len(result) == 1
+        assert result[0].item_key == "JX46GQ9L"
+
+    def test_by_item(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="a.b", item_key="JX46GQ9L", locator="p.1"),
+                Pincite(fn_path="c.d", item_key="JX46GQ9L", locator="p.2"),
+            ],
+        )
+        result = list_pincites(store_with_items, item_key="JX46GQ9L")
+        assert len(result) == 2
+
+    def test_empty_store(self, store):
+        assert list_pincites(store) == []
+
+
+# ── Check keys ───────────────────────────────────────────────────────
+
+
+class TestCheckKeys:
+    def test_valid(self, store_with_items):
+        p = Pincite(fn_path="mod.fn", item_key="JX46GQ9L")
+        errors = check_pincite_keys([p], store_with_items)
+        assert len(errors) == 0
+
+    def test_invalid(self, store_with_items):
+        p = Pincite(fn_path="mod.fn", item_key="NOTFOUND")
+        errors = check_pincite_keys([p], store_with_items)
+        assert len(errors) == 1
+        assert "not found" in errors[0][1]
+
+
+# ── Knowledge graph ──────────────────────────────────────────────────
+
+
+class TestCitationGraph:
+    def test_build(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="mod.fn_a", item_key="JX46GQ9L", locator="p.1"),
+                Pincite(fn_path="mod.fn_b", item_key="JX46GQ9L", locator="p.2"),
+                Pincite(fn_path="mod.fn_a", item_key="9ASETLJ4", locator="§3"),
+            ],
+        )
+        graph = build_citation_graph(store_with_items)
+        assert len(graph.nodes) == 4  # 2 functions + 2 items
+        assert len(graph.edges) == 3
+
+    def test_functions_citing(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="a.b", item_key="JX46GQ9L"),
+                Pincite(fn_path="c.d", item_key="JX46GQ9L"),
+            ],
+        )
+        graph = build_citation_graph(store_with_items)
+        fns = graph.functions_citing("JX46GQ9L")
+        assert set(fns) == {"a.b", "c.d"}
+
+    def test_items_cited_by(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="a.b", item_key="JX46GQ9L"),
+                Pincite(fn_path="a.b", item_key="9ASETLJ4"),
+            ],
+        )
+        graph = build_citation_graph(store_with_items)
+        items = graph.items_cited_by("a.b")
+        assert set(items) == {"JX46GQ9L", "9ASETLJ4"}
+
+    def test_to_mermaid(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="mod.fn", item_key="JX46GQ9L", locator="p.1"),
+            ],
+        )
+        graph = build_citation_graph(store_with_items)
+        mermaid = graph.to_mermaid()
+        assert "graph LR" in mermaid
+        assert "mod_fn" in mermaid
+
+    def test_to_networkx_dict(self, store_with_items):
+        upsert_pincites(
+            store_with_items,
+            [
+                Pincite(fn_path="mod.fn", item_key="JX46GQ9L"),
+            ],
+        )
+        graph = build_citation_graph(store_with_items)
+        d = graph.to_networkx_dict()
+        assert d["directed"] is True
+        assert len(d["nodes"]) == 2
+        assert len(d["links"]) == 1
+
+
+# ── Format + inject ──────────────────────────────────────────────────
+
+
+class TestFormatPinciteBlock:
+    def test_single(self):
+        p = Pincite(fn_path="m.f", item_key="JX46GQ9L", locator="p.14", note="test")
+        block = format_pincite_block([p])
+        assert ":pincite:`JX46GQ9L p.14` -- test" in block
+        assert block.startswith("References")
+
+    def test_no_locator(self):
+        p = Pincite(fn_path="m.f", item_key="JX46GQ9L")
+        block = format_pincite_block([p])
+        assert ":pincite:`JX46GQ9L`" in block
+
+    def test_empty(self):
+        assert format_pincite_block([]) == ""
+
+
+class TestInjectPincites:
+    def test_append_to_docstring(self, tmp_path):
+        source = textwrap.dedent('''
+            def foo():
+                """Existing docstring."""
+                return 1
+        ''').lstrip()
+        f = tmp_path / "test_mod.py"
+        f.write_text(source)
+
+        block = "References\n~~~~~~~~~~\n:pincite:`JX46GQ9L p.14` -- test"
+        result = inject_pincites_into_source(f, "foo", block, dry_run=True)
+        assert result is not None
+        assert ":pincite:`JX46GQ9L p.14`" in result
+        assert "Existing docstring." in result
+
+    def test_skip_no_docstring(self, tmp_path):
+        source = "def foo():\n    return 1\n"
+        f = tmp_path / "test_mod.py"
+        f.write_text(source)
+        result = inject_pincites_into_source(f, "foo", "block")
+        assert result is None
+
+    def test_skip_missing_function(self, tmp_path):
+        source = "def bar():\n    pass\n"
+        f = tmp_path / "test_mod.py"
+        f.write_text(source)
+        result = inject_pincites_into_source(f, "foo", "block")
+        assert result is None