basicmachines-co-basic-memory/src/basic_memory/cli/analytics.py

"""Lightweight CLI analytics via Umami event collector.

Sends anonymous, non-blocking usage events to help understand how the
CLI-to-cloud conversion funnel performs. No PII, no fingerprinting,
no cookies. Respects the same opt-out mechanisms as promo messaging.

Events are fire-and-forget — analytics never blocks or breaks the CLI.

Defaults point to the Basic Memory Umami Cloud instance. Override via:
    BASIC_MEMORY_UMAMI_HOST     — Custom Umami instance URL
    BASIC_MEMORY_UMAMI_SITE_ID  — Custom Website ID
Opt out entirely with BASIC_MEMORY_NO_PROMOS=1.
"""

import json
import os
import threading
import urllib.request
from typing import Optional

import basic_memory


# ---------------------------------------------------------------------------
# Configuration — defaults baked in, overridable via environment
# ---------------------------------------------------------------------------

_DEFAULT_UMAMI_HOST = "https://api-gateway.umami.dev"
_DEFAULT_UMAMI_SITE_ID = "f6479898-ebaf-4e60-bce2-6dc60a3f6c5c"


def _umami_host() -> Optional[str]:
    return os.getenv("BASIC_MEMORY_UMAMI_HOST", "").strip() or _DEFAULT_UMAMI_HOST


def _umami_site_id() -> Optional[str]:
    return os.getenv("BASIC_MEMORY_UMAMI_SITE_ID", "").strip() or _DEFAULT_UMAMI_SITE_ID


def _analytics_disabled() -> bool:
    """True when analytics should not fire."""
    value = os.getenv("BASIC_MEMORY_NO_PROMOS", "").strip().lower()
    return value in {"1", "true", "yes"}


def _is_configured() -> bool:
    """True when both host and site ID are available."""
    return _umami_host() is not None and _umami_site_id() is not None


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

# Well-known event names for the promo/cloud funnel
EVENT_PROMO_SHOWN = "cli-promo-shown"
EVENT_PROMO_OPTED_OUT = "cli-promo-opted-out"
EVENT_CLOUD_LOGIN_STARTED = "cli-cloud-login-started"
EVENT_CLOUD_LOGIN_SUCCESS = "cli-cloud-login-success"
EVENT_CLOUD_LOGIN_SUB_REQUIRED = "cli-cloud-login-sub-required"


def track(event_name: str, data: Optional[dict] = None) -> None:
    """Send an analytics event to Umami. Non-blocking, silent on failure.

    Parameters
    ----------
    event_name:
        Short kebab-case name (e.g. "cli-promo-shown").
    data:
        Optional dict of event properties (all values should be strings/numbers).
    """
    if _analytics_disabled() or not _is_configured():
        return

    host = _umami_host()
    site_id = _umami_site_id()

    # Umami v2 /api/send requires "type" at top level alongside "payload"
    payload = {
        "type": "event",
        "payload": {
            "hostname": "cli.basicmemory.com",
            "language": "en",
            "url": f"/cli/{event_name}",
            "website": site_id,
            "name": event_name,
            "data": {
                "version": basic_memory.__version__,
                **(data or {}),
            },
        },
    }

    def _send():
        try:
            req = urllib.request.Request(
                f"{host}/api/send",
                data=json.dumps(payload).encode("utf-8"),
                headers={
                    "Content-Type": "application/json",
                    # Umami's bot detection rejects non-browser User-Agents
                    "User-Agent": "Mozilla/5.0 (compatible; BasicMemoryCLI/"
                    f"{basic_memory.__version__})",
                },
            )
            urllib.request.urlopen(req, timeout=3)
        except Exception:
            pass  # Never break the CLI for analytics

    # Non-daemon so the process waits for the request to complete.
    # The 3s urllib timeout caps the worst-case exit delay.
    t = threading.Thread(target=_send)
    t.start()