strix/strix/telemetry/posthog.py

import logging
from typing import TYPE_CHECKING, Any

import requests

from strix.config import load_settings
from strix.telemetry._common import (
    SEND_TIMEOUT,
    SESSION_ID,
    base_props,
    is_first_run,
)


if TYPE_CHECKING:
    from strix.report.state import ReportState


logger = logging.getLogger(__name__)

_POSTHOG_PUBLIC_API_KEY = "phc_7rO3XRuNT5sgSKAl6HDIrWdSGh1COzxw0vxVIAR6vVZ"
_POSTHOG_HOST = "https://us.i.posthog.com"


def _is_enabled() -> bool:
    return load_settings().telemetry.enabled


def _send(event: str, properties: dict[str, Any]) -> bool:
    if not _is_enabled():
        logger.debug("posthog disabled; skipping event %s", event)
        return False
    try:
        payload = {
            "api_key": _POSTHOG_PUBLIC_API_KEY,
            "event": event,
            "distinct_id": SESSION_ID,
            "properties": properties,
        }
        with requests.post(f"{_POSTHOG_HOST}/capture/", json=payload, timeout=SEND_TIMEOUT):
            pass
    except Exception:  # noqa: BLE001
        logger.debug("posthog send failed for event %s", event, exc_info=True)
        return False
    else:
        logger.debug("posthog event sent: %s", event)
        return True


def start(
    model: str | None,
    scan_mode: str | None,
    is_whitebox: bool,
    interactive: bool,
    has_instructions: bool,
    auth_mode: str | None = None,
) -> None:
    _send(
        "scan_started",
        {
            **base_props(),
            "model": model or "unknown",
            "auth_mode": auth_mode or "api_key",
            "scan_mode": scan_mode or "unknown",
            "scan_type": "whitebox" if is_whitebox else "blackbox",
            "interactive": interactive,
            "has_instructions": has_instructions,
            "first_run": is_first_run(),
        },
    )


def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None:
    _send(
        "finding_reported",
        {
            **base_props(),
            "severity": severity.lower(),
            "cwe": (cwe or "").strip().lower() or "unknown",
            "is_cve": is_cve,
        },
    )


def skill_loaded(skill_name: str) -> None:
    _send(
        "skill_loaded",
        {
            **base_props(),
            "skill": skill_name,
        },
    )


def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
    if report_state.posthog_scan_ended_sent:
        return
    if report_state.scan_ended_exit_reason is None:
        report_state.scan_ended_exit_reason = exit_reason

    vulnerabilities_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
    for v in report_state.vulnerability_reports:
        sev = v.get("severity", "info").lower()
        if sev in vulnerabilities_counts:
            vulnerabilities_counts[sev] += 1

    duration = report_state.get_process_duration_seconds()

    llm_props: dict[str, int | float] = {}
    try:
        usage = report_state.get_process_llm_usage()
        if isinstance(usage, dict):
            llm_props = {
                "llm_requests": int(usage.get("requests") or 0),
                "llm_input_tokens": int(usage.get("input_tokens") or 0),
                "llm_output_tokens": int(usage.get("output_tokens") or 0),
                "llm_tokens": int(usage.get("total_tokens") or 0),
                "llm_cost": float(usage.get("cost") or 0.0),
            }
    except (TypeError, ValueError, AttributeError):
        pass

    report_state.posthog_scan_ended_sent = _send(
        "scan_ended",
        {
            **base_props(),
            "auth_mode": report_state.run_record.get("auth_mode") or "api_key",
            "exit_reason": report_state.scan_ended_exit_reason,
            "duration_seconds": round(duration),
            "vulnerabilities_total": len(report_state.vulnerability_reports),
            **{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
            **llm_props,
        },
    )


def viewer_opened(source: str, live: bool) -> None:
    _send(
        "viewer_opened",
        {
            **base_props(),
            "source": source,
            "live": live,
        },
    )


def viewer_cta_clicked(cta: str, surface: str | None = None) -> None:
    props = {
        **base_props(),
        "cta": cta[:64],
    }
    if surface:
        props["surface"] = surface[:64]
    _send("viewer_cta_clicked", props)


_VIEWER_EMAIL_STEPS = frozenset(
    {"email_submitted", "email_verified", "report_sent", "work_email_required"}
)


def viewer_email_event(step: str, purpose: str | None = None) -> None:
    if step not in _VIEWER_EMAIL_STEPS:
        return
    _send(
        f"viewer_{step}",
        {
            **base_props(),
            **({"purpose": purpose} if purpose else {}),
        },
    )


def viewer_feedback_submitted() -> None:
    _send("viewer_feedback_submitted", {**base_props()})


def viewer_agent_steered() -> None:
    _send("viewer_agent_steered", {**base_props()})


def error(error_type: str) -> None:
    props = {**base_props(), "error_type": error_type}
    _send("error", props)