Переписка в своей CRM

Что получится: модуль переписки для вашей CRM на Python. Входящие из всех подключённых каналов попадают в диалоги CRM, оператор отвечает из CRM, у сообщений видны статусы «отправлено» и «прочитано». Видно и то, что сотрудник написал с телефона. При первом открытии диалога подтягивается история. Модуль держится на вебхуке и SQLite — замените базу на свою.

                 ┌──────────── CRM ─────────────┐
 Gratio ─вебхук─►│ /gratio/hook/{idInstance}    │
                 │   входящие, свои, статусы,   │
                 │   правки, удаления, syncGap  │──► таблицы conversations, messages
                 │                              │
 Gratio ◄────────│ ответ оператора              │◄── POST /conversations/{id}/reply
  sendMessage    │   clientMessageId = id в CRM │
                 └──────────────────────────────┘

Что понадобится

• один или несколько каналов с ключами — Шаг 1. Подключите номер;
• Python 3.10+, pip install fastapi uvicorn requests;
• публичный HTTPS-адрес CRM.

Шаг 1. Модель данных

Диалог в CRM — пара «канал + chatId». Сообщение в CRM знает свой idMessage в мессенджере и ключ client_key, с которым его отправили: по нему статус находит сообщение, даже если ответ на отправку потерялся.

import sqlite3

db = sqlite3.connect("crm.db", check_same_thread=False)
db.executescript("""
create table if not exists conversations (
  id integer primary key, instance text, chat_id text, title text, needs_review int default 0,
  history_loaded int default 0, unique (instance, chat_id));
create table if not exists messages (
  id integer primary key, conversation_id int, direction text, source text, id_message text,
  client_key text unique, text text, file_url text, status text, deleted int default 0,
  at int, unique (conversation_id, id_message));
""")

Шаг 2. Ключи каналов

У каждого канала свой ключ. CRM хранит их и выбирает по idInstance из уведомления:

import json
import os
import requests

# {"7100000001": "токен…", "7100000002": "токен…"}
CHANNELS: dict[str, str] = json.loads(os.environ["GRATIO_CHANNELS"])
API = os.environ.get("GRATIO_URL", "https://api.gratio.net")
HOOK_TOKEN = os.environ["GRATIO_HOOK_TOKEN"]


def api(instance: str, method: str, payload: dict) -> dict:
    url = f"{API}/waInstance{instance}/{method}/{CHANNELS[instance]}"
    r = requests.post(url, json=payload, timeout=90)
    r.raise_for_status()
    return r.json()

Включите вебхук у каждого канала — адрес с idInstance в конце помогает видеть в журналах, какой канал прислал запрос:

curl -X POST "$GRATIO_URL/waInstance$GRATIO_ID/setSettings/$GRATIO_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{
    "webhookUrl": "https://crm.example.ru/gratio/hook/7100000001",
    "webhookUrlToken": "s3cr3t-hook-token"
  }'

Шаг 3. Входящие и свои сообщения

Один обработчик пишет в диалог и входящие (incomingMessageReceived), и сообщения с телефона сотрудника (outgoingMessageReceived), и отправленные через API (outgoingAPIMessageReceived). Повтор того же сообщения отсекает unique (conversation_id, id_message).

def text_of(data: dict) -> str:
    return ((data.get("textMessageData") or {}).get("textMessage")
            or (data.get("extendedTextMessageData") or {}).get("text")
            or (data.get("fileMessageData") or {}).get("caption") or "")


def conversation(instance: str, sender: dict) -> int:
    db.execute("insert or ignore into conversations (instance, chat_id, title) values (?, ?, ?)",
               (instance, sender["chatId"], sender.get("chatName") or sender["chatId"]))
    return db.execute("select id from conversations where instance = ? and chat_id = ?",
                      (instance, sender["chatId"])).fetchone()[0]


def on_message(instance: str, body: dict) -> None:
    data = body["messageData"]
    kind = data["typeMessage"]
    conv = conversation(instance, body["senderData"])
    if kind == "editedMessage":
        e = data["editedMessageData"]
        db.execute("update messages set text = ? where conversation_id = ? and id_message = ?",
                   (e["textMessage"], conv, e["stanzaId"]))
        return
    if kind == "deletedMessage":
        db.execute("update messages set deleted = 1 where conversation_id = ? and id_message = ?",
                   (conv, data["deletedMessageData"]["stanzaId"]))
        return
    if kind == "reactionMessage":
        return
    direction = "in" if body["typeWebhook"] == "incomingMessageReceived" else "out"
    source = {"incomingMessageReceived": "client", "outgoingMessageReceived": "phone"}.get(
        body["typeWebhook"], "api")
    file_url = (data.get("fileMessageData") or {}).get("downloadUrl")
    db.execute("""insert or ignore into messages (conversation_id, direction, source, id_message, text,
                  file_url, status, at) values (?, ?, ?, ?, ?, ?, ?, ?)""",
               (conv, direction, source, body["idMessage"], text_of(data), file_url,
                None if direction == "in" else "sent", body["timestamp"]))

Шаг 4. Ответ оператора

CRM сначала записывает сообщение у себя, потом отправляет его с clientMessageId — id записи. Если сеть оборвалась, повтор с тем же ключом не создаст второе сообщение у клиента.

import uuid


def reply(conversation_id: int, text: str) -> dict:
    instance, chat_id = db.execute("select instance, chat_id from conversations where id = ?",
                                   (conversation_id,)).fetchone()
    key = f"crm-{uuid.uuid4()}"
    db.execute("""insert into messages (conversation_id, direction, source, client_key, text, status, at)
                  values (?, 'out', 'operator', ?, ?, 'sending', strftime('%s','now'))""",
               (conversation_id, key, text))
    db.commit()
    try:
        sent = api(instance, "sendMessage", {"chatId": chat_id, "message": text, "clientMessageId": key})
    except requests.HTTPError as exc:
        code = exc.response.json().get("code", "") if exc.response is not None else ""
        # outcome_unknown — исход придёт статусом sent или failed с этим ключом
        status = "sending" if code == "outcome_unknown" else "failed"
        db.execute("update messages set status = ? where client_key = ?", (status, key))
        db.commit()
        return {"status": status, "code": code}
    db.execute("update messages set id_message = ?, status = 'sent' where client_key = ?",
               (sent["idMessage"], key))
    db.commit()
    return {"status": "sent", "idMessage": sent["idMessage"]}

Уведомление outgoingAPIMessageReceived об этом же сообщении придёт после ответа API: запись уже есть с этим idMessage, и insert or ignore её не повторит.

Шаг 5. Статусы

outgoingMessageStatus двигает статус вперёд: sent → delivered (WhatsApp) → read. failed приходит, если отправка с неизвестным исходом не нашлась на площадке, — тогда idMessage пуст, а сообщение находится по clientMessageId.

RANK = {"sending": 0, "sent": 1, "delivered": 2, "read": 3}


def on_status(body: dict) -> None:
    status = body["status"]
    if status == "failed" and body.get("clientMessageId"):
        db.execute("update messages set status = 'failed' where client_key = ?",
                   (body["clientMessageId"],))
        return
    row = db.execute("select id, status from messages where id_message = ?",
                     (body["idMessage"],)).fetchone()
    if row and RANK.get(status, 0) > RANK.get(row[1] or "sending", 0):
        db.execute("update messages set status = ? where id = ?", (status, row[0]))

Шаг 6. История при первом открытии

MAX и Telegram отдают историю с сервера мессенджера — в том числе сообщения до подключения к Gratio. Подтяните её, когда оператор впервые открывает диалог:

def load_history(conversation_id: int) -> None:
    instance, chat_id, loaded = db.execute(
        "select instance, chat_id, history_loaded from conversations where id = ?",
        (conversation_id,)).fetchone()
    if loaded:
        return
    for item in api(instance, "getChatHistory", {"chatId": chat_id, "count": 50}):
        db.execute("""insert or ignore into messages (conversation_id, direction, source, id_message,
                      text, file_url, status, at) values (?, ?, 'history', ?, ?, ?, ?, ?)""",
                   (conversation_id, "in" if item["type"] == "incoming" else "out", item["idMessage"],
                    item.get("textMessage", ""), item.get("downloadUrl"), item.get("statusMessage"),
                    item["timestamp"]))
    db.execute("update conversations set history_loaded = 1 where id = ?", (conversation_id,))
    db.commit()

Шаг 7. Разрыв истории

Если канал долго был без связи и Gratio не смог дочитать всё пропущенное, придёт syncGap. Пометьте диалоги, чтобы оператор их проверил, и сбросьте отметку истории — она подтянется заново при открытии:

def on_gap(instance: str, body: dict) -> None:
    for chat_id in body.get("chatIds", []):
        db.execute("""update conversations set needs_review = 1, history_loaded = 0
                      where instance = ? and chat_id = ?""", (instance, chat_id))

Шаг 8. Вебхук

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()


@app.post("/gratio/hook/{instance}")
def hook(instance: str, body: dict, authorization: str = Header("")):
    if authorization != f"Bearer {HOOK_TOKEN}" or instance not in CHANNELS:
        raise HTTPException(status_code=401)
    kind = body["typeWebhook"]
    if kind in ("incomingMessageReceived", "outgoingMessageReceived", "outgoingAPIMessageReceived"):
        on_message(instance, body)
    elif kind == "outgoingMessageStatus":
        on_status(body)
    elif kind == "syncGap":
        on_gap(instance, body)
    db.commit()
    return {"ok": True}


@app.post("/conversations/{conversation_id}/reply")
def reply_endpoint(conversation_id: int, payload: dict):
    return reply(conversation_id, payload["text"])


@app.get("/conversations/{conversation_id}")
def open_conversation(conversation_id: int):
    load_history(conversation_id)
    rows = db.execute("""select direction, source, text, file_url, status, deleted, at from messages
                         where conversation_id = ? order by at""", (conversation_id,)).fetchall()
    return [dict(zip(("direction", "source", "text", "file", "status", "deleted", "at"), r)) for r in rows]

Обработчик вебхука здесь синхронный и быстрый: запись в SQLite занимает миллисекунды. Если ваша обработка дольше — отвечайте 200 сразу и обрабатывайте в фоне, см. Вебхук.

Шаг 9. Запустите и проверьте

Сохраните полный код в crm.py и запустите:

export GRATIO_CHANNELS='{"7100000001": "ваш_apiTokenInstance"}'
export GRATIO_HOOK_TOKEN=s3cr3t-hook-token
uvicorn crm:app --port 8080

1. Напишите на номер канала с телефона. Вы увидите диалог: curl localhost:8080/conversations/1.

2. Ответьте из CRM:

   curl -X POST localhost:8080/conversations/1/reply \
     -H 'Content-Type: application/json' -d '{"text": "Здравствуйте!"}'
   

   Вы увидите {"status": "sent", …} и ответ на телефоне.

3. Прочитайте ответ на телефоне. Вы увидите у сообщения status: read.

4. Напишите клиенту с телефона сотрудника. Вы увидите сообщение с source: phone.

Полный код

import json
import os
import sqlite3
import uuid

import requests
from fastapi import FastAPI, Header, HTTPException

# {"7100000001": "токен…", "7100000002": "токен…"}
CHANNELS: dict[str, str] = json.loads(os.environ["GRATIO_CHANNELS"])
API = os.environ.get("GRATIO_URL", "https://api.gratio.net")
HOOK_TOKEN = os.environ["GRATIO_HOOK_TOKEN"]
RANK = {"sending": 0, "sent": 1, "delivered": 2, "read": 3}

db = sqlite3.connect("crm.db", check_same_thread=False)
db.executescript("""
create table if not exists conversations (
  id integer primary key, instance text, chat_id text, title text, needs_review int default 0,
  history_loaded int default 0, unique (instance, chat_id));
create table if not exists messages (
  id integer primary key, conversation_id int, direction text, source text, id_message text,
  client_key text unique, text text, file_url text, status text, deleted int default 0,
  at int, unique (conversation_id, id_message));
""")
app = FastAPI()


def api(instance: str, method: str, payload: dict) -> dict:
    url = f"{API}/waInstance{instance}/{method}/{CHANNELS[instance]}"
    r = requests.post(url, json=payload, timeout=90)
    r.raise_for_status()
    return r.json()


def text_of(data: dict) -> str:
    return ((data.get("textMessageData") or {}).get("textMessage")
            or (data.get("extendedTextMessageData") or {}).get("text")
            or (data.get("fileMessageData") or {}).get("caption") or "")


def conversation(instance: str, sender: dict) -> int:
    db.execute("insert or ignore into conversations (instance, chat_id, title) values (?, ?, ?)",
               (instance, sender["chatId"], sender.get("chatName") or sender["chatId"]))
    return db.execute("select id from conversations where instance = ? and chat_id = ?",
                      (instance, sender["chatId"])).fetchone()[0]


def on_message(instance: str, body: dict) -> None:
    data = body["messageData"]
    kind = data["typeMessage"]
    conv = conversation(instance, body["senderData"])
    if kind == "editedMessage":
        e = data["editedMessageData"]
        db.execute("update messages set text = ? where conversation_id = ? and id_message = ?",
                   (e["textMessage"], conv, e["stanzaId"]))
        return
    if kind == "deletedMessage":
        db.execute("update messages set deleted = 1 where conversation_id = ? and id_message = ?",
                   (conv, data["deletedMessageData"]["stanzaId"]))
        return
    if kind == "reactionMessage":
        return
    direction = "in" if body["typeWebhook"] == "incomingMessageReceived" else "out"
    source = {"incomingMessageReceived": "client", "outgoingMessageReceived": "phone"}.get(
        body["typeWebhook"], "api")
    file_url = (data.get("fileMessageData") or {}).get("downloadUrl")
    db.execute("""insert or ignore into messages (conversation_id, direction, source, id_message, text,
                  file_url, status, at) values (?, ?, ?, ?, ?, ?, ?, ?)""",
               (conv, direction, source, body["idMessage"], text_of(data), file_url,
                None if direction == "in" else "sent", body["timestamp"]))


def on_status(body: dict) -> None:
    status = body["status"]
    if status == "failed" and body.get("clientMessageId"):
        db.execute("update messages set status = 'failed' where client_key = ?",
                   (body["clientMessageId"],))
        return
    row = db.execute("select id, status from messages where id_message = ?",
                     (body["idMessage"],)).fetchone()
    if row and RANK.get(status, 0) > RANK.get(row[1] or "sending", 0):
        db.execute("update messages set status = ? where id = ?", (status, row[0]))


def on_gap(instance: str, body: dict) -> None:
    for chat_id in body.get("chatIds", []):
        db.execute("""update conversations set needs_review = 1, history_loaded = 0
                      where instance = ? and chat_id = ?""", (instance, chat_id))


def reply(conversation_id: int, text: str) -> dict:
    instance, chat_id = db.execute("select instance, chat_id from conversations where id = ?",
                                   (conversation_id,)).fetchone()
    key = f"crm-{uuid.uuid4()}"
    db.execute("""insert into messages (conversation_id, direction, source, client_key, text, status, at)
                  values (?, 'out', 'operator', ?, ?, 'sending', strftime('%s','now'))""",
               (conversation_id, key, text))
    db.commit()
    try:
        sent = api(instance, "sendMessage", {"chatId": chat_id, "message": text, "clientMessageId": key})
    except requests.HTTPError as exc:
        code = exc.response.json().get("code", "") if exc.response is not None else ""
        status = "sending" if code == "outcome_unknown" else "failed"
        db.execute("update messages set status = ? where client_key = ?", (status, key))
        db.commit()
        return {"status": status, "code": code}
    db.execute("update messages set id_message = ?, status = 'sent' where client_key = ?",
               (sent["idMessage"], key))
    db.commit()
    return {"status": "sent", "idMessage": sent["idMessage"]}


def load_history(conversation_id: int) -> None:
    instance, chat_id, loaded = db.execute(
        "select instance, chat_id, history_loaded from conversations where id = ?",
        (conversation_id,)).fetchone()
    if loaded:
        return
    for item in api(instance, "getChatHistory", {"chatId": chat_id, "count": 50}):
        db.execute("""insert or ignore into messages (conversation_id, direction, source, id_message,
                      text, file_url, status, at) values (?, ?, 'history', ?, ?, ?, ?, ?)""",
                   (conversation_id, "in" if item["type"] == "incoming" else "out", item["idMessage"],
                    item.get("textMessage", ""), item.get("downloadUrl"), item.get("statusMessage"),
                    item["timestamp"]))
    db.execute("update conversations set history_loaded = 1 where id = ?", (conversation_id,))
    db.commit()


@app.post("/gratio/hook/{instance}")
def hook(instance: str, body: dict, authorization: str = Header("")):
    if authorization != f"Bearer {HOOK_TOKEN}" or instance not in CHANNELS:
        raise HTTPException(status_code=401)
    kind = body["typeWebhook"]
    if kind in ("incomingMessageReceived", "outgoingMessageReceived", "outgoingAPIMessageReceived"):
        on_message(instance, body)
    elif kind == "outgoingMessageStatus":
        on_status(body)
    elif kind == "syncGap":
        on_gap(instance, body)
    db.commit()
    return {"ok": True}


@app.post("/conversations/{conversation_id}/reply")
def reply_endpoint(conversation_id: int, payload: dict):
    return reply(conversation_id, payload["text"])


@app.get("/conversations/{conversation_id}")
def open_conversation(conversation_id: int):
    load_history(conversation_id)
    rows = db.execute("""select direction, source, text, file_url, status, deleted, at from messages
                         where conversation_id = ? order by at""", (conversation_id,)).fetchall()
    return [dict(zip(("direction", "source", "text", "file", "status", "deleted", "at"), r)) for r in rows]

Что ещё добавить

• Имя и фото клиента — getContactInfo и getAvatar; ответ хранится в кэше до часа, спрашивать можно при каждом открытии.
• Файлы от оператора — sendFileByUpload; входящие файлы уже в file_url, ссылка открывается без токена.
• Объединение клиента из разных каналов — по номеру: в WhatsApp он в chatId, для MAX и Telegram — checkAccount по номеру из карточки CRM.
• Звонки в карточке — уведомления incomingCall и outgoingCall.

Что дальше

• Отдать ту же ленту аналитике, не мешая CRM, — Вторая система на той же ленте.
• Писать клиентам первыми по расписанию — Напоминания и рассылки без блокировок.