spb/company-atlas

1"""Event clustering (spec §23): the same corporate event seen on several surfaces (newsroom + feed + homepage, leadership + about…)2is folded into one canonical event with aggregated sources. Corroboration raises confidence; later duplicates keep their own row3(`status='duplicate'`, `cluster_id`) so provenance is never lost.45    cluster_key = sha(company_id, event_subtype, normalised entity key, 7-day window bucket)67Only `services/events.py` and `services/llm/enrich.py` call `attach_to_cluster`; the API reads `event_clusters` / `events.cluster_id`.8"""9from __future__ import annotations1011import re12import unicodedata13from datetime import UTC, datetime14from typing import Any1516from sqlalchemy.ext.asyncio import AsyncConnection1718from companyatlas.db import execute, fetch_one, jsonb19from companyatlas.ids import new_id, stable_hash2021CLUSTER_WINDOW_DAYS = 722CORROBORATION_BONUS = 0.03          # confidence bump per extra corroborating surface23CONFIDENCE_CAP = 0.992425_non_alnum = re.compile(r"[^a-z0-9]+")262728def normalize_entity_key(value: str) -> str:29    """Lowercase ASCII, punctuation collapsed — makes "Jane Doe" / "jane  doe" / "Jane DOE," the same entity across surfaces."""30    s = unicodedata.normalize("NFKD", value or "").encode("ascii", "ignore").decode("ascii").lower()31    s = _non_alnum.sub(" ", s).strip()32    return re.sub(r"\s+", " ", s)[:160]333435def window_bucket(at: datetime, *, days: int = CLUSTER_WINDOW_DAYS) -> int:36    if at.tzinfo is None:37        at = at.replace(tzinfo=UTC)38    return int(at.timestamp() // (days * 86400))394041def cluster_key_for(company_id: str, event_subtype: str, entity_key: str, detected_at: datetime) -> str:42    return stable_hash(company_id, event_subtype, normalize_entity_key(entity_key), str(window_bucket(detected_at)), length=32)434445async def attach_to_cluster(conn: AsyncConnection, event: dict[str, Any], *, entity_key: str) -> tuple[str, bool]:46    """Attach a freshly inserted event to its cluster. Returns (cluster_id, is_duplicate).4748    First event of a cluster becomes canonical. A later event from another sensor/surface is marked `duplicate`, the canonical49    event gains a corroboration source and a confidence bump. Re-processing the same change (same dedupe key) never reaches here.50    """51    key = cluster_key_for(event["company_id"], event["event_subtype"], entity_key, event["detected_at"])52    existing = await fetch_one(conn, "select * from event_clusters where cluster_key = :k for update", k=key)53    if existing is None:54        cluster_id = new_id("cluster")55        await execute(conn, """56            insert into event_clusters (id, company_id, cluster_key, event_type, event_subtype, title, first_detected_at, last_detected_at,57                                        source_count, surfaces, confidence, canonical_event_id)58            values (:id, :company_id, :key, :event_type, :event_subtype, :title, :at, :at, 1, cast(:surfaces as text[]), :confidence, :event_id)59            on conflict (cluster_key) do nothing""",60            id=cluster_id, company_id=event["company_id"], key=key, event_type=event["event_type"], event_subtype=event["event_subtype"],61            title=event["title"], at=event["detected_at"], surfaces=[event.get("surface") or "other"], confidence=float(event["confidence"]),62            event_id=event["id"])63        await execute(conn, "update events set cluster_id = :c where id = :e", c=cluster_id, e=event["id"])64        return cluster_id, False6566    cluster_id = existing["id"]67    canonical_id = existing["canonical_event_id"]68    if canonical_id == event["id"]:69        return cluster_id, False70    surfaces: list[str] = list(existing["surfaces"] or [])71    surface = event.get("surface") or "other"72    same_source = surface in surfaces and event.get("sensor_id") is not None and await fetch_one(73        conn, "select 1 from events where id = :c and sensor_id = :s", c=canonical_id, s=event["sensor_id"]) is not None74    if surface not in surfaces:75        surfaces.append(surface)76    extra_surfaces = max(0, len(surfaces) - 1)77    canonical = await fetch_one(conn, "select confidence, payload from events where id = :id", id=canonical_id)78    base_conf = float(canonical["confidence"]) if canonical else float(existing["confidence"])79    new_conf = min(CONFIDENCE_CAP, max(base_conf, float(event["confidence"])) + CORROBORATION_BONUS * extra_surfaces)8081    await execute(conn, """82        update event_clusters set source_count = source_count + 1, surfaces = cast(:surfaces as text[]), confidence = :conf,83               last_detected_at = greatest(last_detected_at, cast(:at as timestamptz)) where id = :id""",84        surfaces=surfaces, conf=new_conf, at=event["detected_at"], id=cluster_id)85    await execute(conn, "update events set cluster_id = :c, status = 'duplicate' where id = :e", c=cluster_id, e=event["id"])8687    payload = dict((canonical or {}).get("payload") or {})88    sources = list(payload.get("sources") or [])89    src = {"event_id": event["id"], "sensor_id": event.get("sensor_id"), "surface": surface, "source_url": event.get("source_url"),90           "detected_at": event["detected_at"].isoformat() if isinstance(event["detected_at"], datetime) else str(event["detected_at"])}91    if src not in sources:92        sources.append(src)93    payload["sources"] = sources[:50]94    payload["corroborations"] = len(sources)95    await execute(conn, """96        update events set confidence = :conf, confidence_label = :label, payload = cast(:payload as jsonb) where id = :id""",97        conf=new_conf, label=_label(new_conf), payload=jsonb(payload), id=canonical_id)98    if event.get("source_url") and not same_source:99        await execute(conn, """100            insert into event_sources (event_id, sensor_id, source_url, snapshot_id, surface, detected_at, kind)101            values (:event_id, :sensor_id, :url, :snap, :surface, :at, 'corroboration') on conflict do nothing""",102            event_id=canonical_id, sensor_id=event.get("sensor_id"), url=event["source_url"], snap=event.get("snapshot_after"),103            surface=surface, at=event["detected_at"])104    return cluster_id, True105106107def _label(confidence: float) -> str:108    from companyatlas.taxonomy import confidence_label109110    return confidence_label(confidence)111112113__all__ = ["CLUSTER_WINDOW_DAYS", "CORROBORATION_BONUS", "attach_to_cluster", "cluster_key_for", "normalize_entity_key", "window_bucket"]114