Привет, мир! Меня зовут Павел, я 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.