feat(identifier-layer): step 3 — doc_identifier table + ingest extractor

videscar · claude · videscar · commit 604ebfaaa66a · 2026-07-14T12:12:37.000Z
New doc_identifier table (document_id, id_kind, id_key, raw, source) with a
(id_kind, id_key) lookup index. api/identifiers.py extracts three classes at
ingest — code (letter-prefixed slash expedient codes + \d{2}I\d{3,4} body
codes), bdns (body), and norm (the doc's own numbered identity, subsuming
norm-pin's title ILIKE). person stays lexical-only; ref stays on its column.

Backfill script mirrors build_doc_references and is wired into ingest_pipeline
best-effort. Full-corpus backfill (51,637 docs, ~15s): code 7,161/2,695,
bdns 3,891/1,807, norm 1,424/665. Compact code regex tightened to the I-shape
after the first pass caught budget-line codes (08R09/28E050) — not identifiers.

Acid test: an exact (id_kind, id_key) lookup returns precisely each probe's gold
docs (match-count &lt;=2, the es/va pair). No retrieval path changed yet, so the
probe set is unchanged at 8/12; the query lane consuming this table is step 4.

Full suite 244 passed. Table/backfill are additive and unread by prod code.

Co-Authored-By: Claude Opus 4.8 &lt;noreply@anthropic.com&gt;
diff --git a/api/identifiers.py b/api/identifiers.py
@@ -0,0 +1,165 @@
+"""Structured-identifier extraction for the identifier layer.
+
+The retrieval pipeline matches through lexical BM25 and semantic embeddings, and
+both are structurally blind to a structured identifier (a code like
+``GACUJIMA/2025/36`` is one atomic tsvector token; embeddings cannot tell twin
+extracts apart). This module extracts those identifiers at ingest into the
+``doc_identifier`` table (see sql/2026-07-doc-identifier.sql) so the query side
+can look them up exactly (see the query lane, step 4).
+
+Classes extracted here:
+- ``code``: letter-prefixed slash expedient/project codes (``GACUJIMA/2025/36``,
+  ``ERESAR/2026/39R07/0008``) and compact body project codes (``24I636``).
+- ``bdns``: BDNS subsidy database IDs (``BDNS 895054``), body-only.
+- ``norm``: the document's OWN norm identity (``DECRETO 74/2026``), read off the
+  start of its title — this is what norm-pin resolves today by title ILIKE;
+  storing it lets step 4 migrate norm-pin onto an exact lookup.
+
+Deliberately NOT extracted:
+- ``person`` — names are lexically matchable and already retrieve correctly
+  (probes IP06/07/08 pass); pinning on common names carries the highest
+  false-pin risk, so names stay lexical-only.
+- ``ref`` — the DOGV publication ref is already a first-class, uniquely-indexed
+  column on ``dogv_documents``; the query lane reads that column directly rather
+  than duplicating it here.
+
+Normalization is centralized (``normalize_code`` / ``normalize_norm_key``) so the
+ingest side and the query-side detector produce identical keys — that identity
+is the whole point of the exact-match lane.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+from .doc_references import _self_identity
+
+# Title + the first slice of the body carry the high-precision identifiers; deep
+# body text is boilerplate/annex noise. Matches doc_references._BODY_SCAN_CHARS.
+_BODY_SCAN_CHARS = 3000
+
+# Letter-prefixed slash code: a >=3-letter acronym followed by 1..4 "/group"
+# segments (GACUJIMA/2025/36, ERESAR/2026/39R07/0008). Precision guard applied
+# after the match: at least one segment must be a 4-digit year, which rejects
+# ordinary prose slashes ("y/o", "km/h", "and/or") that lack a year.
+_SLASH_CODE_RE = re.compile(r"\b([A-Za-z]{3,}(?:/[A-Za-z0-9]+){1,4})\b")
+_YEAR_SEGMENT_RE = re.compile(r"^(?:19|20)\d{2}$")
+
+# Compact body project ("I" = investigación) code: 2-digit year + 'I' + 3-4
+# digits ("codi 25I656", "24I636"). Restricted to the 'I' letter on purpose:
+# the broader \d{2}[A-Za-z]\d+ shape also matches budget-line codes (08R09,
+# 28E050) which are not document identifiers and pollute the class.
+_COMPACT_CODE_RE = re.compile(r"\b(\d{2}[Ii]\d{3,4})\b")
+
+# BDNS subsidy database identifier ("BDNS 895054", "BDNS (Identif.): 895054").
+_BDNS_RE = re.compile(r"\bBDNS\b[^0-9]{0,20}?(\d{5,7})\b", re.IGNORECASE)
+
+
+@dataclass(frozen=True)
+class ExtractedIdentifier:
+    id_kind: str  # 'code' | 'bdns' | 'norm'
+    id_key: str  # normalized lookup key
+    raw: str  # as it appeared in the text
+    source: str  # 'title' | 'body'
+
+    @property
+    def dedup_key(self) -> tuple[str, str]:
+        return (self.id_kind, self.id_key)
+
+
+def normalize_code(raw: str) -> str:
+    """Canonical key for a slash/compact code: lowercase, separators collapsed to
+    a single '/'. The query side normalizes spaced/dashed renderings
+    ('gacujima 2025 36', 'GACUJIMA-2025-36') to this same key."""
+    parts = [p for p in re.split(r"[\s/\-]+", raw.strip()) if p]
+    return "/".join(parts).lower()
+
+
+def normalize_norm_key(tipo: str, numero: int, anyo: int) -> str:
+    """Canonical key for a numbered norm-ref: '<tipo>/<numero>/<anyo>' lowercased.
+    tipo is a canonical tipo key from dogv_resolver._TIPO_TITLE_PATTERNS."""
+    return f"{tipo}/{numero}/{anyo}".lower()
+
+
+def _slash_codes(text: str, source: str) -> list[ExtractedIdentifier]:
+    out: list[ExtractedIdentifier] = []
+    for m in _SLASH_CODE_RE.finditer(text):
+        token = m.group(1)
+        segments = token.split("/")
+        # Precision guard: a real expedient code carries a 4-digit year segment.
+        if not any(_YEAR_SEGMENT_RE.match(seg) for seg in segments):
+            continue
+        # Reject a plain two-number norm-ref (39/2015) that happens to have a
+        # letter word glued by the tokenizer — those are handled by the norm
+        # class, not code. A code has a genuine letter-only lead segment.
+        if not re.match(r"^[A-Za-z]{3,}$", segments[0]):
+            continue
+        out.append(
+            ExtractedIdentifier(
+                id_kind="code", id_key=normalize_code(token), raw=token, source=source
+            )
+        )
+    return out
+
+
+def _compact_codes(text: str, source: str) -> list[ExtractedIdentifier]:
+    out: list[ExtractedIdentifier] = []
+    for m in _COMPACT_CODE_RE.finditer(text):
+        token = m.group(1)
+        out.append(
+            ExtractedIdentifier(
+                id_kind="code", id_key=token.lower(), raw=token, source=source
+            )
+        )
+    return out
+
+
+def _bdns_ids(text: str, source: str) -> list[ExtractedIdentifier]:
+    out: list[ExtractedIdentifier] = []
+    for m in _BDNS_RE.finditer(text):
+        digits = m.group(1)
+        out.append(
+            ExtractedIdentifier(
+                id_kind="bdns", id_key=digits, raw=m.group(0).strip(), source=source
+            )
+        )
+    return out
+
+
+def _norm_identity(title: str) -> list[ExtractedIdentifier]:
+    """The document's own numbered norm identity (DECRETO 74/2026), read off the
+    start of its title. Date-only Resoluciones (no numero) are not pinned here."""
+    self_id = _self_identity(title or "")
+    if self_id.tipo is None or self_id.numero is None or self_id.anyo is None:
+        return []
+    key = normalize_norm_key(self_id.tipo, self_id.numero, self_id.anyo)
+    raw = f"{self_id.tipo} {self_id.numero}/{self_id.anyo}"
+    return [ExtractedIdentifier(id_kind="norm", id_key=key, raw=raw, source="title")]
+
+
+def extract_doc_identifiers(title: str, body: str | None) -> list[ExtractedIdentifier]:
+    """All identifiers found in ``title`` (highest precision) then the first
+    ``_BODY_SCAN_CHARS`` of ``body``. Deduplicated on (id_kind, id_key),
+    keeping the first (title-sourced) occurrence."""
+    title = title or ""
+    ids: list[ExtractedIdentifier] = []
+
+    ids.extend(_slash_codes(title, "title"))
+    ids.extend(_compact_codes(title, "title"))
+    ids.extend(_norm_identity(title))
+
+    if body:
+        snippet = body[:_BODY_SCAN_CHARS]
+        ids.extend(_slash_codes(snippet, "body"))
+        ids.extend(_compact_codes(snippet, "body"))
+        ids.extend(_bdns_ids(snippet, "body"))
+
+    seen: set[tuple[str, str]] = set()
+    deduped: list[ExtractedIdentifier] = []
+    for ident in ids:
+        if ident.dedup_key in seen:
+            continue
+        seen.add(ident.dedup_key)
+        deduped.append(ident)
+    return deduped
diff --git a/api/ingest_pipeline.py b/api/ingest_pipeline.py
@@ -8,6 +8,7 @@
 from api.config import get_settings
 from api.db import SessionLocal
 from scripts.build_chunks import build_chunks_for_range
+from scripts.build_doc_identifiers import build_identifiers_for_range
 from scripts.build_doc_references import build_references_for_range
 from scripts.classify_documents import classify_range
 from scripts.extract_documents import process_issue
@@ -96,5 +97,11 @@ def run_pipeline(
             build_references_for_range(db, start_date, end_date)
         except Exception:
             logger.exception("doc_reference.range_failed start=%s end=%s", start_date, end_date)
+
+        # Structured-identifier extraction (identifier layer): also best-effort.
+        try:
+            build_identifiers_for_range(db, start_date, end_date)
+        except Exception:
+            logger.exception("doc_identifier.range_failed start=%s end=%s", start_date, end_date)
     finally:
         db.close()
diff --git a/docs/identifier_layer_design.md b/docs/identifier_layer_design.md
@@ -182,10 +182,26 @@ year-trapped population, not merely preparatory.
 
 1. Build the identifier-probe set + baseline it on current prod. *(done — 6/12, above)*
 2. Intent year-guard (smallest, independent win). *(done — 8/12, `api/intent.py` `_complex_code_spans`; not yet deployed to prod)*
-3. Ingest extractor + `doc_identifier` table.
+3. Ingest extractor + `doc_identifier` table. *(done — table + `api/identifiers.py` + backfill; see below)*
 4. Query lane + migrate norm-pin onto it.
 5. Fold in code / bdns / person classes (person stays lexical-only — already 3/3; not pinned).
 
+### Step 3 result (dev DB backfill, 2026-07-14)
+
+`doc_identifier` created (`sql/2026-07-doc-identifier.sql`) and backfilled over all
+51,637 docs in ~15 s (`scripts/build_doc_identifiers.py`, wired into `ingest_pipeline`
+best-effort like `doc_reference`). Classes: **code 7,161 rows / 2,695 keys** (2,542 slash
+expedient codes + 153 `I`-shape body codes), **bdns 3,891 / 1,807**, **norm 1,424 / 665**.
+Two calibrations from the backfill: (a) the compact body-code regex was tightened to the
+`\d{2}I\d{3,4}` shape — the broader `\d{2}[A-Za-z]\d+` was also catching budget-line codes
+(`08R09`, `28E050`) that are not document identifiers; (b) `norm` extracts the doc's *own*
+identity (1,424 ≈ the 1,450 titles that start with a numbered norm — the audit's 3,649
+counted norm-refs appearing anywhere, which is the wrong population for a pin). Acid test:
+an exact `(id_kind, id_key)` lookup returns precisely the gold docs for every probe
+(gacujima/2025/36→{90640,90686}, gacujima/2024/26→{11849,33750}, bdns 895054→{85608,88100},
+decreto/74/2026→{86851,89343}), all match-counts ≤2 (es/va pair). `ref 2026/4148` resolves
+via the existing `ref` column. No retrieval path changed yet (probe set stays 8/12).
+
 ## Open questions
 
 - Person-name matching: exact-ish on title/body, but names are common — needs a uniqueness
diff --git a/scripts/build_doc_identifiers.py b/scripts/build_doc_identifiers.py
@@ -0,0 +1,141 @@
+"""
+Backfill the doc_identifier structured-identifier table from existing corpus
+titles/bodies (see sql/2026-07-doc-identifier.sql, api/identifiers.py).
+
+Idempotent (ON CONFLICT DO NOTHING) and safe to re-run over the full corpus.
+
+Usage:
+    python scripts/build_doc_identifiers.py [--batch-size 500] [--limit N]
+"""
+
+import argparse
+import logging
+import time
+
+from sqlalchemy import text as sa_text
+from sqlalchemy.orm import load_only
+
+try:
+    from . import _path  # type: ignore  # ensures project root is on sys.path
+except ImportError:
+    import _path  # type: ignore  # noqa: F401
+
+from api.db import SessionLocal
+from api.identifiers import extract_doc_identifiers
+from api.models import DogvDocument, DogvIssue
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+logger = logging.getLogger("dogv.build_doc_identifiers")
+
+_INSERT_SQL = sa_text(
+    """
+    INSERT INTO doc_identifier (document_id, id_kind, id_key, raw, source)
+    VALUES (:document_id, :id_kind, :id_key, :raw, :source)
+    ON CONFLICT (document_id, id_kind, id_key) DO NOTHING
+    """
+)
+
+
+def build_identifiers_for_doc(db, doc: DogvDocument) -> int:
+    """Extract + insert identifiers for one document. Returns count extracted."""
+    idents = extract_doc_identifiers(doc.title or "", doc.text)
+    for ident in idents:
+        db.execute(
+            _INSERT_SQL,
+            {
+                "document_id": doc.id,
+                "id_kind": ident.id_kind,
+                "id_key": ident.id_key,
+                "raw": ident.raw,
+                "source": ident.source,
+            },
+        )
+    return len(idents)
+
+
+def build_identifiers_for_range(db, start_date=None, end_date=None) -> int:
+    """Extract+insert identifiers for documents ingested in [start_date,
+    end_date] (by issue date). Called from the ingest pipeline right after text
+    extraction; failure-isolated per document so a bad extraction never fails
+    ingest. Returns the number of documents processed."""
+    query = db.query(DogvDocument).options(
+        load_only(DogvDocument.id, DogvDocument.title, DogvDocument.text)
+    )
+    if start_date is not None or end_date is not None:
+        query = query.join(DogvIssue, DogvDocument.issue_id == DogvIssue.id)
+        if start_date is not None:
+            query = query.filter(DogvIssue.date >= start_date)
+        if end_date is not None:
+            query = query.filter(DogvIssue.date <= end_date)
+
+    processed = 0
+    for doc in query.all():
+        try:
+            build_identifiers_for_doc(db, doc)
+        except Exception:
+            logger.exception("doc_identifier.extract_failed document_id=%s", doc.id)
+            db.rollback()
+            continue
+        processed += 1
+    db.commit()
+    logger.info("doc_identifier.range_processed docs=%d", processed)
+    return processed
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--batch-size", type=int, default=500)
+    parser.add_argument("--limit", type=int, default=None, help="cap docs processed (debug)")
+    args = parser.parse_args()
+
+    db = SessionLocal()
+    start = time.monotonic()
+    docs_processed = 0
+    total_extracted = 0
+
+    try:
+        id_query = db.query(DogvDocument.id).order_by(DogvDocument.id)
+        if args.limit:
+            id_query = id_query.limit(args.limit)
+        all_ids = [row[0] for row in id_query.all()]
+
+        for i in range(0, len(all_ids), args.batch_size):
+            batch_ids = all_ids[i : i + args.batch_size]
+            docs = (
+                db.query(DogvDocument)
+                .options(load_only(DogvDocument.id, DogvDocument.title, DogvDocument.text))
+                .filter(DogvDocument.id.in_(batch_ids))
+                .all()
+            )
+            for d in docs:
+                total_extracted += build_identifiers_for_doc(db, d)
+                docs_processed += 1
+            db.commit()
+            elapsed = time.monotonic() - start
+            logger.info(
+                "progress docs=%d identifiers=%d elapsed=%.1fs",
+                docs_processed,
+                total_extracted,
+                elapsed,
+            )
+
+        kind_rows = db.execute(
+            sa_text("SELECT id_kind, count(*) FROM doc_identifier GROUP BY id_kind ORDER BY id_kind")
+        ).all()
+        kind_counts = {row[0]: row[1] for row in kind_rows}
+        distinct_keys = db.execute(
+            sa_text("SELECT id_kind, count(DISTINCT id_key) FROM doc_identifier GROUP BY id_kind")
+        ).all()
+    finally:
+        db.close()
+
+    elapsed = time.monotonic() - start
+    logger.info("=== build_doc_identifiers summary ===")
+    logger.info("docs_processed=%d elapsed=%.1fs", docs_processed, elapsed)
+    logger.info("identifiers_extracted_this_run=%d", total_extracted)
+    logger.info("rows_by_id_kind=%s", kind_counts)
+    logger.info("distinct_keys_by_id_kind=%s", dict(distinct_keys))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/sql/2026-07-doc-identifier.sql b/sql/2026-07-doc-identifier.sql
@@ -0,0 +1,42 @@
+-- Structured-identifier index for the identifier layer.
+--
+-- Lexical BM25 and semantic embeddings are both blind to structured identifiers
+-- (a code like GACUJIMA/2025/36 is one atomic tsvector token; embeddings cannot
+-- separate twin extracts). This table holds every identifier extracted from a
+-- document's title/body at ingest, so the query side can look one up EXACTLY and
+-- hard-pin the document (see api/identifiers.py, and the query lane).
+--
+-- id_kind:
+--   'code' — letter-prefixed slash expedient/project codes (GACUJIMA/2025/36,
+--            ERESAR/2026/39R07/0008) and compact body codes (24I636)
+--   'bdns' — BDNS subsidy database IDs (895054), body-only
+--   'norm' — the document's own numbered norm identity (DECRETO 74/2026),
+--            read off the start of its title (subsumes norm-pin's title ILIKE)
+-- Deliberately excluded: 'person' (names stay lexical-only) and 'ref' (already a
+-- uniquely-indexed column on dogv_documents; the query lane reads it directly).
+--
+-- id_key is the normalized lookup key (lowercase, separators collapsed to '/'),
+-- produced by the same api/identifiers.normalize_* helpers on both the ingest
+-- and query sides so an exact match is possible.
+
+CREATE TABLE IF NOT EXISTS doc_identifier (
+    id            BIGSERIAL PRIMARY KEY,
+    document_id   BIGINT NOT NULL,
+    id_kind       TEXT NOT NULL,
+    id_key        TEXT NOT NULL,
+    raw           TEXT,
+    source        TEXT NOT NULL,          -- 'title' | 'body'
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+
+    CONSTRAINT fk_doc_identifier_document
+        FOREIGN KEY (document_id) REFERENCES dogv_documents(id) ON DELETE CASCADE,
+    CONSTRAINT uq_doc_identifier
+        UNIQUE (document_id, id_kind, id_key)
+);
+
+-- The exact-match query lane: SELECT document_id WHERE id_kind = ? AND id_key = ?.
+CREATE INDEX IF NOT EXISTS idx_doc_identifier_key
+  ON doc_identifier (id_kind, id_key);
+
+CREATE INDEX IF NOT EXISTS idx_doc_identifier_document
+  ON doc_identifier (document_id);
diff --git a/tests/test_identifiers.py b/tests/test_identifiers.py
