auralynq/utils.py

"""Small shared utilities: deterministic IDs, seeding, text helpers."""

from __future__ import annotations

import hashlib
import os
import random
import re
import unicodedata
from collections.abc import Iterable


def stable_id(*parts: object, length: int = 16) -> str:
    """Deterministic short hex id from arbitrary parts (idempotency keys)."""
    h = hashlib.sha256("\x1f".join(str(p) for p in parts).encode("utf-8"))
    return h.hexdigest()[:length]


def content_hash(text: str) -> str:
    """Full content hash used to detect unchanged documents (idempotent ingest)."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def seed_everything(seed: int = 42) -> None:
    """Seed Python/NumPy (and torch if present) for reproducibility."""
    os.environ.setdefault("PYTHONHASHSEED", str(seed))
    random.seed(seed)
    try:
        import numpy as np

        np.random.seed(seed)
    except ImportError:  # pragma: no cover
        pass
    try:  # pragma: no cover - torch optional
        import torch

        torch.manual_seed(seed)
    except Exception:
        pass


_WORD_RE = re.compile(r"[a-z0-9]+")


def normalize_text(text: str) -> str:
    return unicodedata.normalize("NFKC", text).strip()


def tokenize(text: str) -> list[str]:
    """Lowercase alnum tokenizer used by sparse/lexical components."""
    return _WORD_RE.findall(text.lower())


def sentence_split(text: str) -> list[str]:
    """Lightweight sentence splitter (no nltk dependency)."""
    text = normalize_text(text)
    if not text:
        return []
    # Split on sentence-ending punctuation followed by whitespace + capital/quote.
    parts = re.split(r"(?<=[.!?])\s+(?=[\"'(A-Z0-9])", text)
    return [p.strip() for p in parts if p.strip()]


def chunked(seq: list, size: int) -> Iterable[list]:
    for i in range(0, len(seq), size):
        yield seq[i : i + size]