"""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", "watch_provider": "watch_provider", "user": "user", } 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[a-z_]+)_(?P[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