"""Structured exact-citation lookup over the registry legal catalogue.
Exact citations ("art. 27.2 LGT", ``ley-58-2003:art-27.2``) do not go
through the FTS index: the wheel already ships typed citation data (the
registry ``legal_refs`` with their ``corpus_ref`` and BOE permalinks),
so this is a direct structured key lookup, not a parallel citation
parser. The registry legal catalogue is the single citation authority
(:data:`aeat-registry-authority-flow`); this module reuses it and adds
the one thing the product lacked at runtime: resolving a citation id to
the verbatim authoritative text its ``corpus_ref`` points at.
The verbatim text is read from the bundled ``*.extracted.json`` sidecar
(the clean, structured extraction the corpus ships) rather than the raw
HTML, and sliced to the unit the citation's anchor names when the source
file carries multiple units.
See Also:
:class:`~application.corpus_search.CitationResolution`
Typed result carrying catalogue metadata and verbatim text.
:func:`~application.corpus_search.search_corpus`
Exact citation ids short-circuit through this lookup before ranking.
:func:`~entrypoints.mcp._resources._read_corpus_resource`
``aeat://corpus`` resource reader that accepts citation ids and
retrieval ``corpus_ref`` values.
"""
from __future__ import annotations
import json
from collections.abc import Mapping
from pathlib import Path
from ...core.external_constants import UTF_8_ENCODING
from ...domain.calculations.registry import LegalReference, bundled_authority
from ._errors import CorpusSearchInputError
from ._models import CitationResolution
[docs]
class CitationLookup:
"""Resolve a registry citation id to metadata plus verbatim text.
The lookup is built over the reviewed legal catalogue and the corpus
source root (the bundled ``_data`` tree the catalogue's ``corpus_ref``
values are relative to).
"""
def __init__(self, legal: Mapping[str, LegalReference], *, source_root: Path) -> None:
self._legal = dict(legal)
self._source_root = source_root.resolve()
@property
def citation_ids(self) -> tuple[str, ...]:
"""Return every resolvable citation id, sorted."""
return tuple(sorted(self._legal))
[docs]
def resolve(self, citation_id: str) -> CitationResolution:
"""Resolve ``citation_id`` to a :class:`CitationResolution`.
Args:
citation_id: A registry ``legal_refs`` id, e.g.
``ley-58-2003:art-27.2``.
Returns:
The citation's catalogue metadata plus the verbatim
authoritative text its ``corpus_ref`` points at.
Raises:
CorpusSearchInputError: If the id is unknown to the catalogue
or its backing corpus text cannot be read.
"""
key = citation_id.strip()
reference = self._legal.get(key)
if reference is None:
raise CorpusSearchInputError(
"unknown citation id",
context={"citation_id": citation_id},
suggestion="List resolvable ids via CitationLookup.citation_ids.",
)
path_part, _, anchor_part = reference.corpus_ref.partition("#")
anchor = anchor_part or None
verbatim = self._verbatim_text(reference, path_part=path_part, anchor=anchor)
return CitationResolution(
citation_id=reference.id,
document_id=reference.document_id,
kind=reference.kind,
corpus_ref=reference.corpus_ref,
permalink=reference.permalink,
article=reference.article,
section=reference.section,
anchor=anchor,
verbatim_text=verbatim,
)
[docs]
def resolve_corpus_text(self, ref: str) -> str:
"""Resolve a citation id OR a corpus_ref (``path#anchor``) to verbatim text.
The ``aeat://corpus/{ref}`` resource accepts either form: a retrieval
hit's ``corpus_ref`` or a bare citation id. A known citation id routes
through :meth:`resolve`; otherwise ``ref`` is read as a corpus path and
anchor.
Raises:
CorpusSearchInputError: If ``ref`` resolves to no readable text or
escapes the corpus root.
"""
key = ref.strip()
if key in self._legal:
return self.resolve(key).verbatim_text
path_part, _, anchor_part = key.partition("#")
text = self._read_corpus_text(path_part, anchor=anchor_part or None)
if text is None:
raise CorpusSearchInputError("no readable corpus text for reference", context={"ref": ref})
return text
def _verbatim_text(self, reference: LegalReference, *, path_part: str, anchor: str | None) -> str:
text = self._read_corpus_text(path_part, anchor=anchor)
if text is None:
raise CorpusSearchInputError(
"citation has no readable extracted corpus text",
context={"citation_id": reference.id, "corpus_ref": reference.corpus_ref},
)
return text
def _read_corpus_text(self, path_part: str, *, anchor: str | None) -> str | None:
source_path = (self._source_root / path_part).resolve()
if self._source_root not in source_path.parents:
raise CorpusSearchInputError(
"corpus_ref escapes the corpus root",
context={"path": path_part},
)
extracted_json = source_path.with_name(source_path.name + ".extracted.json")
if extracted_json.is_file():
text = _text_from_units(extracted_json, anchor=anchor)
if text:
return text
extracted_md = source_path.with_name(source_path.name + ".extracted.md")
if extracted_md.is_file():
text = extracted_md.read_text(encoding=UTF_8_ENCODING).strip()
if text:
return text
if source_path.is_file():
text = _text_from_html(source_path)
if text:
return text
return None
def _text_from_units(extracted_json: Path, *, anchor: str | None) -> str:
payload = json.loads(extracted_json.read_text(encoding=UTF_8_ENCODING))
units = payload.get("units") or ()
texts = [(unit.get("anchor"), (unit.get("text") or "").strip()) for unit in units]
texts = [(_clean_anchor(item_anchor), text) for item_anchor, text in texts if text]
if not texts:
return ""
if anchor is not None:
matched = [text for unit_anchor, text in texts if unit_anchor == anchor]
if matched:
return "\n\n".join(matched).strip()
if len(texts) == 1:
return texts[0][1]
return "\n\n".join(text for _anchor, text in texts).strip()
def _text_from_html(html_path: Path) -> str:
"""Strip a light BOE HTML excerpt to verbatim, case-preserving text.
A minority of corpus refs point at per-article HTML files that carry
no ``*.extracted`` sidecar. These are clean BOE excerpts with light
``<h5>``/``<p>`` markup and an HTML-comment provenance header;
BeautifulSoup's ``get_text`` drops the comment and the tags while
preserving the case and accents the extracted sidecars also preserve.
"""
from bs4 import BeautifulSoup
soup = BeautifulSoup(html_path.read_text(encoding=UTF_8_ENCODING), "html.parser")
lines = [line.strip() for line in soup.get_text(separator="\n").splitlines()]
return "\n".join(line for line in lines if line).strip()
def _clean_anchor(value: object) -> str | None:
if not isinstance(value, str):
return None
cleaned = value.strip().lstrip("#")
return cleaned or None
[docs]
def bundled_citation_lookup() -> CitationLookup:
"""Return a :class:`CitationLookup` over the bundled registry catalogue."""
authority = bundled_authority()
return CitationLookup(authority.catalogues.legal, source_root=authority.source_root)
__all__ = ["CitationLookup", "bundled_citation_lookup"]