"""Stable internal identifiers: prefixed ULIDs (`co_01J…`). Slugs, names and URLs change; ids never do (spec §173).""" from __future__ import annotations import hashlib import re from slugify import slugify as _slugify from ulid import ULID PREFIXES: dict[str, str] = { "company": "co", "domain": "dom", "sensor": "sen", "observation": "obs", "snapshot": "snap", "change": "chg", "event": "evt", "cluster": "cl", "job": "job", "person": "person", "product": "prod", "plan": "plan", "location": "loc", "news": "news", "relationship": "rel", "queue_job": "qj", "llm_job": "llm", "review": "rev", "failure": "fail", "crawl_run": "run", "watchlist": "wl", "alert": "alert", "api_key": "key", "signal": "sig", } def new_id(kind: str) -> str: prefix = PREFIXES.get(kind) if prefix is None: raise ValueError(f"unknown id kind {kind!r}") return f"{prefix}_{ULID()}" def kind_of(entity_id: str) -> str | None: prefix = entity_id.split("_", 1)[0] for kind, p in PREFIXES.items(): if p == prefix: return kind return None _slug_clean = re.compile(r"[^a-z0-9.-]+") def slugify(text: str, *, max_length: int = 80) -> str: """URL slug (`stripe`, `jpmorgan-chase`, `l-oreal`).""" s = _slugify(text, lowercase=True, regex_pattern=r"[^a-z0-9-]+", replacements=[("&", " and "), ("/", "-"), ("_", "-"), ("@", "-at-")]) s = _slug_clean.sub("-", s).strip("-.") return s[:max_length].rstrip("-.") or "company" def normalize_alias(text: str) -> str: """Deterministic alias key: lowercase ASCII, punctuation collapsed, legal suffixes dropped. Resolution only — never display.""" s = _slugify(text, lowercase=True, regex_pattern=r"[^a-z0-9]+") s = s.replace("-", "") for suffix in ("incorporated", "corporation", "limited", "company", "holdings", "group", "inc", "corp", "ltd", "plc", "llc", "gmbh", "ag", "sa", "nv", "bv", "se", "ab", "oyj", "asa", "spa", "srl", "kk", "co"): if s.endswith(suffix) and len(s) > len(suffix) + 2: s = s[: -len(suffix)] break return s def stable_hash(*parts: str, length: int = 24) -> str: """Deterministic key for idempotent rows (queue jobs `sensor_id + window`, job fingerprints, cluster keys).""" h = hashlib.sha256("\x1f".join(parts).encode("utf-8")).hexdigest() return h[:length] __all__ = ["PREFIXES", "kind_of", "new_id", "normalize_alias", "slugify", "stable_hash"]