Привет, мир! Меня зовут Павел, я IT инженер и руководитель службы технической поддержки.

Эта статья - вторая часть инструкции по внедрению коннектора WhatsApp и Telegram для Открытых линий CRM Bitrix24. С логикой подключения к Битрикс вы можете ознакомиться в первой части, а в этой статье мы рассмотрим логику обмена сообщениями через WhatsApp.

Обработчик WhatsApp Web

Как бы очевидно это не звучало, но первично нам понадобится обработчик WhatsApp. В ходе долгих поисков адекватного решения, выбор пал на Evolution API в виду неплохой документации относительно аналогов и "живого" комьюнити (хоть и испаноязычного).

Инструкция по развертыванию на сервере есть на официальном сайте - https://doc.evolution-api.com/v1/en/install/docker. Однако я бы рекомендовал использовать образ evoapicloud/evolution-api:v2.3.1, в котором исправлена работа с групповыми чатами. При установке, помимо обязательных переменных, в .env надо указать:

WEBHOOK_GLOBAL_ENABLED=true
WEBHOOK_GLOBAL_URL='https://<url_коннектора>/api/waweb/?api-key=XXXX'
AUTHENTICATION_API_KEY=YYY
CONFIG_SESSION_PHONE_VERSION="2.3000.1025062854"

Модуль коннектора WhatsApp в нашем веб-приложении

Создадим django приложение waweb. Далее реализуем модели в models.py. Модель для хранения данных о сервере Evolution API и его базовой конфигурации:

class Server(models.Model):
    url = models.URLField(max_length=255, unique=True, verbose_name="Server URL")
    api_key = models.CharField(max_length=255, verbose_name="API Key")
    max_connections = models.PositiveIntegerField(default=100)
    groups_ignore = models.BooleanField(default=True)
    always_online = models.BooleanField(default=False)
    read_messages = models.BooleanField(default=False)

    def __str__(self):
        return self.url

Модель для хранения данных о сессии WhatsApp:

class Session(models.Model):
    session = models.UUIDField(default=uuid.uuid4, editable=False, unique=True)
    server = models.ForeignKey(Server, on_delete=models.SET_NULL, related_name="sessions", null=True, blank=True)
    apikey = models.CharField(max_length=255, blank=True, null=True)
    instanceId = models.CharField(max_length=255, blank=True, null=True)
    date_end = models.DateTimeField(null=True, blank=True)
    phone = models.CharField(max_length=15, blank=True, null=True)
    groups_ignore = models.BooleanField(default=True)
    sms_service = models.BooleanField(default=True)
    status = models.CharField(max_length=15, blank=True, null=True)
    owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True)
    app_instance = models.ForeignKey(
        AppInstance, on_delete=models.SET_NULL, related_name="wawebs", null=True, blank=True)
    line = models.ForeignKey(
        Line,
        on_delete=models.SET_NULL,
        blank=True,
        null=True,
        related_name="wawebs",
    )

    def __str__(self):
        return f"Session: {self.session}, Phone: {self.phone or 'Not connected'}"

Приступим к реализации api эндпоинта для обработки событий WhatsApp в файле api/views.py. Реализуем функцию, принимающую события от WhatsApp и формирующую корректный payload для Битрикс, в зависимости от типа сообщения. Таким образом, обработаем сразу сообщения из групп, медиафайлы и голосовые сообщения. Также реализуем загрузку локация через GoogleMap. Для каждого сообщения проверяется связь между Whatsapp сессией и открытой линией Битрикс, а также реализована защита от отправки повторных сообщений путем фильтрации по message_id:

from rest_framework.viewsets import GenericViewSet
from rest_framework.response import Response
from rest_framework.decorators import action
from waweb.models import Session
from rest_framework import permissions
import requests

from django.conf import settings
import redis
import logging
import uuid
import re
from django.utils import timezone

import waweb.tasks as tasks

import waweb.utils as utils
import bitrix.utils as bitrix_utils
import bitrix.tasks as bitrix_tasks


redis_client = redis.StrictRedis(host=settings.REDIS_HOST, port=6379, db=0)

logger = logging.getLogger("django")


class EventsHandler(GenericViewSet):
    def create(self, request, *args, **kwargs):
        event_data = request.data
        sessionid = event_data.get('instance')
        if not sessionid:
            return Response({'error': 'sessionId is required'})
        try:
            session = Session.objects.get(session=sessionid)
        except Session.DoesNotExist:
            return Response({'error': f'Session with sessionId {sessionid} does not exist'})
        
        if not session.owner:
            return Response({'error': 'Session has no owner'})
        
        event = event_data.get("event")
        data = event_data.get("data", {})
        apikey = event_data.get('apikey')
        if apikey and session.apikey != apikey:
            session.apikey = apikey
            session.save(update_fields=["apikey"])

        server = session.server
        headers = {"apikey": session.apikey}
        
        if event == "connection.update":
            state = data.get('state')
            if session.status != state:
                session.status = state
                session.save(update_fields=["status"])

            if state == "open":
                wuid = data.get("wuid")
                number = wuid.split("@")[0]
                session.phone = number
                session.save(update_fields=["phone"])

                if Session.objects.exclude(pk=session.pk).filter(phone=number).exists():
                    headers = {"apikey": server.api_key}
                    response = requests.delete(f"{server.url}instance/logout/{sessionid}", headers=headers, timeout=60)
                    response = requests.delete(f"{server.url}instance/delete/{sessionid}", headers=headers, timeout=60)
                    session.delete()
                    return Response({'error': 'Phone number already in use, session deleted'})
                

        elif event in ["messages.upsert", "send.message"]:
            if session.date_end and timezone.now() > session.date_end:
                return Response({'error': 'tariff has expired'})
            
            message = data.get('message', {})
            key_data = data.get('key', {})
            message_id = key_data.get('id')

            if redis_client.exists(f'waweb:{message_id}'):
                return Response({'message': 'loop message'})

            fromme = key_data.get('fromMe')
            sender = event_data.get('sender').split('@')[0]
            remoteJid = key_data.get('remoteJid')
            pushName = data.get("pushName")
            group_message = False
            if remoteJid.endswith('@g.us'):
                group_message = True
                params = {"groupJid": remoteJid}
                group_name = requests.get(f"{server.url}group/findGroupInfos/{sessionid}", params=params, headers=headers, timeout=60)
                if group_name.status_code == 200:
                    pushName = group_name.json().get("subject")
            file_data = {}
            remoteJid = remoteJid.split('@')[0]

            profilepic_url = None
            if not group_message:
                profilepic = requests.post(f"{server.url}chat/fetchProfilePictureUrl/{sessionid}", 
                                        json={"number": remoteJid}, headers=headers, timeout=60)
                if profilepic.status_code == 200:
                    profilepic = profilepic.json()
                    profilepic_url = profilepic.get("profilePictureUrl")
            
            payload = {
                'sender': sender,
                'remoteJid': remoteJid,
                'fromme': fromme,
            }

            msg_type = data.get('messageType')
            fileName = None

            print(f"Message Recieved: {data}")

            if msg_type == 'conversation':
                if group_message:
                    group_sender_name = f"{data.get('pushName')}:" if data.get('pushName') else ""
                    payload.update({'content': f"*{pushName}*\n{group_sender_name} {message.get('conversation')}"})
                else:
                    payload.update({'content': message.get('conversation')})

            elif msg_type == 'locationMessage':
                location = message.get(msg_type, {})
                latitude  = location.get('degreesLatitude')
                longitude  = location.get('degreesLongitude')
                description = f"{location.get('name')}: {location.get('address')}"
                body = f"Link: https://www.google.com/maps/place/{latitude},{longitude}"
                if "None" not in description:
                    body = f"Address: {description} \n {body}"
                payload.update({'content': body})

            elif msg_type == 'contactMessage':
                payload.update({'content': message.get(msg_type, {}).get("vcard")})

            elif msg_type == 'templateMessage':
                hydratedTemplate = message.get(msg_type, {}).get("hydratedTemplate", {})
                hydratedTitleText = hydratedTemplate.get("hydratedTitleText")
                hydratedContentText = hydratedTemplate.get("hydratedContentText")
                hydratedFooterText = hydratedTemplate.get("hydratedFooterText")
                payload.update({'content': f"{hydratedTitleText} \n {hydratedContentText} \n {hydratedFooterText}"})

            elif msg_type in ["imageMessage", "documentMessage", "videoMessage", "audioMessage"]:
                payload.update({'content': message.get(msg_type, {}).get("caption")})
                media_url = f"{server.url}chat/getBase64FromMediaMessage/{sessionid}"
                msg_payload = {"message": {"key": {"id": message_id}}}
                response = requests.post(media_url, json=msg_payload, headers=headers, timeout=60)
                if response.status_code == 201:
                    file_data = response.json()
                    file_body = file_data.get('base64')
                    fileName = file_data.get('fileName')
                    mimetype = file_data.get('mimetype')
                    if file_body:
                        from io import BytesIO
                        import base64
                        file_bytes = base64.b64decode(file_body)
                        file_like = BytesIO(file_bytes)
                        file_like.name = fileName
                        payload.update({'attachments': (file_like.name, file_like, mimetype)})
            else:
                return Response({'message': 'ok'})
            
            try:
                if session.line:
                    file_url = None
                    text = payload.get("content") or fileName
                    if file_data:
                        domain = session.line.portal.domain
                        chat_key = f'bitrix_chat:{domain}:{session.line.line_id}:{remoteJid}'
                        if redis_client.exists(chat_key):
                            chat_id = redis_client.get(chat_key).decode('utf-8')
                            chat_folder = None
                            try:
                                chat_folder = bitrix_utils.call_method(session.app_instance, "im.disk.folder.get", {"CHAT_ID": chat_id})
                                if isinstance(chat_folder, dict) and chat_folder.get("error"):
                                    logger.error(chat_folder["detail"])
                            except Exception as e:
                                logger.error(f"Bitrix error: {e}")
                            if chat_folder and "result" in chat_folder:
                                bitrix_utils.call_method(session.app_instance, "imopenlines.session.join", {"CHAT_ID": chat_id})
                                folder_id =  chat_folder.get("result").get("ID")
                                upload_file = bitrix_utils.upload_file(
                                    session.app_instance, folder_id,
                                    file_body, fileName)
                                if upload_file:
                                    file_url = upload_file.get("DOWNLOAD_URL")
                    if fromme:
                        bitrix_tasks.message_add(session.app_instance.id, session.line.line_id,
                                                       remoteJid, text, session.line.connector.code)
                        if file_url:
                            file_upd = {
                                "CHAT_ID": chat_id,
                                "UPLOAD_ID": upload_file.get("FILE_ID"),
                                "DISK_ID": upload_file.get("ID"),
                                "SILENT_MODE": "Y",
                                "MESSAGE": fileName
                            }
                            bitrix_tasks.call_api(session.app_instance.id, "im.disk.file.commit", file_upd)
                    else:
                        attach = None
                        if file_url:
                            attach = [
                                {
                                    "url": file_url,
                                    "name": fileName
                                }
                            ]
                        bitrix_tasks.send_messages(session.app_instance.id, remoteJid, text, session.line.connector.code, session.line.line_id,
                                                            False, pushName, message_id, attach, profilepic_url)
                        

                return Response({"message": "message processed"})

            except Exception as e:
                print(f'Failed to send API message: {str(e)}')
                return Response({'error': f'Failed to send API message: {str(e)}'}, status=500)
            
        return Response({'message': 'ok'})
    
    
    @action(detail=False, methods=['post'], url_path=r'(?P<session>[^/.]+)/send', permission_classes=[permissions.AllowAny])
    def send(self, request, session=None, *args, **kwargs):
        session_id = session

        if not session_id:
            return Response({'error': 'session is required'})

        try:
            session = Session.objects.get(session=session_id)
        except Exception as e:
            return Response({'error': 'An error occurred', 'details': str(e)})
        
        if session.date_end and timezone.now() > session.date_end:
            return Response({'error': 'tariff has expired'}, status=402)

        data = request.data
        event = data.get('event')       
        message_type = data.get('message_type')
        attachments = data.get('attachments', {})

        if event == "message_created" and message_type == "outgoing":
            message_id = data.get('id')

            if redis_client.exists(f'chatwoot:{message_id}'):
                return Response({'message': 'loop message'})
        
            content = data.get('content')
            conversation = data.get('conversation', {})
            meta = conversation.get('meta', {})
            sender = meta.get('sender', {})
            phone_number = sender.get('phone_number')

            if content:
                wa_resp = utils.send_message(session_id, phone_number, content)
                if wa_resp.status_code == 201:
                    utils.store_msg(wa_resp)

                if session.line:
                    cleaned_phone = re.sub(r'\D', '', phone_number)
                    bitrix_tasks.message_add(session.app_instance.id, session.line.line_id,
                                                   cleaned_phone, content, session.line.connector.code)
            
            if attachments:
                for attachment in attachments:
                    tasks.send_message_task(str(session.session), [phone_number], attachment, 'media')
                return Response({'message': 'All files sent successfully'})

        return Response({'message': f'Session {session_id} authorized'})

Приступим к реализации функциональной части модуля. Создадим файл utils.py.

Предварительно рассмотрим импорты:

import requests
import base64
import mimetypes
import re
import redis
import time
import magic

from django.conf import settings
from .models import Session


redis_client = redis.StrictRedis(host=settings.REDIS_HOST, port=6379, db=0)

Далее реализуем функционал работы с файлами. Функция download_file принимает url отправленного файла и id сообщения. Далее нам нужно получить mimetype файла с помощью библиотеки python-magic, определить его расширение и передать данные в формате base64:

def download_file(attachment):
    data_url = attachment.get("data_url") or attachment.get("link")
    message_id = attachment.get("message_id", time.time())
    try:
        response = requests.get(data_url, stream=True, timeout=60)
        if response.status_code != 200:
            raise Exception(f"Failed to download file: {response.status_code} {response.text}")
        
        file_content = response.content
        mime = magic.Magic(mime=True)
        mimetype = mime.from_buffer(file_content)
        extension = mimetypes.guess_extension(mimetype) or ''
        filename = f"{message_id}{extension}"
        
        base64_encoded_data = base64.b64encode(file_content).decode("utf-8")
        
        return {
                "mimetype": mimetype,
                "data": base64_encoded_data,
                "filename": filename
        }
    except Exception as e:
        print(f"Error processing file: {str(e)}")
        return None

Также для исключения повторных сообщений, настроим сохранение id сообщений в redis:

def store_msg(resp):
    data = resp.json()
    msg_data = data.get('key', {})
    message_id = msg_data.get('id')
    if message_id:
        redis_client.setex(f'waweb:{message_id}', 600, message_id)

Реализуем непосредственно функцию работы с сообщениями. Данная функция очищает телефон от лишних символов (для корректной работы с лидами) и формирует payload в зависимости от типа сообщения:

def send_message(session_id, recipient, content, cont_type="string"):
    session = Session.objects.get(session=session_id)
    server = session.server
    headers = {"apikey": session.apikey}
    
    cleaned = re.sub(r'\D', '', recipient)
    
    if cont_type == "string":
        payload = {
            "number": cleaned,
            "text": content,
            "linkPreview": True,
        }
        url = f"{server.url}message/sendText/{session_id}"
        print("Sending WA message by url: ", url)
        print("WA message payload: ", payload)
        return requests.post(url, json=payload, headers=headers, timeout=60)
    
    elif cont_type == "media":
        url = f"{server.url}message/sendMedia/{session_id}"
        mimetype = content.get("mimetype", "")
        base_type = mimetype.split('/')[0]
        mediatype = base_type if base_type in ["image"] else "document"
        payload = {
            "number": cleaned,
            "mediatype": mediatype,
            "mimetype": content.get("mimetype"),
            "media": content.get("data"),
            "fileName": content.get("filename")
        }

        return requests.post(url, json=payload, headers=headers, timeout=60)

Реализуем celery функции для асинхронной работы приложения:

import requests
from django.utils import timezone
from django.db.models import Q
from datetime import timedelta
from celery import shared_task
import waweb.utils as utils
from django.conf import settings
from waweb.models import Session


@shared_task
def send_message_task(session_id, recipients, content, cont_type="string", from_web=False):
    if cont_type == "media":
        content = utils.download_file(content)
    for recipient in recipients:
        resp = utils.send_message(session_id, recipient, content, cont_type)
        if resp.status_code == 201 and not from_web:
            utils.store_msg(resp)


@shared_task
def delete_sessions(days=None):
    now = timezone.now()
    filters = Q((Q(phone__isnull=True) | Q(phone='')) & Q(date_end__lt=now))
    if days is not None:
        try:
            days_int = int(days)
            date_limit = now - timedelta(days=days_int)
            filters = filters | Q(date_end__lt=date_limit)
        except (TypeError, ValueError):
            pass

    sessions = Session.objects.filter(filters)
    for session in sessions:
        server = session.server
        headers = {"apikey": server.api_key}
        url = f"{server.url}instance/delete/{session.session}"
        requests.delete(url, headers=headers, timeout=60)
        session.delete()

Далее создадим views.py, где опишем функционал подключения WhatsApp к коннектору:

import requests
import redis
import uuid
from requests.exceptions import RequestException
from django.shortcuts import render, redirect, get_object_or_404
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from django.db.models import Count, F
from django.utils import timezone
from django.conf import settings

from bitrix.models import AppInstance, Line, Connector
import bitrix.utils as bitrix_utils

from users.models import Message
from main.decorators import login_message_required

from .models import Session, Server
from .forms import SendMessageForm
from .tasks import send_message_task

apps = settings.INSTALLED_APPS

redis_client = redis.StrictRedis(host=settings.REDIS_HOST, port=6379, db=0, decode_responses=True)

LINK_TTL = 60 * 60 * 24

Для начала создадим вьюшку для главной страницы модуля, в которой получим доступные сервера Evolution API и подключенные сессии (если таковые имеются).

@login_message_required(code="waweb")
def wa_sessions(request):
    connector_service = "waweb"
    connector = Connector.objects.filter(service=connector_service).first()
    if request.method == "POST":
        session_id = request.POST.get("session_id")
        line_id = request.POST.get("line_id")
        if not line_id:
            messages.warning(request, "Необходимо выбрать линию из списка или создать новую.")
            return redirect('waweb')
        phone = get_object_or_404(Session, id=session_id, owner=request.user)
        if not phone.phone:
            messages.error(request, "Сначала необходимо подключить WhatsApp.")
            return redirect('waweb')
        if phone.line and str(phone.line.id) == str(line_id):
            messages.warning(request, "Эта линия уже подключена к выбранной сессии.")
            return redirect('waweb')
        try:
            bitrix_utils.connect_line(request, line_id, phone, connector, connector_service)
        except Exception as e:
            messages.error(request, str(e))
            return redirect('waweb')
        return redirect('waweb')

    sessions = Session.objects.filter(owner=request.user)
    instances = AppInstance.objects.filter(owner=request.user, app__connectors=connector)
    wa_lines = Line.objects.filter(connector=connector, owner=request.user)

    for session in sessions:
        session.show_link = session.status == "open"

    return render(
        request, 'waweb/wa_sessions.html', {
            "sessions": sessions,
            "instances": instances,
            "wa_lines": wa_lines,
        }
    )

Далее добавим функцию для подключения новых сессий, в которой получим доступный сервер Evolution API (если их несколько) и вызовем функцию создания сессии:

@login_required
def connect_number(request, session_id=None):
    if not session_id:
        sessions = Session.objects.filter(
            phone__isnull=True,
            owner=request.user
        )
        if sessions and not request.user.integrator:
            messages.warning(request, "У вас уже есть незавершенное подключение. Нажмите 'Подключить'")
            return redirect('waweb')
        
        new_session = Session.objects.create(owner=request.user)
        session_id = new_session.session

    else:
        new_session = get_object_or_404(Session, session=session_id, owner=request.user)

    server = (
        Server.objects.annotate(connected_sessions=Count('sessions'))
        .filter(connected_sessions__lt=F('max_connections'))
        .order_by('id')
        .first()
    )

    if not server:
        messages.error(request, "Нет доступных серверов.")
        new_session.delete()
        return redirect('waweb')

    new_session.server = server
    new_session.save()

    try:
        img_data = create_instance(new_session)
        if img_data:
            request.session['qr_image'] = img_data       
        return redirect('', session_id=session_id)
    except Exception as e:
        print("request error:", e)
        new_session.delete()
        messages.error(request, "Failed to initiate session.")
        return redirect('waweb')

В функции create_instance отправим запрос в Evolution API на создание новой сессии и получим QR код:

def create_instance(session):
    server = session.server
    headers = {"apikey": server.api_key}
    
    payload = {
        "instanceName": str(session.session),
        "qrcode": True,
        "integration": "WHATSAPP-BAILEYS",
        "alwaysOnline": server.always_online,
        "groupsIgnore": server.groups_ignore,
        "readMessages": server.read_messages,
    }

    try:
        response = requests.post(f"{server.url}instance/create", json=payload, headers=headers, timeout=150)
        inst_data = response.json()
        instanceId = inst_data.get("instance", {}).get("instanceId")
        session.instanceId = instanceId
        session.save()
        print(inst_data)
        img_data = inst_data.get("qrcode", {}).get("base64", "")
        print(img_data)
        if "," in img_data:
            img_data = img_data.split(",", 1)[1]
        else:
            raise Exception("QR code base64 data is invalid or missing.")
        return img_data
    except RequestException as e:
        print("request error:", e)
        return None

Для отрисовки используем qr_code_page:

@login_required
def qr_code_page(request, session_id):
    qr_image = request.session.pop('qr_image', '')
    try:
        session = Session.objects.get(session=session_id)
    except Session.DoesNotExist:
        messages.error(request, "Session not found.")
        return redirect('waweb')
    
    if not qr_image:
        qr_image = get_gr(request, session)

    if not qr_image:
        return redirect('waweb')

    public_id = redis_client.get(f"public_qr:{session_id}")
    if not public_id:
        public_id = str(uuid.uuid4())
        redis_client.set(f"public_qr:{session_id}", public_id, ex=LINK_TTL)
        redis_client.set(f"public_qr:{public_id}", str(session_id), ex=LINK_TTL)    

    message = Message.objects.filter(code="waweb_instruction").first()
    if message:
        messages.info(request, message.message)
    return render(request, 'waweb/qr_code.html', {
        'qr_image': qr_image,
        'request': request,
        'public_id': public_id,
    })

А также добавим функцию для декодирования QR кода:

def get_gr(request, session):

    server = session.server
    if not server:
        messages.error(request, "Session is not attached to a server.")
        return
    
    if session.status == "open":
        messages.warning(request, "Session is connected.")
        return

    gr_url = f"{server.url}instance/connect/{session.session}"
    headers = {"apikey": server.api_key}
    try:
        response = requests.get(gr_url, headers=headers, timeout=60)
        if response.status_code == 404:
            return create_instance(session)
        
        inst_data = response.json()
        img_data = inst_data.get("base64", "")
        if img_data:
            qr_image = img_data.split(",", 1)[1]
            return qr_image
        else:
            messages.error(request, f"Failed to restart session. {inst_data}")
            return
    except RequestException:
        messages.error(request, "Failed connect to server")
        return

Пропишем маршруты к нашим вьюшкам:

from django.urls import path
from . import views

urlpatterns = [
    path('waweb/', views.wa_sessions, name='waweb'),
    path('connect/', views.connect_number, name='connect_number'),
    path('qr/<uuid:session_id>/', views.qr_code_page, name='qr_code_page'),
]

Модуль Telegram коннектора

Коннекторы WhatsApp и Telegram довольно схожи в своей реализации и базовой логике обработки сообщений. Однако ввиду политики открытого API, интеграция с телегой потребовала гораздо меньшего количество ресурсов. Как минимум, вам не потребуется знание испанского для баг-фиксов.

Создадим django приложение telegram. В models.py нам нужно лишь хранить данные о подключенном Telegram боте, а также выставить вебхук для пересылки сообщений:

import requests
from django.conf import settings
from django.db import models

from bitrix.models import AppInstance, Line

class TelegramBot(models.Model):
    bot_token = models.CharField(max_length=255, unique=True, verbose_name="Bot Token")
    bot_username = models.CharField(max_length=255, verbose_name="Bot Username", blank=True, null=True)
    receive_messages = models.BooleanField(default=True)
    send_typing_actions = models.BooleanField(default=True)
    groups_ignore = models.BooleanField(default=True)
    date_end = models.DateTimeField(null=True, blank=True)
    owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True)
    app_instance = models.ForeignKey(
        AppInstance, on_delete=models.SET_NULL, related_name="telegrams", null=True, blank=True)
    line = models.ForeignKey(
        Line,
        on_delete=models.SET_NULL,
        blank=True,
        null=True,
        related_name="telegrams",
    )

    def __str__(self):
        return self.bot_username

    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)

        if self.bot_token and self.bot_username:
            site_domain = self.app_instance.app.site.domain
            webhook_url = f"https://{site_domain}/api/telegram/{self.bot_username}/"
            api_url = f"https://api.telegram.org/bot{self.bot_token}/setWebhook"

            response = requests.post(api_url, data={"url": webhook_url})
            print(response.json())
            if response.status_code != 200:
                raise Exception(
                    f"Не удалось установить webhook: {response.text}"
                )

Обработчик событий в api/views.py функционально идентичен обработчику WhatsApp, с небольшими поправками на структуру сообщений Telegram. Поэтому задерживаться на этом моменте мы не будем, просто приложу готовый код:

import logging

import bitrix.tasks as bitrix_tasks
import redis
import requests
import telegram.tasks as telegram_tasks
from django.conf import settings
from rest_framework import permissions
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet
from telegram.models import TelegramBot

logger = logging.getLogger("django")
redis_client = redis.StrictRedis(host=settings.REDIS_HOST, port=6379, db=0)


class TelegramEventsHandler(GenericViewSet):
    permission_classes = [permissions.AllowAny]

    def create(self, request, *args, **kwargs):
        data = request.data
        print(data)
        bot_name = kwargs.get("bot_name")
        message = data.get("message")
        if not message:
            return Response({"message": "No message"}, status=200)

        try:
            bot = TelegramBot.objects.get(bot_username=bot_name)
        except TelegramBot.DoesNotExist:
            return Response({'error': f'Bot {bot_name} does not exist'})

        chat_id = message["chat"]["id"]
        is_group = message["chat"]["type"] in ["group", "supergroup"]
        message_id = message["message_id"]
        from_user = message["from"]
        username = from_user.get("username") or from_user.get("first_name")
        name = from_user.get("first_name") or "" + from_user.get("last_name") or ""
        user_full_name = name if name != "" else from_user.get("username")
        pushName = username

        if redis_client.exists(f"telegram:{message_id}"):
            return Response({"message": "loop message"}, status=200)
        redis_client.setex(f"telegram:{message_id}", 600, message_id)

        try:
            if bot.line:
                group_title = message["chat"].get("title") if is_group else None
                attachments = None

                media_keys = [
                    "photo",
                    "document",
                    "video",
                    "audio",
                    "voice",
                    "video_note"
                ]

                text = message.get("text", " ")

                for key in media_keys:
                    if key in message:
                        attachments = []
                        if key == "photo":
                            items = [message[key][-1]]
                        else:
                            items = message[key] if isinstance(message[key], list) else [message[key]]
                        text = message.get("capture", " ")

                        for item in items:
                            file_id = item["file_id"]

                            tg_response = requests.get(
                                f"https://api.telegram.org/bot{bot.bot_token}/getFile?file_id={file_id}"
                            )
                            tg_response.raise_for_status()
                            file_info = tg_response.json().get("result", {})
                            file_path = file_info.get("file_path")

                            if not file_path:
                                continue

                            file_url = f"https://api.telegram.org/file/bot{bot.bot_token}/{file_path}"
                            filename = file_path.split("/")[-1]

                            attachments.append({
                                "url": file_url,
                                "name": filename,
                                "type": key
                            })

                if is_group:
                    text_to_send = f"*{group_title}*\n{user_full_name}: {text}"
                    chat_key = f'bitrix_chat:telegram_group:{chat_id}'
                    redis_client.setex(chat_key, 60 * 60 * 24, chat_id)
                else:
                    text_to_send = text

                bitrix_tasks.send_messages(
                    bot.app_instance.id,
                    chat_id,
                    text_to_send,
                    bot.line.connector.code,
                    bot.line.line_id,
                    False,
                    pushName,
                    str(message_id),
                    attachments=attachments,
                )

        except Exception as e:
            logger.error(f"Bitrix send error: {e}")

        return Response({"message": "processed"}, status=200)

    @action(detail=False, methods=['post'], url_path=r'(?P<session>[^/.]+)/send')
    def send(self, request, session=None, *args, **kwargs):
        """
        Send outgoing message to Telegram using session id
        """
        session_id = session

        if not session_id:
            return Response({'error': 'session is required'})

        try:
            bot = TelegramBot.objects.get(id=session_id)
        except Exception as e:
            return Response({'error': 'An error occurred', 'details': str(e)})

        if bot.date_end and timezone.now() > bot.date_end:
            return Response({'error': 'tariff has expired'}, status=402)

        data = request.data
        event = data.get('event')
        message_type = data.get('message_type')
        attachments = data.get('attachments', {})

        if event == "message_created" and message_type == "outgoing":
            message_id = data.get('id')

            if redis_client.exists(f'chatwoot:{message_id}'):
                return Response({'message': 'loop message'})

            redis_client.setex(f'chatwoot:{message_id}', 600, message_id)

            content = data.get('content')
            conversation = data.get('conversation', {})
            meta = conversation.get('meta', {})
            sender = meta.get('sender', {})

            chat_id = sender.get('chat_id')
            if not chat_id:
                return Response({'error': 'chat_id not provided in sender meta'})

            if content:
                telegram_tasks.send_telegram_message_task.delay(
                    bot_token=bot.bot_token,
                    recipient=chat_id,
                    content=content,
                    cont_type="string"
                )

                if bot.line:
                    bitrix_tasks.message_add.delay(
                        bot.app_instance.id,
                        bot.line.line_id,
                        str(chat_id),
                        content,
                        bot.line.connector.code
                    )

            if attachments:
                for attachment in attachments:
                    telegram_tasks.send_telegram_message_task.delay(
                        bot_token=bot.bot_token,
                        recipient=chat_id,
                        content=attachment,
                        cont_type="media"
                    )

                return Response({'message': 'All files sent successfully'})
        return Response({'message': f'Session {session_id} authorized'})

Создадим асинхронную функцию для отправки сообщений в tasks.py, выполним проверку на тип сообщения (текст или медиа файл):

import base64
import os
import tempfile

import requests
from celery import shared_task


@shared_task
def send_telegram_message_task(bot_token, recipient, content, cont_type="string"):
    """
    Отправляет сообщение или медиа в Telegram.
    recipient: chat_id (str или int) или @username
    content: строка (текст) или dict с base64-данными
    cont_type: 'string' или 'media'
    """
    print("send_telegram_message_task is called")
    TELEGRAM_API_BASE = f"https://api.telegram.org/bot{bot_token}"
    if cont_type == "string":
        url = f"{TELEGRAM_API_BASE}/sendMessage"
        payload = {
            "chat_id": recipient,
            "text": content,
            "parse_mode": "Markdown",  # или HTML
            "disable_web_page_preview": False,
        }
        response = requests.post(url, json=payload, timeout=30)
        response.raise_for_status()
        return response.json()

    elif cont_type == "media":
        mimetype = content.get("mimetype", "")
        base_type = mimetype.split("/")[0]
        file_url = content.get("url") or content.get("link")
        file_name = content.get("filename") or content.get("name", "file")

        if not file_url:
            raise ValueError("Media content must include 'link' or 'url'")

        try:
            file_response = requests.get(file_url, stream=True, timeout=30)
            file_response.raise_for_status()

            with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
                for chunk in file_response.iter_content(chunk_size=8192):
                    tmp_file.write(chunk)
                tmp_file_path = tmp_file.name
            files = {}
            data = {"chat_id": recipient}

            if base_type == "image":
                url = f"{TELEGRAM_API_BASE}/sendPhoto"
                files["photo"] = (file_name, open(tmp_file_path, "rb"), mimetype)
                data["caption"] = ""

            else:
                url = f"{TELEGRAM_API_BASE}/sendDocument"
                files["document"] = (file_name, open(tmp_file_path, "rb"), mimetype)
                data["caption"] = ""

            response = requests.post(url, data=data, files=files, timeout=60)
            response.raise_for_status()
            return response.json()

        finally:
            if os.path.exists(tmp_file_path):
                os.remove(tmp_file_path)

Не вижу смысла в реализации отдельных страниц для подключения Telegram бота, ввиду наличия "из коробки" прекрасной админ панели в django, поэтому создадим лишь страницу с отображением подключенных ботов:

import requests
import redis
import uuid
from requests.exceptions import RequestException
from django.shortcuts import render, redirect, get_object_or_404
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from django.db.models import Count, F
from django.utils import timezone
from django.conf import settings

from bitrix.models import AppInstance, Line, Connector
import bitrix.utils as bitrix_utils

from main.decorators import login_message_required

from .models import TelegramBot

apps = settings.INSTALLED_APPS

redis_client = redis.StrictRedis(host=settings.REDIS_HOST, port=6379, db=0, decode_responses=True)

LINK_TTL = 60 * 60 * 24


@login_message_required(code="telegram")
def tg_bots(request):
    connector_service = "telegram"
    connector = Connector.objects.filter(service=connector_service).first()
    if request.method == "POST":
        session_id = request.POST.get("session_id")
        line_id = request.POST.get("line_id")
        if not line_id:
            messages.warning(request, "Необходимо выбрать линию из списка или создать новую.")
            return redirect('telegram')
        bot = get_object_or_404(TelegramBot, id=session_id, owner=request.user)
        if not bot.bot_username:
            messages.error(request, "Сначала необходимо подключить WhatsApp.")
            return redirect('telegram')
        if bot.line and str(bot.line.id) == str(line_id):
            messages.warning(request, "Эта линия уже подключена к выбранной сессии.")
            return redirect('telegram')
        try:
            bitrix_utils.connect_line(request, line_id, bot, connector, connector_service)
        except Exception as e:
            messages.error(request, str(e))
            return redirect('telegram')
        return redirect('telegram')

    bots = TelegramBot.objects.filter(owner=request.user)
    instances = AppInstance.objects.filter(owner=request.user, app__connectors=connector)
    tg_lines = Line.objects.filter(connector=connector, owner=request.user)

    return render(
        request, 'telegram/tg_sessions.html', {
            "bots": bots,
            "instances": instances,
            "tg_lines": tg_lines,
        }
    )

На этом реализация нашего коннектора завершена, а в третьей части мы рассмотрим подключение Локальных приложений к Открытым линиям CRM Bitrix24.

WhatsApp Web и Telegram коннектор для Bitrix24: наш опыт реализации и внедрения. Часть 2 — WhatsApp и Telegram коннектор