fix(bib): merge every source before capping code-tags; batch tag commits per item (refs #688)
Ruling A7: stack bib code-tags called apply_code_tags once per collection with an independent mapping, so the max-codes-per-item cap was evaluated per source instead of per item — 54 items straddling rules/corpus could dodge it in both calls. Now every selected source (rules anchors + chunks, comments chunks, corpus chunks) merges into one item_key -> (codes, families) mapping before a single apply_code_tags call. Store.add_tags batches an item's new labels into one commit instead of one commit per add_tag call; apply_code_tags uses it. Fixes the 34-minute backfill's commit-per-tag overhead. Also: noqa on the codetags._con() access, a CLI test for --anchors-only on a non-rules collection, and add_tags coverage in test_store.py.
This commit is contained in:
@@ -18,13 +18,22 @@ Two sources feed the same shape (``item_key -> (codes, families)``):
|
|||||||
``--anchors-only``).
|
``--anchors-only``).
|
||||||
|
|
||||||
:func:`apply_code_tags` stamps the tags. It is additive and idempotent
|
:func:`apply_code_tags` stamps the tags. It is additive and idempotent
|
||||||
— ``Store.add_tag`` only writes when the tag is new — and caps the
|
— ``Store.add_tags`` only writes labels that are new, one transaction
|
||||||
per-item ``code:`` cloud at ``max_codes_per_item`` distinct codes: an
|
per item — and caps the per-item ``code:`` cloud at
|
||||||
RVU file or code-list item can name thousands of codes, and a tag cloud
|
``max_codes_per_item`` distinct codes: an RVU file or code-list item
|
||||||
that size helps nobody. Wide items still get every ``family:`` tag,
|
can name thousands of codes, and a tag cloud that size helps nobody.
|
||||||
plus ``code:`` for codes in a hand-registered family
|
Wide items still get every ``family:`` tag, plus ``code:`` for codes in
|
||||||
(``pfs.families.HAND_FAMILIES`` — ACP/CCM/PCM/TCM/APCM) since those are
|
a hand-registered family (``pfs.families.HAND_FAMILIES`` —
|
||||||
the ones anyone actually filters by.
|
ACP/CCM/PCM/TCM/APCM) since those are the ones anyone actually filters
|
||||||
|
by.
|
||||||
|
|
||||||
|
The cap is evaluated against an item's *merged* code set — a caller
|
||||||
|
that sources ``mapping`` from more than one collection (e.g. an item
|
||||||
|
whose chunks live in both ``rules`` and ``corpus``) must union all
|
||||||
|
sources into one ``mapping`` before calling :func:`apply_code_tags`;
|
||||||
|
calling it once per source would cap each source's codes independently
|
||||||
|
and let a straddling item slip under the cap in each call while still
|
||||||
|
being wide overall (Ruling A7).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
@@ -84,7 +93,7 @@ def item_codes_from_anchors(
|
|||||||
) -> CodeMap:
|
) -> CodeMap:
|
||||||
"""``item_key -> (codes, families)`` for rules, computed straight from
|
"""``item_key -> (codes, families)`` for rules, computed straight from
|
||||||
``fr_anchors.text`` via ``anchor_metadata`` — no pgvector needed."""
|
``fr_anchors.text`` via ``anchor_metadata`` — no pgvector needed."""
|
||||||
con = store._con()
|
con = store._con() # noqa: SLF001
|
||||||
rows = con.execute(
|
rows = con.execute(
|
||||||
"SELECT item_key, text FROM fr_anchors ORDER BY item_key, p_id"
|
"SELECT item_key, text FROM fr_anchors ORDER BY item_key, p_id"
|
||||||
).fetchall()
|
).fetchall()
|
||||||
@@ -115,15 +124,22 @@ def apply_code_tags(
|
|||||||
"""Stamp ``code:``/``family:`` tags for every item in *mapping*.
|
"""Stamp ``code:``/``family:`` tags for every item in *mapping*.
|
||||||
|
|
||||||
Additive and idempotent: an item's current tags are read once and
|
Additive and idempotent: an item's current tags are read once and
|
||||||
only labels not already present count toward ``tags_added`` (and
|
only labels not already present count toward ``tags_added``. Under
|
||||||
are written, unless ``dry_run``) — a rerun over an unchanged mapping
|
a real run all new labels for an item are written in a single
|
||||||
adds zero tags. Items with more than ``max_codes_per_item`` distinct
|
``Store.add_tags`` call — one transaction per item, not one per
|
||||||
codes get ``family:`` tags plus ``code:`` only for hand-family codes
|
label — so a rerun over an unchanged mapping adds zero tags without
|
||||||
|
the per-tag commit cost that made the first backfill take 34
|
||||||
|
minutes. Items with more than ``max_codes_per_item`` distinct codes
|
||||||
|
get ``family:`` tags plus ``code:`` only for hand-family codes
|
||||||
(:data:`HAND_FAMILY_CODES`); such items count toward
|
(:data:`HAND_FAMILY_CODES`); such items count toward
|
||||||
``skipped_wide``. Items absent from *store* are skipped silently —
|
``skipped_wide``. Items absent from *store* are skipped silently —
|
||||||
a caller can build *mapping* from a wider scope (e.g. pgvector) than
|
a caller can build *mapping* from a wider scope (e.g. pgvector) than
|
||||||
what's currently in bib.
|
what's currently in bib.
|
||||||
|
|
||||||
|
``mapping`` must already be merged across every source an item can
|
||||||
|
appear in (see the module docstring, Ruling A7) — this function
|
||||||
|
applies the cap to whatever code set it is handed, per item, once.
|
||||||
|
|
||||||
Returns ``{"items": n, "tags_added": n, "skipped_wide": n}``.
|
Returns ``{"items": n, "tags_added": n, "skipped_wide": n}``.
|
||||||
"""
|
"""
|
||||||
items = 0
|
items = 0
|
||||||
@@ -147,11 +163,12 @@ def apply_code_tags(
|
|||||||
labels = [Tag.code(c).label for c in tag_codes] + [
|
labels = [Tag.code(c).label for c in tag_codes] + [
|
||||||
Tag.family(f).label for f in families
|
Tag.family(f).label for f in families
|
||||||
]
|
]
|
||||||
for label in labels:
|
new_labels = [label for label in labels if label not in current]
|
||||||
if label in current:
|
if not new_labels:
|
||||||
continue
|
continue
|
||||||
tags_added += 1
|
if dry_run:
|
||||||
if not dry_run:
|
tags_added += len(new_labels)
|
||||||
store.add_tag(item_key, label)
|
else:
|
||||||
|
tags_added += store.add_tags(item_key, new_labels)
|
||||||
|
|
||||||
return {"items": items, "tags_added": tags_added, "skipped_wide": skipped_wide}
|
return {"items": items, "tags_added": tags_added, "skipped_wide": skipped_wide}
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ import shutil
|
|||||||
import sqlite3
|
import sqlite3
|
||||||
from importlib import resources
|
from importlib import resources
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any, Iterable
|
||||||
|
|
||||||
from bib.dockets import Docket
|
from bib.dockets import Docket
|
||||||
from bib.item import Item
|
from bib.item import Item
|
||||||
@@ -560,6 +560,37 @@ class Store:
|
|||||||
self._stamp(item_key)
|
self._stamp(item_key)
|
||||||
con.commit()
|
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 remove_tag(self, item_key: str, tag: str) -> None:
|
def remove_tag(self, item_key: str, tag: str) -> None:
|
||||||
"""Remove a tag from an item."""
|
"""Remove a tag from an item."""
|
||||||
con = self._con()
|
con = self._con()
|
||||||
|
|||||||
@@ -944,7 +944,13 @@ def code_tags(
|
|||||||
use only that and skip pgvector for rules (and skip comments/corpus
|
use only that and skip pgvector for rules (and skip comments/corpus
|
||||||
entirely — they have no other source). Additive and idempotent
|
entirely — they have no other source). Additive and idempotent
|
||||||
(``bib.codetags.apply_code_tags``) — safe to rerun after every
|
(``bib.codetags.apply_code_tags``) — safe to rerun after every
|
||||||
``stack llm restamp``.
|
``stack llm restamp``. Every selected source is merged into one
|
||||||
|
``item_key -> (codes, families)`` mapping before tagging (Ruling
|
||||||
|
A7), so an item that straddles two collections (e.g. rules and
|
||||||
|
corpus) is capped by ``--max-codes-per-item`` once, on its combined
|
||||||
|
code set — the printed per-source lines are informational only;
|
||||||
|
``tags_added``/``skipped_wide`` in the total come from that one
|
||||||
|
merged pass.
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
|
|
||||||
@@ -977,39 +983,47 @@ def code_tags(
|
|||||||
code_index = code_family_index(FAMILIES)
|
code_index = code_family_index(FAMILIES)
|
||||||
|
|
||||||
engine = None
|
engine = None
|
||||||
totals = {"items": 0, "tags_added": 0, "skipped_wide": 0}
|
merged: dict[str, tuple[set[str], set[str]]] = {}
|
||||||
|
processed_any = False
|
||||||
|
|
||||||
|
def _merge(source: dict[str, tuple[set[str], set[str]]]) -> None:
|
||||||
|
for key, (codes, fams) in source.items():
|
||||||
|
cur_codes, cur_fams = merged.setdefault(key, (set(), set()))
|
||||||
|
merged[key] = (cur_codes | codes, cur_fams | fams)
|
||||||
|
|
||||||
for target in targets:
|
for target in targets:
|
||||||
if anchors_only and target != "rules":
|
if anchors_only and target != "rules":
|
||||||
typer.echo(f"{target}: skipped (--anchors-only)")
|
typer.echo(f"{target}: skipped (--anchors-only)")
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if target == "rules":
|
if target == "rules":
|
||||||
mapping = item_codes_from_anchors(store, code_index=code_index)
|
anchors_map = item_codes_from_anchors(store, code_index=code_index)
|
||||||
|
typer.echo(f"{target} (anchors): items={len(anchors_map)}")
|
||||||
|
_merge(anchors_map)
|
||||||
|
processed_any = True
|
||||||
if not anchors_only:
|
if not anchors_only:
|
||||||
if engine is None:
|
if engine is None:
|
||||||
engine = _engine(cfg)
|
engine = _engine(cfg)
|
||||||
for key, (codes, fams) in item_codes_from_chunks(
|
chunk_map = item_codes_from_chunks(engine, collection=target)
|
||||||
engine, collection=target
|
typer.echo(f"{target} (chunks): items={len(chunk_map)}")
|
||||||
).items():
|
_merge(chunk_map)
|
||||||
cur_codes, cur_fams = mapping.setdefault(key, (set(), set()))
|
|
||||||
mapping[key] = (cur_codes | codes, cur_fams | fams)
|
|
||||||
else:
|
else:
|
||||||
if engine is None:
|
if engine is None:
|
||||||
engine = _engine(cfg)
|
engine = _engine(cfg)
|
||||||
mapping = item_codes_from_chunks(engine, collection=target)
|
chunk_map = item_codes_from_chunks(engine, collection=target)
|
||||||
|
typer.echo(f"{target} (chunks): items={len(chunk_map)}")
|
||||||
|
_merge(chunk_map)
|
||||||
|
processed_any = True
|
||||||
|
|
||||||
|
if processed_any:
|
||||||
stats = apply_code_tags(
|
stats = apply_code_tags(
|
||||||
store, mapping, dry_run=dry_run, max_codes_per_item=max_codes_per_item
|
store, merged, dry_run=dry_run, max_codes_per_item=max_codes_per_item
|
||||||
)
|
|
||||||
for k in totals:
|
|
||||||
totals[k] += stats[k]
|
|
||||||
typer.echo(
|
|
||||||
f"{target}: items={stats['items']} tags_added={stats['tags_added']} "
|
|
||||||
f"skipped_wide={stats['skipped_wide']}"
|
|
||||||
)
|
)
|
||||||
|
else:
|
||||||
|
stats = {"items": 0, "tags_added": 0, "skipped_wide": 0}
|
||||||
|
|
||||||
prefix = "would-" if dry_run else ""
|
prefix = "would-" if dry_run else ""
|
||||||
typer.echo(
|
typer.echo(
|
||||||
f"total: {prefix}items={totals['items']} {prefix}tags_added={totals['tags_added']} "
|
f"total: {prefix}items={stats['items']} {prefix}tags_added={stats['tags_added']} "
|
||||||
f"skipped_wide={totals['skipped_wide']}"
|
f"skipped_wide={stats['skipped_wide']}"
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -418,6 +418,30 @@ class TestTags:
|
|||||||
s.add_tag("NOEXIST1", "tag")
|
s.add_tag("NOEXIST1", "tag")
|
||||||
s.close()
|
s.close()
|
||||||
|
|
||||||
|
def test_add_tags_returns_count_of_new_labels(self) -> None:
|
||||||
|
s = Store(":memory:")
|
||||||
|
key = s.create(Rule(title="Test"), tags=["existing"])
|
||||||
|
added = s.add_tags(key, ["existing", "code:99490", "family:CCM"])
|
||||||
|
assert added == 2
|
||||||
|
item = s.get(key)
|
||||||
|
assert {"existing", "code:99490", "family:CCM"} <= set(item.tags)
|
||||||
|
s.close()
|
||||||
|
|
||||||
|
def test_add_tags_rerun_returns_zero(self) -> None:
|
||||||
|
s = Store(":memory:")
|
||||||
|
key = s.create(Rule(title="Test"))
|
||||||
|
first = s.add_tags(key, ["code:99490", "family:CCM"])
|
||||||
|
second = s.add_tags(key, ["code:99490", "family:CCM"])
|
||||||
|
assert first == 2
|
||||||
|
assert second == 0
|
||||||
|
s.close()
|
||||||
|
|
||||||
|
def test_add_tags_missing_item(self) -> None:
|
||||||
|
s = Store(":memory:")
|
||||||
|
with pytest.raises(KeyError, match="Item not found"):
|
||||||
|
s.add_tags("NOEXIST1", ["tag"])
|
||||||
|
s.close()
|
||||||
|
|
||||||
def test_remove_tag(self) -> None:
|
def test_remove_tag(self) -> None:
|
||||||
s = Store(":memory:")
|
s = Store(":memory:")
|
||||||
key = s.create(Rule(title="Test"), tags=["keep", "remove"])
|
key = s.create(Rule(title="Test"), tags=["keep", "remove"])
|
||||||
|
|||||||
@@ -275,7 +275,7 @@ class TestCodeTags:
|
|||||||
@patch("llm.index._engine")
|
@patch("llm.index._engine")
|
||||||
@patch("llm.config.load")
|
@patch("llm.config.load")
|
||||||
@patch("bib.connect")
|
@patch("bib.connect")
|
||||||
def test_all_collections_run_rules_via_anchors_and_chunks(
|
def test_all_collections_merge_into_one_apply_call(
|
||||||
self,
|
self,
|
||||||
mc_connect,
|
mc_connect,
|
||||||
mc_cfg,
|
mc_cfg,
|
||||||
@@ -289,16 +289,56 @@ class TestCodeTags:
|
|||||||
):
|
):
|
||||||
mc_anchors.return_value = {"R1": ({"99490"}, {"CCM"})}
|
mc_anchors.return_value = {"R1": ({"99490"}, {"CCM"})}
|
||||||
mc_chunks.return_value = {"C1": ({"99497"}, set())}
|
mc_chunks.return_value = {"C1": ({"99497"}, set())}
|
||||||
mc_apply.return_value = {"items": 1, "tags_added": 1, "skipped_wide": 0}
|
mc_apply.return_value = {"items": 2, "tags_added": 2, "skipped_wide": 0}
|
||||||
|
|
||||||
result = runner.invoke(app, ["code-tags", "--collection", "all"])
|
result = runner.invoke(app, ["code-tags", "--collection", "all"])
|
||||||
|
|
||||||
assert result.exit_code == 0, result.output
|
assert result.exit_code == 0, result.output
|
||||||
mc_anchors.assert_called_once()
|
mc_anchors.assert_called_once()
|
||||||
# rules + comments + corpus == 3 apply_code_tags calls
|
# rules(anchors) + rules(chunks) + comments(chunks) + corpus(chunks)
|
||||||
assert mc_apply.call_count == 3
|
# all feed ONE merged mapping — exactly one apply_code_tags call
|
||||||
for _, kwargs in mc_apply.call_args_list:
|
# (Ruling A7: the cap must see an item's combined code set).
|
||||||
assert kwargs["dry_run"] is False
|
mc_apply.assert_called_once()
|
||||||
|
mapping = mc_apply.call_args.args[1]
|
||||||
|
assert mapping["R1"] == ({"99490"}, {"CCM"})
|
||||||
|
assert mapping["C1"] == ({"99497"}, set())
|
||||||
|
assert mc_apply.call_args.kwargs["dry_run"] is False
|
||||||
|
|
||||||
|
@patch("bib.codetags.apply_code_tags")
|
||||||
|
@patch("bib.codetags.item_codes_from_chunks")
|
||||||
|
@patch("bib.codetags.item_codes_from_anchors")
|
||||||
|
@patch("pfs.families.refresh_from")
|
||||||
|
@patch("pfs.anchors.code_family_index", return_value={})
|
||||||
|
@patch("os.path.exists", return_value=False)
|
||||||
|
@patch("llm.index._engine")
|
||||||
|
@patch("llm.config.load")
|
||||||
|
@patch("bib.connect")
|
||||||
|
def test_straddling_item_is_merged_before_apply(
|
||||||
|
self,
|
||||||
|
mc_connect,
|
||||||
|
mc_cfg,
|
||||||
|
mc_engine,
|
||||||
|
mc_exists,
|
||||||
|
mc_code_index,
|
||||||
|
mc_refresh,
|
||||||
|
mc_anchors,
|
||||||
|
mc_chunks,
|
||||||
|
mc_apply,
|
||||||
|
):
|
||||||
|
# Same item_key surfaces from rules-anchors and again from
|
||||||
|
# corpus-chunks (a rule item whose PDF also landed in corpus) —
|
||||||
|
# the codes must union into one mapping entry, not two capped
|
||||||
|
# separately.
|
||||||
|
mc_anchors.return_value = {"X1": ({"99490"}, {"CCM"})}
|
||||||
|
mc_chunks.return_value = {"X1": ({"99497"}, {"TCM"})}
|
||||||
|
mc_apply.return_value = {"items": 1, "tags_added": 3, "skipped_wide": 0}
|
||||||
|
|
||||||
|
result = runner.invoke(app, ["code-tags", "--collection", "all"])
|
||||||
|
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
mc_apply.assert_called_once()
|
||||||
|
mapping = mc_apply.call_args.args[1]
|
||||||
|
assert mapping["X1"] == ({"99490", "99497"}, {"CCM", "TCM"})
|
||||||
|
|
||||||
@patch("bib.codetags.apply_code_tags")
|
@patch("bib.codetags.apply_code_tags")
|
||||||
@patch("bib.codetags.item_codes_from_chunks")
|
@patch("bib.codetags.item_codes_from_chunks")
|
||||||
@@ -333,6 +373,7 @@ class TestCodeTags:
|
|||||||
assert result.exit_code == 0, result.output
|
assert result.exit_code == 0, result.output
|
||||||
assert mc_apply.call_args.kwargs["dry_run"] is True
|
assert mc_apply.call_args.kwargs["dry_run"] is True
|
||||||
store.add_tag.assert_not_called()
|
store.add_tag.assert_not_called()
|
||||||
|
store.add_tags.assert_not_called()
|
||||||
|
|
||||||
@patch("bib.codetags.apply_code_tags")
|
@patch("bib.codetags.apply_code_tags")
|
||||||
@patch("bib.codetags.item_codes_from_chunks")
|
@patch("bib.codetags.item_codes_from_chunks")
|
||||||
@@ -368,6 +409,44 @@ class TestCodeTags:
|
|||||||
# only the rules mapping gets applied — comments/corpus skipped
|
# only the rules mapping gets applied — comments/corpus skipped
|
||||||
assert mc_apply.call_count == 1
|
assert mc_apply.call_count == 1
|
||||||
|
|
||||||
|
@patch("bib.codetags.apply_code_tags")
|
||||||
|
@patch("bib.codetags.item_codes_from_chunks")
|
||||||
|
@patch("bib.codetags.item_codes_from_anchors")
|
||||||
|
@patch("pfs.families.refresh_from")
|
||||||
|
@patch("pfs.anchors.code_family_index", return_value={})
|
||||||
|
@patch("os.path.exists", return_value=False)
|
||||||
|
@patch("llm.index._engine")
|
||||||
|
@patch("llm.config.load")
|
||||||
|
@patch("bib.connect")
|
||||||
|
def test_anchors_only_with_non_rules_collection_skips_everything(
|
||||||
|
self,
|
||||||
|
mc_connect,
|
||||||
|
mc_cfg,
|
||||||
|
mc_engine,
|
||||||
|
mc_exists,
|
||||||
|
mc_code_index,
|
||||||
|
mc_refresh,
|
||||||
|
mc_anchors,
|
||||||
|
mc_chunks,
|
||||||
|
mc_apply,
|
||||||
|
):
|
||||||
|
store = MagicMock()
|
||||||
|
mc_connect.return_value = store
|
||||||
|
|
||||||
|
result = runner.invoke(
|
||||||
|
app, ["code-tags", "--collection", "comments", "--anchors-only"]
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
assert "comments: skipped (--anchors-only)" in result.output
|
||||||
|
assert "total: items=0 tags_added=0 skipped_wide=0" in result.output
|
||||||
|
mc_anchors.assert_not_called()
|
||||||
|
mc_chunks.assert_not_called()
|
||||||
|
mc_apply.assert_not_called()
|
||||||
|
mc_engine.assert_not_called()
|
||||||
|
store.add_tag.assert_not_called()
|
||||||
|
store.add_tags.assert_not_called()
|
||||||
|
|
||||||
def test_help(self):
|
def test_help(self):
|
||||||
assert runner.invoke(app, ["code-tags", "--help"]).exit_code == 0
|
assert runner.invoke(app, ["code-tags", "--help"]).exit_code == 0
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user