Kommo + Hex: аналитика продаж в collaborative data notebooks без BI-лицензии

Hex - modern data workspace: SQL + Python notebooks с возможностью публикации как интерактивных дашбордов. Используется data-командами как альтернатива Jupyter + Tableau: одно место для исследования данных, построения моделей и шеринга результатов с бизнесом. Для команд с Kommo, Hex позволяет выгрузить данные сделок, построить когортный анализ, визуализировать воронку конверсии и опубликовать как живой дашборд для руководителя - без BI-инструментов и без повторных SQL-запросов.

Архитектура: Python ETL выгружает данные из Kommo API в PostgreSQL/Snowflake, Hex подключается к базе данных и запускает SQL + Python cells. Опционально - запуск Hex Project через API по расписанию.

Hex API использует Bearer token. Основные операции: POST /api/v1/projects/{hex_project_id}/runs - запустить обновление проекта (обновить все SQL-ячейки с актуальными данными). GET /api/v1/projects/{id}/runs/{run_id} - статус запуска.

Hex Project - notebook с SQL и Python ячейками + Published App (дашборд) для non-технической аудитории. Проект можно запустить по расписанию (Hex Schedules) или через API.

ETL: Kommo -> PostgreSQL для Hex

import requests, os, psycopg2
from datetime import datetime, timezone

KOMMO_SUBDOMAIN = os.environ["KOMMO_SUBDOMAIN"]
KOMMO_TOKEN     = os.environ["KOMMO_ACCESS_TOKEN"]
PG_DSN          = os.environ["PG_DSN"]

KOMMO_BASE = f"https://{KOMMO_SUBDOMAIN}.kommo.com/api/v4"
KOMMO_HDR  = {"Authorization": f"Bearer {KOMMO_TOKEN}"}

def fetch_leads_paginated(params: dict) -> list[dict]:
    all_leads = []
    page = 1
    while True:
        params["page"] = page
        r = requests.get(f"{KOMMO_BASE}/leads", headers=KOMMO_HDR, params=params)
        if r.status_code == 204:
            break
        items = r.json().get("_embedded", {}).get("leads", []) or []
        if not items:
            break
        all_leads.extend(items)
        page += 1
    return all_leads

def upsert_leads(leads: list[dict], conn):
    cur = conn.cursor()
    cur.execute(
        "CREATE TABLE IF NOT EXISTS kommo_leads ("
        "  id BIGINT PRIMARY KEY, name TEXT, status_id INT, pipeline_id INT,"
        "  responsible_id INT, price NUMERIC,"
        "  created_at TIMESTAMPTZ, updated_at TIMESTAMPTZ, closed_at TIMESTAMPTZ,"
        "  loss_reason_id INT"
        ")"
    )
    for lead in leads:
        ts = lambda k: (
            datetime.fromtimestamp(lead[k], tz=timezone.utc) if lead.get(k) else None
        )
        cur.execute(
            "INSERT INTO kommo_leads"
            " (id,name,status_id,pipeline_id,responsible_id,price,created_at,updated_at,closed_at,loss_reason_id)"
            " VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
            " ON CONFLICT (id) DO UPDATE SET"
            " name=EXCLUDED.name, status_id=EXCLUDED.status_id, price=EXCLUDED.price,"
            " updated_at=EXCLUDED.updated_at, closed_at=EXCLUDED.closed_at",
            (lead["id"], lead.get("name"), lead.get("status_id"), lead.get("pipeline_id"),
             lead.get("responsible_user_id"), lead.get("price"),
             ts("created_at"), ts("updated_at"), ts("closed_at"), lead.get("loss_reason_id")),
        )
    conn.commit()
    cur.close()

def run_etl():
    conn   = psycopg2.connect(PG_DSN)
    params = {"limit": 250, "with": "custom_fields_values"}
    leads  = fetch_leads_paginated(params)
    upsert_leads(leads, conn)
    conn.close()
    print(f"ETL: {len(leads)} leads upserted")

if __name__ == "__main__":
    run_etl()

Запуск Hex Project через API

HEX_TOKEN      = os.environ["HEX_API_TOKEN"]
HEX_PROJECT_ID = os.environ["HEX_PROJECT_ID"]
HEX_BASE       = "https://app.hex.tech/api/v1"
HEX_HDR        = {"Authorization": f"Bearer {HEX_TOKEN}", "Content-Type": "application/json"}

def trigger_hex_run(input_params: dict | None = None) -> str:
    payload = {}
    if input_params:
        payload["inputParams"] = input_params
    r = requests.post(
        f"{HEX_BASE}/projects/{HEX_PROJECT_ID}/runs",
        headers=HEX_HDR,
        json=payload,
    )
    r.raise_for_status()
    return r.json().get("runId", "")

def wait_hex_run(run_id: str, timeout_sec: int = 300) -> str:
    import time
    deadline = time.time() + timeout_sec
    while time.time() < deadline:
        r = requests.get(
            f"{HEX_BASE}/projects/{HEX_PROJECT_ID}/runs/{run_id}",
            headers=HEX_HDR,
        )
        status = r.json().get("status", "")
        if status in ("COMPLETED", "ERRORED", "KILLED"):
            return status
        time.sleep(10)
    return "TIMEOUT"

# Использование: запускать по cron после ETL
run_id = trigger_hex_run()
status = wait_hex_run(run_id)
print(f"Hex run {run_id}: {status}")

SQL-ячейки в Hex для анализа Kommo

-- Воронка конверсии по месяцам
SELECT
    DATE_TRUNC('month', created_at) AS month,
    COUNT(*)                         AS total_leads,
    COUNT(*) FILTER (WHERE status_id = {{ won_stage_id }})  AS won,
    COUNT(*) FILTER (WHERE status_id = {{ lost_stage_id }}) AS lost,
    ROUND(
        100.0 * COUNT(*) FILTER (WHERE status_id = {{ won_stage_id }}) / COUNT(*),
        1
    ) AS win_rate_pct
FROM kommo_leads
WHERE created_at >= NOW() - INTERVAL '12 months'
GROUP BY 1
ORDER BY 1;

-- Среднее время сделки (days to close) по менеджерам
SELECT
    responsible_id,
    ROUND(AVG(EXTRACT(DAY FROM (closed_at - created_at))), 1) AS avg_days_to_close,
    COUNT(*) FILTER (WHERE status_id = {{ won_stage_id }})    AS won_count,
    SUM(price) FILTER (WHERE status_id = {{ won_stage_id }})  AS total_revenue
FROM kommo_leads
WHERE closed_at IS NOT NULL
  AND status_id = {{ won_stage_id }}
GROUP BY 1
ORDER BY 3 DESC;

В Hex {{ variable }} - это Input Cell (параметр). Создайте Input Cells для won_stage_id и lost_stage_id - пользователи дашборда смогут менять их без кода.

Hex Scheduled Runs

Hex App -> [Project] -> Schedules -> Add Schedule. Например: каждый день в 7:00 UTC - запускает SQL-ячейки с актуальными данными. Published App обновляется автоматически.

Для связки ETL + Hex Run: настройте cron (0 6 * * *) который запускает ETL-скрипт, затем через API триггерит Hex. Hex запускается после ETL завершится.

Для кого актуально

Data-команды в B2B SaaS с Kommo: аналитик или data scientist, которому нужно регулярно готовить отчёт по воронке для sales-директора. Hex позволяет написать SQL + Python один раз и опубликовать как живой дашборд без настройки BI-инструмента.

Аналогичный подход с dbt-слоем: Kommo + Lightdash.

Часто задаваемые вопросы

Нужен ли Hex платный план для API-запусков?

Да. API-доступ к запуску проектов доступен на Hex Team ($24/user/mo) и выше. На бесплатном плане запуск только вручную. Scheduled Runs - только на платных планах.

Как Hex подключается к PostgreSQL?

Hex Connections -> Add Connection -> PostgreSQL -> ввести host, port, database, user, password. Hex поддерживает SSH-туннели и SSL. После подключения все SQL-ячейки проекта могут использовать это соединение. Данные не кэшируются в Hex - каждый запуск делает свежий запрос к базе.

Можно ли экспортировать данные из Hex в CSV автоматически?

Hex позволяет экспортировать результаты ячеек в CSV через UI. Для автоматического экспорта через API: после запуска проекта используйте GET /api/v1/projects/{id}/runs/{run_id}/results/{cell_id} для получения данных конкретной ячейки в JSON. CSV-экспорт через API в development.

Итог

Kommo + Hex - аналитика продаж в data notebooks:

• ETL Python: fetch_leads_paginated -> upsert в PostgreSQL с ON CONFLICT DO UPDATE
• Hex API: POST /api/v1/projects/{id}/runs -> trigger после ETL
• SQL ячейки: воронка конверсии, days_to_close, revenue по менеджерам
• Input Cells {{ won_stage_id }} - параметры для non-технической аудитории
• Scheduled Runs: автообновление дашборда без ручного запуска

Если нужна настройка аналитики по данным Kommo в Hex или другом data warehouse - опишите задачу команде Exceltic.dev.