Ingestion
Auditable corpus ingestion primitives.
"""Auditable corpus ingestion primitives.
The module deliberately separates *document identity* from *content identity*.
A logical document can have many versions, while the SHA-256 content hash is
stable across identifiers and ingestion runs. This distinction is essential
for reproducible RAG indexes: an index can record the exact manifest snapshot
from which it was built, and deletions can be represented by tombstones rather
than by silently erasing history.
Only the Python standard library is used. The near-duplicate detector uses
exact shingle Jaccard similarity; at very large scale the same interface can be
backed by MinHash/LSH without changing the manifest semantics.
"""
import hashlib
import json
import re
import unicodedata
from dataclasses import dataclass, field
from typing import Any, Dict, FrozenSet, Iterable, List, Mapping, Optional, Sequence, Tuple
from .models import Document
WORD_RE = re.compile(r"\w+(?:[-']\w+)?", re.UNICODE)
def canonicalize_text(text: str) -> str:
"""Return a conservative canonical form suitable for content identity.
Canonicalization normalizes Unicode to NFC, normalizes line endings and
non-breaking spaces, removes horizontal whitespace noise, and collapses
runs of blank lines. It intentionally preserves case and paragraph
boundaries because those can carry meaning in source documents.
"""
normalized = unicodedata.normalize("NFC", text).replace("\r\n", "\n").replace("\r", "\n")
normalized = normalized.replace("\u00a0", " ")
lines = [re.sub(r"[\t ]+", " ", line).strip() for line in normalized.split("\n")]
paragraphs: List[str] = []
current: List[str] = []
for line in lines:
if line:
current.append(line)
elif current:
paragraphs.append("\n".join(current))
current = []
if current:
paragraphs.append("\n".join(current))
return "\n\n".join(paragraphs)
def content_hash(text: str) -> str:
"""Return the hexadecimal SHA-256 digest of canonicalized UTF-8 text."""
return hashlib.sha256(canonicalize_text(text).encode("utf-8")).hexdigest()
def token_shingles(text: str, width: int = 5) -> FrozenSet[Tuple[str, ...]]:
"""Create lowercase word shingles for transparent near-duplicate tests."""
if width <= 0:
raise ValueError("shingle width must be positive")
tokens = [match.group(0).casefold() for match in WORD_RE.finditer(canonicalize_text(text))]
if not tokens:
return frozenset()
if len(tokens) < width:
return frozenset((tuple(tokens),))
return frozenset(
tuple(tokens[offset : offset + width])
for offset in range(len(tokens) - width + 1)
)
def shingle_jaccard(left: Iterable[Tuple[str, ...]], right: Iterable[Tuple[str, ...]]) -> float:
"""Compute set Jaccard similarity, defining two empty sets as identical."""
left_set, right_set = frozenset(left), frozenset(right)
union = left_set | right_set
return len(left_set & right_set) / len(union) if union else 1.0
@dataclass(frozen=True)
class DuplicateMatch:
"""The most similar already-seen corpus item."""
document_id: str
similarity: float
class ExactDeduplicator:
"""Track the first logical owner of each canonical content hash."""
def __init__(self) -> None:
self._owners: Dict[str, str] = {}
def find(self, text: str) -> Optional[str]:
"""Return the first identifier with byte-identical canonical content."""
return self._owners.get(content_hash(text))
def add(self, document_id: str, text: str) -> Optional[str]:
"""Record content and return its prior owner, if one exists."""
digest = content_hash(text)
prior = self._owners.get(digest)
if prior is None:
self._owners[digest] = document_id
return prior
class NearDuplicateIndex:
"""Exact shingle-Jaccard near-duplicate index.
This implementation scans stored signatures and is therefore a correctness
baseline. Production systems can compare an LSH candidate generator
against it to measure candidate recall.
"""
def __init__(self, threshold: float = 0.82, shingle_width: int = 5) -> None:
if not 0.0 <= threshold <= 1.0:
raise ValueError("threshold must be between zero and one")
if shingle_width <= 0:
raise ValueError("shingle_width must be positive")
self.threshold = threshold
self.shingle_width = shingle_width
self._signatures: Dict[str, FrozenSet[Tuple[str, ...]]] = {}
def find(self, text: str, exclude: Optional[str] = None) -> Optional[DuplicateMatch]:
"""Return the best qualifying match, breaking ties by document id."""
signature = token_shingles(text, self.shingle_width)
candidates = [
DuplicateMatch(document_id, shingle_jaccard(signature, other))
for document_id, other in self._signatures.items()
if document_id != exclude
]
candidates = [match for match in candidates if match.similarity >= self.threshold]
if not candidates:
return None
return min(candidates, key=lambda match: (-match.similarity, match.document_id))
def add(self, document_id: str, text: str) -> None:
"""Add or replace the signature associated with a logical document."""
self._signatures[document_id] = token_shingles(text, self.shingle_width)
def remove(self, document_id: str) -> None:
"""Remove a document from future comparisons without altering history."""
self._signatures.pop(document_id, None)
@dataclass(frozen=True)
class ACLPolicy:
"""Simple allow/deny principal policy with deny taking precedence.
``"*"`` denotes every principal. An empty allow list is private, whereas
:meth:`public` explicitly allows everyone. Callers should pass all of a
user's principals (for example user id plus group ids) to :meth:`allows`.
"""
allow: Tuple[str, ...] = ()
deny: Tuple[str, ...] = ()
@classmethod
def public(cls) -> "ACLPolicy":
"""Construct a policy readable by every principal."""
return cls(allow=("*",))
@classmethod
def restricted(cls, principals: Sequence[str]) -> "ACLPolicy":
"""Construct a policy readable only by the supplied principals."""
return cls(allow=tuple(sorted(set(principals))))
def allows(self, principals: Iterable[str]) -> bool:
"""Return whether any supplied principal is allowed and none denied."""
identities = set(principals)
if "*" in self.deny or identities.intersection(self.deny):
return False
return "*" in self.allow or bool(identities.intersection(self.allow))
@dataclass(frozen=True)
class ManifestEntry:
"""Immutable state of one logical document version."""
document_id: str
version: int
content_hash: str
canonical_text: str
title: str = ""
source: str = ""
date: str = ""
metadata: Mapping[str, Any] = field(default_factory=dict)
acl: ACLPolicy = field(default_factory=ACLPolicy.public)
previous_content_hash: Optional[str] = None
tombstoned: bool = False
tombstone_reason: str = ""
def to_document(self) -> Document:
"""Reconstruct the indexable document, rejecting tombstones."""
if self.tombstoned:
raise ValueError("a tombstone has no indexable document")
return Document(
id=self.document_id,
text=self.canonical_text,
title=self.title,
source=self.source,
date=self.date,
metadata=self.metadata,
)
@dataclass(frozen=True)
class IngestionResult:
"""Decision produced by :class:`CorpusManifest.ingest`."""
action: str
entry: Optional[ManifestEntry]
duplicate_of: Optional[str] = None
similarity: float = 0.0
class CorpusManifest:
"""Versioned, ACL-aware corpus manifest with deduplication decisions."""
def __init__(self, near_duplicate_threshold: float = 0.82, shingle_width: int = 5) -> None:
self._history: Dict[str, List[ManifestEntry]] = {}
self._exact = ExactDeduplicator()
self._near = NearDuplicateIndex(near_duplicate_threshold, shingle_width)
def ingest(
self,
document: Document,
acl: Optional[ACLPolicy] = None,
accept_near_duplicate: bool = False,
) -> IngestionResult:
"""Validate, deduplicate, and append a new manifest version.
Exact duplicates under another logical id are always rejected. Near
duplicates are reported and rejected unless explicitly accepted.
Re-ingesting unchanged content and policy under the same id is
idempotent and returns ``action="unchanged"``.
"""
canonical = canonicalize_text(document.text)
if not canonical:
raise ValueError("canonical document text must be non-empty")
digest = content_hash(canonical)
policy = acl if acl is not None else ACLPolicy.public()
current = self.current(document.id)
unchanged = (
current is not None
and not current.tombstoned
and current.content_hash == digest
and current.title == document.title
and current.source == document.source
and current.date == document.date
and dict(current.metadata) == dict(document.metadata)
and current.acl == policy
)
if unchanged:
return IngestionResult("unchanged", current)
exact_owner = self._exact.find(canonical)
if exact_owner is not None and exact_owner != document.id:
return IngestionResult(
"exact_duplicate", None, duplicate_of=exact_owner, similarity=1.0
)
near = self._near.find(canonical, exclude=document.id)
if near is not None and not accept_near_duplicate:
return IngestionResult(
"near_duplicate", None, duplicate_of=near.document_id, similarity=near.similarity
)
version = 1 if current is None else current.version + 1
entry = ManifestEntry(
document_id=document.id,
version=version,
content_hash=digest,
canonical_text=canonical,
title=document.title,
source=document.source,
date=document.date,
metadata=dict(document.metadata),
acl=policy,
previous_content_hash=current.content_hash if current is not None else None,
)
self._history.setdefault(document.id, []).append(entry)
self._exact.add(document.id, canonical)
self._near.add(document.id, canonical)
return IngestionResult("created" if current is None else "updated", entry)
def tombstone(self, document_id: str, reason: str) -> ManifestEntry:
"""Append a deletion marker and remove the item from active dedup search."""
current = self.current(document_id)
if current is None:
raise KeyError(document_id)
if current.tombstoned:
return current
tombstone = ManifestEntry(
document_id=document_id,
version=current.version + 1,
content_hash=current.content_hash,
canonical_text="",
title=current.title,
source=current.source,
date=current.date,
metadata=dict(current.metadata),
acl=current.acl,
previous_content_hash=current.content_hash,
tombstoned=True,
tombstone_reason=reason,
)
self._history[document_id].append(tombstone)
self._near.remove(document_id)
return tombstone
def current(self, document_id: str) -> Optional[ManifestEntry]:
"""Return the latest state for a logical id."""
history = self._history.get(document_id)
return history[-1] if history else None
def history(self, document_id: str) -> Tuple[ManifestEntry, ...]:
"""Return every immutable version of a logical document."""
history = self._history.get(document_id)
return tuple(history) if history else ()
def active_entries(self, principals: Iterable[str] = ()) -> Tuple[ManifestEntry, ...]:
"""Return current, non-tombstoned entries visible to the principals."""
identities = tuple(principals)
entries = [history[-1] for history in self._history.values() if history]
return tuple(
sorted(
(
entry
for entry in entries
if not entry.tombstoned and entry.acl.allows(identities)
),
key=lambda entry: entry.document_id,
)
)
def snapshot_hash(self) -> str:
"""Hash the complete current manifest state for index provenance."""
state = [
{
"acl_allow": entry.acl.allow,
"acl_deny": entry.acl.deny,
"content_hash": entry.content_hash,
"document_id": entry.document_id,
"tombstoned": entry.tombstoned,
"version": entry.version,
}
for entry in sorted(
(history[-1] for history in self._history.values() if history),
key=lambda item: item.document_id,
)
]
payload = json.dumps(state, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()