b5e07e67b2
The PFS register-and-walk path now serves the annual IPPS/LTCH and OPPS/ASC rules and DMEPOS payment rules: title-filtered FR discovery (discover-*-rules) and docket registration + comment walks (fetch-*-comments). Correction documents (C1-...) parse, a null FR abstract no longer becomes 'None', and the store waits out the mail poller's write bursts (busy_timeout) instead of failing SQLITE_BUSY.
888 lines
31 KiB
Python
888 lines
31 KiB
Python
"""Own-storage bibliography backend — replaces Zotero contexts.
|
|
|
|
``Store`` owns a SQLite database with the denormalized 8-table schema
|
|
(``schema.sql``) and provides all CRUD, querying, tagging, attachment,
|
|
and citation-formatting operations.
|
|
|
|
Usage::
|
|
|
|
from bib.store import Store
|
|
from bib.item import Rule
|
|
|
|
s = Store()
|
|
key = s.create(Rule(title="CY 2026 PFS Final Rule", fr_volume="90"))
|
|
item = s.get(key)
|
|
print(s.format_citation(key))
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import mimetypes
|
|
import random
|
|
import shutil
|
|
import sqlite3
|
|
from importlib import resources
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
from bib.dockets import Docket
|
|
from bib.item import Item
|
|
|
|
|
|
def _generate_key() -> str:
|
|
"""8-char key using Zotero-compatible charset (no 0, 1, L, O)."""
|
|
chars = "23456789ABCDEFGHIJKMNPQRSTUVWXYZ"
|
|
return "".join(random.choices(chars, k=8)) # noqa: S311
|
|
|
|
|
|
class Store:
|
|
"""SQLite-backed bibliography store.
|
|
|
|
Parameters
|
|
----------
|
|
database : str or Path
|
|
Path to the SQLite file. Use ``":memory:"`` for in-memory.
|
|
storage_dir : str or Path, optional
|
|
Directory for file attachments. Defaults to a ``storage/``
|
|
sibling of the database file.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
database: str | Path | None = None,
|
|
*,
|
|
storage_dir: str | Path = "",
|
|
) -> None:
|
|
if database is None:
|
|
from conf import path
|
|
|
|
database = str(path("db.bib"))
|
|
self._db_path = str(database)
|
|
self._connection: sqlite3.Connection | None = None
|
|
|
|
if storage_dir:
|
|
self._storage = Path(storage_dir)
|
|
elif self._db_path == ":memory:":
|
|
self._storage = Path("storage")
|
|
else:
|
|
self._storage = Path(self._db_path).parent / "storage"
|
|
|
|
self._init_schema()
|
|
|
|
# ── Connection / schema ──────────────────────────────────────
|
|
|
|
def _con(self) -> sqlite3.Connection:
|
|
if self._connection is None:
|
|
self._connection = sqlite3.connect(self._db_path)
|
|
self._connection.row_factory = sqlite3.Row
|
|
self._connection.execute("PRAGMA journal_mode=WAL")
|
|
self._connection.execute("PRAGMA foreign_keys=ON")
|
|
# The mail poller writes this file on its own cadence. A comment
|
|
# walk waits through that burst instead of dying on SQLITE_BUSY.
|
|
self._connection.execute("PRAGMA busy_timeout=60000")
|
|
return self._connection
|
|
|
|
def _init_schema(self) -> None:
|
|
ddl = resources.files("bib").joinpath("schema.sql").read_text()
|
|
self._con().executescript(ddl)
|
|
# The (item_id, filename) unique index can only exist once the
|
|
# dedupe migration has run (dev/scripts/dedupe_attachments.py);
|
|
# until then we leave it off rather than fail to open the store.
|
|
# Once it exists, sqlite_master says so directly — skip the
|
|
# GROUP BY scan (which otherwise runs on every Store() open).
|
|
con = self._con()
|
|
indexed = con.execute(
|
|
"SELECT 1 FROM sqlite_master WHERE type = 'index' "
|
|
"AND name = 'idx_attachments_item_filename'"
|
|
).fetchone()
|
|
if indexed is not None:
|
|
return
|
|
dup = con.execute(
|
|
"SELECT 1 FROM attachments GROUP BY item_id, filename HAVING count(*) > 1 LIMIT 1"
|
|
).fetchone()
|
|
if dup is None:
|
|
con.execute(
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS idx_attachments_item_filename "
|
|
"ON attachments(item_id, filename)"
|
|
)
|
|
con.commit()
|
|
|
|
def close(self) -> None:
|
|
if self._connection is not None:
|
|
self._connection.close()
|
|
self._connection = None
|
|
|
|
def __del__(self) -> None:
|
|
try:
|
|
self.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Item CRUD ────────────────────────────────────────────────
|
|
|
|
def create(
|
|
self,
|
|
item: Item,
|
|
*,
|
|
tags: list[Any] | None = None,
|
|
collection: str = "",
|
|
) -> str:
|
|
"""Insert a new item. Returns the item key."""
|
|
if tags:
|
|
for tag in tags:
|
|
label = tag.label if hasattr(tag, "label") else str(tag)
|
|
item.add_tag(label)
|
|
if collection and collection not in item.collections:
|
|
item.collections.append(collection)
|
|
|
|
item.stamp_access()
|
|
row = item.to_row()
|
|
key = row["key"] or _generate_key()
|
|
row["key"] = key
|
|
|
|
con = self._con()
|
|
con.execute(
|
|
"""INSERT INTO items
|
|
(key, item_type, title, url, date_published,
|
|
access_date, abstract, institution, extra, extra_json)
|
|
VALUES (:key, :item_type, :title, :url,
|
|
:date_published, :access_date, :abstract,
|
|
:institution, :extra, :extra_json)""",
|
|
row,
|
|
)
|
|
item_id = con.execute("SELECT id FROM items WHERE key = ?", (key,)).fetchone()[
|
|
"id"
|
|
]
|
|
|
|
self._sync_tags(item_id, item.tags)
|
|
self._sync_collections(item_id, item.collections)
|
|
con.commit()
|
|
return key
|
|
|
|
def get(self, key: str) -> Item:
|
|
"""Fetch an item by key."""
|
|
con = self._con()
|
|
row = con.execute("SELECT * FROM items WHERE key = ?", (key,)).fetchone()
|
|
if row is None:
|
|
raise KeyError(f"Item not found: {key}")
|
|
|
|
d = dict(row)
|
|
item_id = d.pop("id")
|
|
d.pop("created_at", None)
|
|
d.pop("updated_at", None)
|
|
|
|
# Load tags
|
|
tag_rows = con.execute(
|
|
"""SELECT t.name FROM item_tags it
|
|
JOIN tags t ON it.tag_id = t.id
|
|
WHERE it.item_id = ?""",
|
|
(item_id,),
|
|
).fetchall()
|
|
d["tags"] = [r["name"] for r in tag_rows]
|
|
|
|
# Load collections
|
|
col_rows = con.execute(
|
|
"""SELECT c.key FROM collection_items ci
|
|
JOIN collections c ON ci.collection_id = c.id
|
|
WHERE ci.item_id = ?""",
|
|
(item_id,),
|
|
).fetchall()
|
|
d["collections"] = [r["key"] for r in col_rows]
|
|
|
|
return Item.from_row(d)
|
|
|
|
def update(self, key: str, **fields: Any) -> None:
|
|
"""Update specific fields on an item."""
|
|
if not fields:
|
|
return
|
|
|
|
con = self._con()
|
|
# Separate tag/collection updates from column updates
|
|
new_tags = fields.pop("tags", None)
|
|
new_collections = fields.pop("collections", None)
|
|
|
|
if fields:
|
|
sets = ", ".join(f"{k} = ?" for k in fields)
|
|
vals = list(fields.values()) + [key]
|
|
con.execute(
|
|
f"UPDATE items SET {sets}, " # noqa: S608
|
|
"updated_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') "
|
|
f"WHERE key = ?",
|
|
vals,
|
|
)
|
|
|
|
if new_tags is not None or new_collections is not None:
|
|
row = con.execute("SELECT id FROM items WHERE key = ?", (key,)).fetchone()
|
|
if row is None:
|
|
raise KeyError(f"Item not found: {key}")
|
|
item_id = row["id"]
|
|
if new_tags is not None:
|
|
self._sync_tags(item_id, new_tags)
|
|
if new_collections is not None:
|
|
self._sync_collections(item_id, new_collections)
|
|
if not fields:
|
|
# A tags/collections-only rewrite still changes the item
|
|
# as the indexer sees it (year:/project:/cms-rule: land in
|
|
# chunk metadata), and nothing else stamped the row.
|
|
self._stamp(key)
|
|
|
|
con.commit()
|
|
|
|
def delete(self, key: str) -> None:
|
|
"""Delete an item and its associations."""
|
|
con = self._con()
|
|
con.execute("DELETE FROM items WHERE key = ?", (key,))
|
|
con.commit()
|
|
|
|
_COMPARE_COLS = (
|
|
"item_type",
|
|
"title",
|
|
"url",
|
|
"date_published",
|
|
"abstract",
|
|
"institution",
|
|
"extra",
|
|
"extra_json",
|
|
)
|
|
|
|
def upsert(
|
|
self,
|
|
item: Item,
|
|
*,
|
|
tags: list[Any] | None = None,
|
|
collection: str = "",
|
|
) -> str:
|
|
"""Create or update an item, deduplicating by URL."""
|
|
return self.upsert_status(item, tags=tags, collection=collection)[0]
|
|
|
|
def upsert_status(
|
|
self,
|
|
item: Item,
|
|
*,
|
|
tags: list[Any] | None = None,
|
|
collection: str = "",
|
|
) -> tuple[str, str]:
|
|
"""Like :meth:`upsert` but also report what happened:
|
|
``"created"``, ``"updated"`` or ``"unchanged"``.
|
|
|
|
``unchanged`` means every compared column, the merged tag set and
|
|
the merged collection set already match the stored row — nothing
|
|
is written, no ``access_date``/``updated_at`` stamp, no tag row
|
|
churn. A re-farm over a complete docket is therefore a read-only
|
|
pass over ``items``.
|
|
|
|
Merge rules, all of which make ``unchanged`` a comparison against
|
|
the *merged* row rather than the incoming one:
|
|
|
|
* columns: an empty incoming value keeps the stored one (a
|
|
list-walk row carries no body, and must not blank an enriched
|
|
abstract). ``extra_json`` is merged per key instead of
|
|
column-wise: a stored key survives unless the incoming
|
|
payload sets that key to a non-empty value, so a list-walk
|
|
row's empty subclass fields (``{"effective_date": "", ...}``)
|
|
never blank an already-stored value.
|
|
* tags and collections: union-merged with what is stored, never
|
|
replaced. Removal goes through :meth:`remove_tag`.
|
|
"""
|
|
if tags:
|
|
for tag in tags:
|
|
label = tag.label if hasattr(tag, "label") else str(tag)
|
|
item.add_tag(label)
|
|
if collection and collection not in item.collections:
|
|
item.collections.append(collection)
|
|
|
|
if not item.url:
|
|
return self.create(item), "created"
|
|
|
|
con = self._con()
|
|
existing = con.execute(
|
|
"SELECT key FROM items WHERE url = ?", (item.url,)
|
|
).fetchone()
|
|
if not existing:
|
|
return self.create(item), "created"
|
|
|
|
ekey = existing["key"]
|
|
# Merge with what's already stored — a re-ingest must never
|
|
# clobber tags/collections curated on the row since the last
|
|
# ingest (#624). Deliberate removal goes through remove_tag.
|
|
current = self.get(ekey)
|
|
merged_tags = list(dict.fromkeys([*current.tags, *item.tags]))
|
|
merged_cols = list(dict.fromkeys([*current.collections, *item.collections]))
|
|
|
|
new_row = item.to_row()
|
|
cur_row = current.to_row()
|
|
# Merge, don't replace: an empty incoming value carries no
|
|
# information (a list-walk row has no body), so the stored value
|
|
# wins.
|
|
for c in self._COMPARE_COLS:
|
|
if c != "extra_json" and not new_row.get(c) and cur_row.get(c):
|
|
new_row[c] = cur_row[c]
|
|
setattr(item, c, cur_row[c])
|
|
|
|
# extra_json is a full serialization of subclass-specific fields
|
|
# every time, so a column-wise empty check can't apply — merge
|
|
# the decoded dict per key instead: a stored key survives unless
|
|
# the incoming payload sets it to a non-empty value.
|
|
cur_extra = json.loads(cur_row.get("extra_json") or "{}")
|
|
new_extra = json.loads(new_row.get("extra_json") or "{}")
|
|
merged_extra = {**cur_extra, **{k: v for k, v in new_extra.items() if v}}
|
|
merged_extra_json = json.dumps(merged_extra)
|
|
new_row["extra_json"] = merged_extra_json
|
|
|
|
same_cols = all(
|
|
new_row.get(c, "") == cur_row.get(c, "") for c in self._COMPARE_COLS
|
|
)
|
|
if (
|
|
same_cols
|
|
and set(merged_tags) == set(current.tags)
|
|
and set(merged_cols) == set(current.collections)
|
|
):
|
|
return ekey, "unchanged"
|
|
|
|
item.stamp_access()
|
|
row = item.to_row()
|
|
row.pop("key", None)
|
|
row["extra_json"] = merged_extra_json
|
|
row["tags"] = merged_tags
|
|
row["collections"] = merged_cols
|
|
self.update(ekey, **row)
|
|
return ekey, "updated"
|
|
|
|
# ── Query ────────────────────────────────────────────────────
|
|
|
|
def _filter_clause(
|
|
self, tag: str, item_type: str, collection: str, query: str
|
|
) -> tuple[str, list[Any]]:
|
|
"""Build the shared JOIN/WHERE suffix for item filters."""
|
|
joins: list[str] = []
|
|
wheres: list[str] = []
|
|
params: list[Any] = []
|
|
|
|
if tag:
|
|
joins.append(
|
|
" JOIN item_tags it ON i.id = it.item_id"
|
|
" JOIN tags t ON it.tag_id = t.id"
|
|
)
|
|
wheres.append("t.name = ?")
|
|
params.append(tag)
|
|
|
|
if collection:
|
|
joins.append(
|
|
" JOIN collection_items ci ON i.id = ci.item_id"
|
|
" JOIN collections c ON ci.collection_id = c.id"
|
|
)
|
|
wheres.append("c.key = ?")
|
|
params.append(collection)
|
|
|
|
if item_type:
|
|
wheres.append("i.item_type = ?")
|
|
params.append(item_type)
|
|
|
|
if query:
|
|
wheres.append("(i.title LIKE ? OR i.abstract LIKE ?)")
|
|
params.extend([f"%{query}%", f"%{query}%"])
|
|
|
|
clause = "".join(joins)
|
|
if wheres:
|
|
clause += " WHERE " + " AND ".join(wheres)
|
|
return clause, params
|
|
|
|
def list_items(
|
|
self,
|
|
*,
|
|
tag: str = "",
|
|
item_type: str = "",
|
|
collection: str = "",
|
|
query: str = "",
|
|
limit: int | None = None,
|
|
) -> list[Item]:
|
|
"""List items with optional filters.
|
|
|
|
Pass ``limit`` to bound the scan — each returned item costs a
|
|
``get()`` round-trip, so an unbounded call on a large store
|
|
(Zotero sync is ~180k items) takes minutes.
|
|
"""
|
|
con = self._con()
|
|
clause, params = self._filter_clause(tag, item_type, collection, query)
|
|
full_sql = f"SELECT DISTINCT i.key FROM items i{clause} ORDER BY i.id"
|
|
if limit is not None:
|
|
full_sql += " LIMIT ?"
|
|
params.append(limit)
|
|
|
|
rows = con.execute(full_sql, params).fetchall()
|
|
return [self.get(r["key"]) for r in rows]
|
|
|
|
def count(
|
|
self,
|
|
*,
|
|
tag: str = "",
|
|
item_type: str = "",
|
|
collection: str = "",
|
|
query: str = "",
|
|
) -> int:
|
|
"""Count items matching filters (single COUNT query — no item
|
|
hydration; ``len(list_items())`` is O(n) get() calls)."""
|
|
con = self._con()
|
|
clause, params = self._filter_clause(tag, item_type, collection, query)
|
|
full_sql = f"SELECT COUNT(DISTINCT i.key) AS n FROM items i{clause}"
|
|
return int(con.execute(full_sql, params).fetchone()["n"])
|
|
|
|
def to_dataframe(self, **filters: Any) -> Any:
|
|
"""Export items as a polars DataFrame (lazy import)."""
|
|
import polars as pl
|
|
|
|
items = self.list_items(**filters)
|
|
rows = []
|
|
for item in items:
|
|
rows.append(
|
|
{
|
|
"key": item.key,
|
|
"item_type": item.item_type,
|
|
"title": item.title,
|
|
"url": item.url,
|
|
"date_published": item.date_published,
|
|
"tags": "; ".join(item.tags),
|
|
"institution": item.institution,
|
|
}
|
|
)
|
|
|
|
return pl.DataFrame(rows)
|
|
|
|
# ── Collections ──────────────────────────────────────────────
|
|
|
|
def list_collections(self) -> list[dict[str, Any]]:
|
|
"""List all collections."""
|
|
con = self._con()
|
|
rows = con.execute(
|
|
"""SELECT c.key, c.name, pc.key AS parent_key,
|
|
(SELECT count(*) FROM collection_items ci
|
|
WHERE ci.collection_id = c.id) AS item_count
|
|
FROM collections c
|
|
LEFT JOIN collections pc ON c.parent_id = pc.id
|
|
ORDER BY c.name"""
|
|
).fetchall()
|
|
return [
|
|
{
|
|
"key": r["key"],
|
|
"name": r["name"],
|
|
"parent_key": r["parent_key"] or "",
|
|
"item_count": r["item_count"],
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
def ensure_collections(
|
|
self,
|
|
hierarchy: dict[str, dict],
|
|
parent_key: str | None = None,
|
|
) -> dict[str, str]:
|
|
"""Create collections from a nested dict. Returns {name: key}."""
|
|
con = self._con()
|
|
result: dict[str, str] = {}
|
|
|
|
parent_id: int | None = None
|
|
if parent_key:
|
|
row = con.execute(
|
|
"SELECT id FROM collections WHERE key = ?",
|
|
(parent_key,),
|
|
).fetchone()
|
|
if row:
|
|
parent_id = row["id"]
|
|
|
|
for name, children in hierarchy.items():
|
|
if parent_id is not None:
|
|
row = con.execute(
|
|
"SELECT id, key FROM collections WHERE name = ? AND parent_id = ?",
|
|
(name, parent_id),
|
|
).fetchone()
|
|
else:
|
|
row = con.execute(
|
|
"SELECT id, key FROM collections "
|
|
"WHERE name = ? AND parent_id IS NULL",
|
|
(name,),
|
|
).fetchone()
|
|
|
|
if row:
|
|
key = row["key"]
|
|
else:
|
|
key = _generate_key()
|
|
con.execute(
|
|
"INSERT INTO collections (key, name, parent_id) VALUES (?, ?, ?)",
|
|
(key, name, parent_id),
|
|
)
|
|
|
|
result[name] = key
|
|
|
|
if children:
|
|
child_map = self.ensure_collections(children, parent_key=key)
|
|
result.update(child_map)
|
|
|
|
con.commit()
|
|
return result
|
|
|
|
# ── Tags ─────────────────────────────────────────────────────
|
|
|
|
def _ensure_tag(self, name: str) -> int:
|
|
con = self._con()
|
|
row = con.execute("SELECT id FROM tags WHERE name = ?", (name,)).fetchone()
|
|
if row:
|
|
return row["id"]
|
|
cur = con.execute("INSERT INTO tags (name) VALUES (?)", (name,))
|
|
return cur.lastrowid # type: ignore[return-value]
|
|
|
|
def _sync_tags(self, item_id: int, tag_names: list[str]) -> None:
|
|
con = self._con()
|
|
con.execute("DELETE FROM item_tags WHERE item_id = ?", (item_id,))
|
|
for name in tag_names:
|
|
if not name:
|
|
continue
|
|
tag_id = self._ensure_tag(name)
|
|
con.execute(
|
|
"INSERT OR IGNORE INTO item_tags (item_id, tag_id) VALUES (?, ?)",
|
|
(item_id, tag_id),
|
|
)
|
|
|
|
def _sync_collections(self, item_id: int, collection_keys: list[str]) -> None:
|
|
con = self._con()
|
|
con.execute("DELETE FROM collection_items WHERE item_id = ?", (item_id,))
|
|
for ckey in collection_keys:
|
|
row = con.execute(
|
|
"SELECT id FROM collections WHERE key = ?", (ckey,)
|
|
).fetchone()
|
|
if row:
|
|
con.execute(
|
|
"INSERT OR IGNORE INTO collection_items "
|
|
"(collection_id, item_id) VALUES (?, ?)",
|
|
(row["id"], item_id),
|
|
)
|
|
|
|
def _stamp(self, item_key: str) -> None:
|
|
"""Mark the item row as changed *now*.
|
|
|
|
Tags carry indexed metadata (``year:``, ``project:``,
|
|
``cms-rule:``), so a tag-only edit has to move ``updated_at`` or
|
|
the llm indexer's fingerprint can never see it (#615).
|
|
"""
|
|
self._con().execute(
|
|
"UPDATE items SET updated_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') "
|
|
"WHERE key = ?",
|
|
(item_key,),
|
|
)
|
|
|
|
def add_tag(self, item_key: str, tag: str) -> None:
|
|
"""Add a single tag to an item."""
|
|
con = self._con()
|
|
row = con.execute("SELECT id FROM items WHERE key = ?", (item_key,)).fetchone()
|
|
if row is None:
|
|
raise KeyError(f"Item not found: {item_key}")
|
|
tag_id = self._ensure_tag(tag)
|
|
cur = con.execute(
|
|
"INSERT OR IGNORE INTO item_tags (item_id, tag_id) VALUES (?, ?)",
|
|
(row["id"], tag_id),
|
|
)
|
|
if cur.rowcount: # re-tagging an already-tagged item changes nothing
|
|
self._stamp(item_key)
|
|
con.commit()
|
|
|
|
def add_tags(self, item_key: str, tags: Iterable[str]) -> int:
|
|
"""Add multiple tags to an item in one transaction.
|
|
|
|
Same ``INSERT OR IGNORE`` semantics as :meth:`add_tag` — an
|
|
already-present tag is a no-op — but batched into a single
|
|
commit instead of one per tag. A backfill stamping ~486k
|
|
``code:``/``family:`` labels one-at-a-time via ``add_tag`` took
|
|
34 minutes, almost entirely commit/fsync overhead
|
|
(``bib.codetags.apply_code_tags``).
|
|
|
|
Returns the number of tags newly added (0 if every tag in
|
|
*tags* was already on the item).
|
|
"""
|
|
con = self._con()
|
|
row = con.execute("SELECT id FROM items WHERE key = ?", (item_key,)).fetchone()
|
|
if row is None:
|
|
raise KeyError(f"Item not found: {item_key}")
|
|
item_id = row["id"]
|
|
added = 0
|
|
for tag in tags:
|
|
tag_id = self._ensure_tag(tag)
|
|
cur = con.execute(
|
|
"INSERT OR IGNORE INTO item_tags (item_id, tag_id) VALUES (?, ?)",
|
|
(item_id, tag_id),
|
|
)
|
|
added += cur.rowcount
|
|
if added: # re-tagging with already-present tags only changes nothing
|
|
self._stamp(item_key)
|
|
con.commit()
|
|
return added
|
|
|
|
def merge_extra(self, item_key: str, updates: dict[str, Any]) -> None:
|
|
"""Merge *updates* into the item's ``extra_json`` (top-level keys
|
|
replace; everything else is kept). A missing item is a no-op."""
|
|
con = self._con()
|
|
row = con.execute(
|
|
"SELECT extra_json FROM items WHERE key = ?", (item_key,)
|
|
).fetchone()
|
|
if row is None:
|
|
return
|
|
current = json.loads(row[0] or "{}") if row[0] else {}
|
|
current.update(updates)
|
|
con.execute(
|
|
"UPDATE items SET extra_json = ?, "
|
|
"updated_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') WHERE key = ?",
|
|
(json.dumps(current), item_key),
|
|
)
|
|
con.commit()
|
|
|
|
def remove_tag(self, item_key: str, tag: str) -> None:
|
|
"""Remove a tag from an item."""
|
|
con = self._con()
|
|
cur = con.execute(
|
|
"""DELETE FROM item_tags
|
|
WHERE item_id = (SELECT id FROM items WHERE key = ?)
|
|
AND tag_id = (SELECT id FROM tags WHERE name = ?)""",
|
|
(item_key, tag),
|
|
)
|
|
if cur.rowcount: # the tag wasn't there — nothing changed
|
|
self._stamp(item_key)
|
|
con.commit()
|
|
|
|
def list_tags(self, *, namespace: str = "") -> list[dict[str, int]]:
|
|
"""List tags with item counts. Filter by namespace prefix."""
|
|
con = self._con()
|
|
if namespace:
|
|
rows = con.execute(
|
|
"""SELECT t.name, count(*) AS cnt
|
|
FROM tags t
|
|
JOIN item_tags it ON t.id = it.tag_id
|
|
WHERE t.name LIKE ?
|
|
GROUP BY t.name ORDER BY t.name""",
|
|
(f"{namespace}:%",),
|
|
).fetchall()
|
|
else:
|
|
rows = con.execute(
|
|
"""SELECT t.name, count(*) AS cnt
|
|
FROM tags t
|
|
JOIN item_tags it ON t.id = it.tag_id
|
|
GROUP BY t.name ORDER BY t.name"""
|
|
).fetchall()
|
|
return [{"name": r["name"], "count": r["cnt"]} for r in rows]
|
|
|
|
# ── Dockets ──────────────────────────────────────────────────
|
|
|
|
_DOCKET_COLS = (
|
|
"id",
|
|
"rule_cms_id",
|
|
"fr_document_id",
|
|
"fr_object_id",
|
|
"comment_end_date",
|
|
"pull_watermark",
|
|
"last_pull_at",
|
|
"last_pull_new",
|
|
"sealed_at",
|
|
"seal_reason",
|
|
"counts_json",
|
|
)
|
|
|
|
def _docket_from_row(self, row: sqlite3.Row | None) -> Docket | None:
|
|
if row is None:
|
|
return None
|
|
d = {c: row[c] for c in self._DOCKET_COLS}
|
|
return Docket(**d)
|
|
|
|
def docket_get(self, docket_id: str) -> Docket | None:
|
|
row = (
|
|
self._con()
|
|
.execute("SELECT * FROM dockets WHERE id = ?", (docket_id,))
|
|
.fetchone()
|
|
)
|
|
return self._docket_from_row(row)
|
|
|
|
def docket_for_rule(self, cms_rule_id: str) -> Docket | None:
|
|
row = (
|
|
self._con()
|
|
.execute(
|
|
"SELECT * FROM dockets WHERE rule_cms_id = ? ORDER BY id LIMIT 1",
|
|
(cms_rule_id,),
|
|
)
|
|
.fetchone()
|
|
)
|
|
return self._docket_from_row(row)
|
|
|
|
def docket_upsert(self, docket: Docket) -> None:
|
|
cols = ", ".join(self._DOCKET_COLS)
|
|
marks = ", ".join(f":{c}" for c in self._DOCKET_COLS)
|
|
sets = ", ".join(f"{c} = excluded.{c}" for c in self._DOCKET_COLS if c != "id")
|
|
con = self._con()
|
|
con.execute(
|
|
f"INSERT INTO dockets ({cols}) VALUES ({marks}) " # noqa: S608
|
|
f"ON CONFLICT(id) DO UPDATE SET {sets}",
|
|
{c: getattr(docket, c) for c in self._DOCKET_COLS},
|
|
)
|
|
con.commit()
|
|
|
|
def dockets(self) -> list[Docket]:
|
|
rows = self._con().execute("SELECT * FROM dockets ORDER BY id").fetchall()
|
|
return [self._docket_from_row(r) for r in rows]
|
|
|
|
def sealed_dockets(self) -> dict[str, str]:
|
|
"""docket id → sealed_at for every sealed docket."""
|
|
rows = (
|
|
self._con()
|
|
.execute("SELECT id, sealed_at FROM dockets WHERE sealed_at <> ''")
|
|
.fetchall()
|
|
)
|
|
return {r["id"]: r["sealed_at"] for r in rows}
|
|
|
|
def docket_seal(
|
|
self, docket_id: str, *, reason: str, counts: dict[str, int]
|
|
) -> None:
|
|
from datetime import datetime, timezone
|
|
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
con = self._con()
|
|
con.execute(
|
|
"UPDATE dockets SET sealed_at = ?, seal_reason = ?, counts_json = ? WHERE id = ?",
|
|
(now, reason, json.dumps(counts, sort_keys=True), docket_id),
|
|
)
|
|
con.commit()
|
|
|
|
def docket_unseal(self, docket_id: str) -> None:
|
|
con = self._con()
|
|
con.execute(
|
|
"UPDATE dockets SET sealed_at = '', seal_reason = '' WHERE id = ?",
|
|
(docket_id,),
|
|
)
|
|
con.commit()
|
|
|
|
# ── Attachments & Notes ──────────────────────────────────────
|
|
|
|
def attach_file(self, item_key: str, path: Path, *, title: str = "") -> str:
|
|
"""Copy a file into storage and link it to an item."""
|
|
con = self._con()
|
|
row = con.execute("SELECT id FROM items WHERE key = ?", (item_key,)).fetchone()
|
|
if row is None:
|
|
raise KeyError(f"Item not found: {item_key}")
|
|
|
|
filename = title or path.name
|
|
dup = con.execute(
|
|
"SELECT key FROM attachments WHERE item_id = ? AND filename = ?",
|
|
(row["id"], filename),
|
|
).fetchone()
|
|
if dup:
|
|
return dup["key"]
|
|
|
|
att_key = _generate_key()
|
|
dest_dir = self._storage / att_key
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest = dest_dir / path.name
|
|
shutil.copy2(path, dest)
|
|
|
|
content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
|
|
con.execute(
|
|
"""INSERT INTO attachments
|
|
(item_id, key, filename, content_type, storage_path)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(
|
|
row["id"],
|
|
att_key,
|
|
filename,
|
|
content_type,
|
|
# resolve(): through a symlinked storage dir (a git
|
|
# worktree's data/ pointing at the real data dir) the
|
|
# unresolved path dies with the worktree (#628).
|
|
str(dest.resolve()),
|
|
),
|
|
)
|
|
con.commit()
|
|
return att_key
|
|
|
|
def attach_note(self, item_key: str, content: str, *, title: str = "") -> int:
|
|
"""Add a note to an item. Returns the note ID."""
|
|
con = self._con()
|
|
row = con.execute("SELECT id FROM items WHERE key = ?", (item_key,)).fetchone()
|
|
if row is None:
|
|
raise KeyError(f"Item not found: {item_key}")
|
|
|
|
cur = con.execute(
|
|
"INSERT INTO notes (item_id, title, content) VALUES (?, ?, ?)",
|
|
(row["id"], title, content),
|
|
)
|
|
con.commit()
|
|
return cur.lastrowid # type: ignore[return-value]
|
|
|
|
# ── Citation formatting ──────────────────────────────────────
|
|
|
|
def format_citation(self, key: str, style: str = "apa") -> str:
|
|
"""Format a single item as a citation string."""
|
|
from bib.format import format_citation
|
|
|
|
item = self.get(key)
|
|
return format_citation(item, style=style)
|
|
|
|
def format_bibliography(self, keys: list[str], style: str = "apa") -> str:
|
|
"""Format multiple items as a bibliography."""
|
|
from bib.format import format_bibliography
|
|
|
|
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
|