Aurora-Core/app/core/ids.py

"""Aurora's own identity system.

Aurora Core owns its entity identities. Provider identifiers (TMDb, IMDb,
TVDb, ...) are stored as external references, never as primary keys — see
`app.models.external_id`.

Every Aurora entity has:

- a UUIDv7 database primary key (time-ordered, efficient as a btree index,
  generated with :func:`new_uuid7`)
- a public "Aurora ID" string derived from that same UUID, e.g.
  ``aur_movie_1jz8f2n3k4m5p6q7r8s9t0v1w2``, produced by :func:`aurora_id`

Because the public ID is a direct encoding of the UUIDv7 primary key, the two
never drift apart, and Aurora IDs sort lexicographically in creation order —
useful for cursors and debugging without leaking a sequential integer.
"""

from __future__ import annotations

import re

from uuid6 import uuid7

# Crockford base32: excludes I, L, O, U to avoid visual ambiguity in IDs.
_CROCKFORD_ALPHABET = "0123456789abcdefghjkmnpqrstvwxyz"
_CROCKFORD_INDEX = {ch: i for i, ch in enumerate(_CROCKFORD_ALPHABET)}

# Aurora entity type -> ID prefix. Keep stable once shipped; these are
# effectively part of the public API contract.
ENTITY_PREFIXES: dict[str, str] = {
    "movie": "movie",
    "series": "series",
    "season": "season",
    "episode": "episode",
    "person": "person",
    "credit": "credit",
    "company": "company",
    "collection": "collection",
    "genre": "genre",
    "keyword": "keyword",
    "artwork": "art",
    "video": "vid",
    "external_id": "ext",
    "provider": "provider",
    "provider_record": "prec",
    "ingestion_job": "job",
    "certification": "cert",
    "language": "lang",
    "country": "country",
    "alternate_title": "altt",
    "release": "rel",
}


def new_uuid7() -> str:
    """Return a fresh UUIDv7 as its canonical hyphenated string form."""

    return str(uuid7())


def _encode_uuid(uuid_str: str) -> str:
    """Encode a UUID's 128 bits as lowercase Crockford base32 (26 chars).

    Byte-order-preserving: since the UUID is time-ordered (v7) and this
    encoding processes bytes most-significant-first, the resulting strings
    sort lexicographically the same way the UUIDs (and thus creation time)
    do.
    """

    raw = bytes.fromhex(uuid_str.replace("-", ""))
    value = int.from_bytes(raw, "big")
    chars: list[str] = []
    # 128 bits / 5 bits-per-symbol = 25.6 -> 26 symbols, top symbol uses 3 bits.
    for shift in range(125, -5, -5):
        chars.append(_CROCKFORD_ALPHABET[(value >> shift) & 0b11111])
    return "".join(chars)


def _decode_to_uuid(encoded: str) -> str:
    if len(encoded) != 26:
        raise ValueError(f"invalid Aurora ID segment length: {encoded!r}")
    value = 0
    for ch in encoded:
        try:
            value = (value << 5) | _CROCKFORD_INDEX[ch]
        except KeyError as exc:
            raise ValueError(f"invalid Aurora ID character: {ch!r}") from exc
    try:
        raw = value.to_bytes(16, "big")
    except OverflowError as exc:
        raise ValueError(f"invalid Aurora ID segment: {encoded!r}") from exc
    hex_str = raw.hex()
    return f"{hex_str[0:8]}-{hex_str[8:12]}-{hex_str[12:16]}-{hex_str[16:20]}-{hex_str[20:32]}"


def aurora_id(entity_type: str, uuid_str: str) -> str:
    """Build a public Aurora ID string from an entity type and UUIDv7 PK."""

    prefix = ENTITY_PREFIXES.get(entity_type)
    if prefix is None:
        raise ValueError(f"unknown Aurora entity type: {entity_type!r}")
    return f"aur_{prefix}_{_encode_uuid(uuid_str)}"


_AURORA_ID_RE = re.compile(r"^aur_(?P<prefix>[a-z_]+)_(?P<encoded>[0-9a-hjkmnp-tv-z]{26})$")

_PREFIX_TO_ENTITY: dict[str, str] = {v: k for k, v in ENTITY_PREFIXES.items()}


def parse_aurora_id(aid: str, expected_type: str | None = None) -> str:
    """Validate an Aurora ID and return the underlying UUID string.

    Raises ValueError if the ID is malformed or (when `expected_type` is
    given) does not belong to that entity type.
    """

    match = _AURORA_ID_RE.match(aid)
    if not match:
        raise ValueError(f"malformed Aurora ID: {aid!r}")
    prefix = match.group("prefix")
    entity_type = _PREFIX_TO_ENTITY.get(prefix)
    if entity_type is None:
        raise ValueError(f"unknown Aurora ID prefix: {prefix!r}")
    if expected_type is not None and entity_type != expected_type:
        raise ValueError(f"expected Aurora ID of type {expected_type!r}, got {entity_type!r}")
    return _decode_to_uuid(match.group("encoded"))


def is_valid_aurora_id(aid: str, expected_type: str | None = None) -> bool:
    try:
        parse_aurora_id(aid, expected_type=expected_type)
    except ValueError:
        return False
    return True