spb/company-atlas

1"""Alert evaluation (spec §140–141): match new events against `alerts.condition`2`{event_types, event_subtypes, min_importance, min_confidence, company, industries, countries, tags, metrics: {activity_score: {gt: 80}}}`3and record `alert_deliveries`. Channels: `web` (row only, the UI reads it) and `webhook` (POST JSON through the shared `fetch.Fetcher`4client after the SSRF guard, signed `X-CompanyAtlas-Signature: sha256=<hmac(body, alert id)>`). Email is stored as queued (no sender yet).56`evaluate_alerts(event_ids)` runs right after event creation; `alerts-sweep` catches up every 2 min via a watermark in settings_kv, and7evaluates metric-only alerts (no event) at most once per 24 h per alert.8"""9from __future__ import annotations1011import hashlib12import hmac13import json14import logging15from datetime import UTC, datetime, timedelta16from typing import Any1718from companyatlas.config import settings19from companyatlas.db import execute, fetch_all, fetch_val, jsonb, transaction20from companyatlas.fetch import BlockedDestination, Fetcher, validate_destination_async21from companyatlas.ids import new_id22from companyatlas.services.periodic import periodic2324log = logging.getLogger(__name__)2526WATERMARK_KEY = "alerts:last_event_created_at"27METRIC_ALERT_COOLDOWN_H = 2428_OPS = {"gt": lambda v, t: v > t, "gte": lambda v, t: v >= t, "lt": lambda v, t: v < t, "lte": lambda v, t: v <= t, "eq": lambda v, t: v == t}293031def _lower_set(values: Any) -> set[str]:32    if not values:33        return set()34    if isinstance(values, str):35        values = [values]36    return {str(v).strip().lower() for v in values if str(v).strip()}373839def metrics_match(cond: dict[str, Any], metrics: dict[str, float]) -> bool:40    spec = cond.get("metrics") or {}41    if not isinstance(spec, dict) or not spec:42        return True43    for metric, rule in spec.items():44        value = metrics.get(str(metric))45        if value is None:46            return False47        if isinstance(rule, dict):48            for op, threshold in rule.items():49                fn = _OPS.get(str(op))50                try:51                    if fn is None or not fn(float(value), float(threshold)):52                        return False53                except (TypeError, ValueError):54                    return False55        else:56            try:57                if float(value) < float(rule):58                    return False59            except (TypeError, ValueError):60                return False61    return True626364def event_matches(alert: dict[str, Any], event: dict[str, Any], company: dict[str, Any], metrics: dict[str, float]) -> bool:65    """Pure matcher. `alert.company_id` pins a company; the condition narrows further."""66    cond = alert.get("condition") or {}67    if alert.get("company_id") and alert["company_id"] != event["company_id"]:68        return False69    if cond.get("company") and cond["company"] not in (company.get("slug"), company.get("id")):70        return False71    types = _lower_set(cond.get("event_types"))72    if types and str(event["event_type"]).lower() not in types and str(event["event_subtype"]).lower() not in types:73        return False74    subtypes = _lower_set(cond.get("event_subtypes"))75    if subtypes and str(event["event_subtype"]).lower() not in subtypes:76        return False77    if cond.get("min_importance") is not None and float(event.get("importance") or 0) < float(cond["min_importance"]):78        return False79    if cond.get("min_confidence") is not None and float(event.get("confidence") or 0) < float(cond["min_confidence"]):80        return False81    countries = {c.upper() for c in _lower_set(cond.get("countries"))}82    if countries and str(company.get("country") or "").upper() not in countries:83        return False84    industries = _lower_set(cond.get("industries"))85    if industries and not (industries & {str(i).lower() for i in (company.get("industries") or [])}):86        return False87    tags = _lower_set(cond.get("tags"))88    if tags and not (tags & {str(t).lower() for t in (event.get("tags") or [])}):89        return False90    return metrics_match(cond, metrics)919293def sign(body: bytes, alert_id: str) -> str:94    return "sha256=" + hmac.new(alert_id.encode("utf-8"), body, hashlib.sha256).hexdigest()959697def webhook_payload(alert: dict[str, Any], event: dict[str, Any] | None, company: dict[str, Any] | None, metrics: dict[str, float] | None = None) -> dict[str, Any]:98    return {99        "type": "event" if event else "metric",100        "alert": {"id": alert["id"], "name": alert.get("name"), "condition": alert.get("condition")},101        "company": {k: company.get(k) for k in ("id", "slug", "display_name", "canonical_domain", "country")} if company else None,102        "event": {k: event.get(k) for k in ("id", "event_type", "event_subtype", "importance", "confidence", "confidence_label", "title", "summary", "old_value",103                                           "new_value", "detected_at", "source_url", "surface", "origin", "status")} if event else None,104        "metrics": metrics or {},105        "delivered_at": datetime.now(UTC).isoformat(),106        "docs": f"{settings.site_url}/api",107    }108109110async def deliver_webhook(alert: dict[str, Any], payload: dict[str, Any]) -> tuple[str, str]:111    target = str(alert.get("target") or "").strip()112    if not target:113        return "failed", "no webhook target"114    try:115        await validate_destination_async(target)116    except BlockedDestination as exc:117        return "failed", f"blocked destination: {exc}"118    body = json.dumps(payload, default=str, ensure_ascii=False).encode("utf-8")119    headers = {"Content-Type": "application/json", "X-CompanyAtlas-Signature": sign(body, alert["id"]), "X-CompanyAtlas-Alert": alert["id"],120               "User-Agent": settings.user_agent}121    try:122        async with Fetcher(timeout_s=settings.webhook_timeout_s, http2=False, max_connections=4) as f:123            r = await f.client.post(target, content=body, headers=headers)124        if r.status_code < 300:125            return "sent", f"HTTP {r.status_code}"126        return "failed", f"HTTP {r.status_code}"127    except Exception as exc:  # noqa: BLE001128        return "failed", f"{exc.__class__.__name__}: {exc}"[:300]129130131async def _record(conn, alert: dict[str, Any], event_id: str | None, status: str, detail: str | None) -> None:  # type: ignore[no-untyped-def]132    await execute(conn, "insert into alert_deliveries (id, alert_id, event_id, channel, status, detail) values (:id, :a, :e, :ch, :st, :d)",133                  id=new_id("alert"), a=alert["id"], e=event_id, ch=alert.get("channel") or "web", st=status, d=detail)134    await execute(conn, "update alerts set last_fired_at = now() where id = :id", id=alert["id"])135136137async def evaluate_alerts(event_ids: list[str]) -> dict[str, int]:138    """Match the given events against all enabled alerts; deliver once per (alert, event)."""139    stats = {"events": 0, "matched": 0, "delivered": 0, "failed": 0}140    if not event_ids:141        return stats142    async with transaction() as conn:143        alerts = await fetch_all(conn, "select * from alerts where enabled")144        if not alerts:145            return stats146        events = await fetch_all(conn, """select e.*, co.slug, co.display_name, co.canonical_domain, co.country as company_country, co.industries as company_industries147                                          from events e join companies co on co.id = e.company_id where e.id = any(cast(:ids as text[])) and e.status in ('active', 'review')""",148                                 ids=event_ids)149        stats["events"] = len(events)150        metrics_cache: dict[str, dict[str, float]] = {}151        pending: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, float]]] = []152        for ev in events:153            company = {"id": ev["company_id"], "slug": ev["slug"], "display_name": ev["display_name"], "canonical_domain": ev["canonical_domain"],154                       "country": ev["company_country"], "industries": ev["company_industries"]}155            for alert in alerts:156                cond = alert.get("condition") or {}157                needs_metrics = bool(cond.get("metrics"))158                if needs_metrics and ev["company_id"] not in metrics_cache:159                    metrics_cache[ev["company_id"]] = {r["metric"]: float(r["value"]) for r in await fetch_all(conn, "select metric, value from metrics_current where company_id = :c", c=ev["company_id"])}160                metrics = metrics_cache.get(ev["company_id"], {})161                if not event_matches(alert, ev, company, metrics):162                    continue163                dup = await fetch_val(conn, "select 1 from alert_deliveries where alert_id = :a and event_id = :e", a=alert["id"], e=ev["id"])164                if dup:165                    continue166                stats["matched"] += 1167                pending.append((alert, ev, company, metrics))168                if (alert.get("channel") or "web") != "webhook":169                    await _record(conn, alert, ev["id"], "delivered" if alert.get("channel") in (None, "web") else "queued", None)170                    stats["delivered"] += 1171    for alert, ev, company, metrics in pending:                # network outside the transaction172        if (alert.get("channel") or "web") != "webhook":173            continue174        status, detail = await deliver_webhook(alert, webhook_payload(alert, ev, company, metrics))175        async with transaction() as conn:176            await _record(conn, alert, ev["id"], status, detail)177        stats["delivered" if status == "sent" else "failed"] += 1178    return stats179180181async def evaluate_metric_alerts() -> dict[str, int]:182    """Alerts with only a metrics condition (no event filter) fire on the current metrics, at most once per cooldown."""183    stats = {"checked": 0, "fired": 0}184    fired: list[tuple[dict[str, Any], dict[str, Any], dict[str, float]]] = []185    async with transaction() as conn:186        alerts = await fetch_all(conn, """select * from alerts where enabled and jsonb_exists(condition, 'metrics') and coalesce(condition->'event_types', 'null'::jsonb) in ('null'::jsonb, '[]'::jsonb)187                                          and (last_fired_at is null or last_fired_at < :cutoff)""", cutoff=datetime.now(UTC) - timedelta(hours=METRIC_ALERT_COOLDOWN_H))188        for alert in alerts:189            stats["checked"] += 1190            cond = alert.get("condition") or {}191            if alert.get("company_id"):192                companies = await fetch_all(conn, "select id, slug, display_name, canonical_domain, country, industries from companies where id = :id", id=alert["company_id"])193            elif cond.get("company"):194                companies = await fetch_all(conn, "select id, slug, display_name, canonical_domain, country, industries from companies where slug = :s or id = :s", s=cond["company"])195            else:196                continue                                        # metric alerts must target one company197            for company in companies:198                metrics = {r["metric"]: float(r["value"]) for r in await fetch_all(conn, "select metric, value from metrics_current where company_id = :c", c=company["id"])}199                if not metrics or not metrics_match(cond, metrics):200                    continue201                detail = json.dumps({k: metrics.get(k) for k in (cond.get("metrics") or {})})202                if (alert.get("channel") or "web") == "webhook":203                    fired.append((alert, company, metrics))204                else:205                    await _record(conn, alert, None, "delivered", detail)206                stats["fired"] += 1207    for alert, company, metrics in fired:208        status, detail = await deliver_webhook(alert, webhook_payload(alert, None, company, metrics))209        async with transaction() as conn:210            await _record(conn, alert, None, status, detail)211    return stats212213214async def sweep(*, window_s: int | None = None) -> dict[str, int]:215    """Catch-up: evaluate events created since the watermark (bounded by the sweep window)."""216    window_s = window_s or settings.alerts_sweep_window_s217    async with transaction() as conn:218        raw = await fetch_val(conn, "select value from settings_kv where key = :k", k=WATERMARK_KEY)219        since = datetime.fromisoformat(str(raw).strip('"')) if raw else datetime.now(UTC) - timedelta(seconds=window_s)220        since = max(since, datetime.now(UTC) - timedelta(seconds=window_s))221        rows = await fetch_all(conn, "select id, created_at from events where created_at > :since and status in ('active', 'review') order by created_at limit 2000", since=since)222        if rows:223            await execute(conn, "insert into settings_kv (key, value, updated_at) values (:k, cast(:v as jsonb), now()) on conflict (key) do update set value = excluded.value, updated_at = now()",224                          k=WATERMARK_KEY, v=jsonb(rows[-1]["created_at"].isoformat()))225    stats = await evaluate_alerts([r["id"] for r in rows]) if rows else {"events": 0, "matched": 0, "delivered": 0, "failed": 0}226    stats.update({f"metric_{k}": v for k, v in (await evaluate_metric_alerts()).items()})227    return stats228229230@periodic("alerts-sweep", every_s=120, initial_delay_s=60)231async def alerts_sweep_task() -> None:232    stats = await sweep()233    if stats.get("matched") or stats.get("metric_fired"):234        log.info("alerts-sweep", extra=stats)235236237__all__ = ["deliver_webhook", "evaluate_alerts", "evaluate_metric_alerts", "event_matches", "metrics_match", "sign", "sweep", "webhook_payload"]238