Интеграция API WhatsApp и Discord с Python: мощная автоматизация

Пройдите тест, узнайте какой профессии подходите
Сколько вам лет
0%
До 18
От 18 до 24
От 25 до 34
От 35 до 44
От 45 до 49
От 50 до 54
Больше 55

Для кого эта статья:

  • Разработчики и программисты, интересующиеся интеграцией API и автоматизацией коммуникационных процессов.
  • Студенты и профессионалы, обучающиеся Python и желающие улучшить свои навыки разработки.
  • Руководители и владельцы бизнеса, стремящиеся оптимизировать взаимодействие с клиентами через мессенджеры и платформы.

    Интеграция API WhatsApp и Discord через Python открывает безграничные возможности для автоматизации коммуникаций. 🚀 Представьте: утром клиент отправляет запрос в WhatsApp, и через минуту ваша команда получает уведомление в Discord, а система уже готовит автоматический ответ. Без ручного копирования, без задержек, без человеческого фактора. В этом руководстве я раскрою все технические аспекты такой интеграции — от базовой настройки до продвинутых кейсов с полноценными примерами кода и практическими советами, которые я накопил за годы разработки коммуникационных систем.

Погружаясь в мир API-интеграций с Python, вы открываете путь к высокооплачиваемой профессии разработчика. Курс Обучение Python-разработке от Skypro фокусируется именно на практических навыках работы с API, веб-сервисами и автоматизацией. Студенты создают реальные проекты с интеграцией мессенджеров, осваивают RESTful API и асинхронное программирование — всё то, что делает разработчика востребованным на рынке с зарплатой от 150 000 рублей.

Основы интеграции Python с API WhatsApp и Discord

Перед погружением в код необходимо понять архитектурные особенности API WhatsApp и Discord. Эти платформы используют различные подходы к авторизации, обработке данных и методам взаимодействия, что требует разных стратегий при интеграции.

WhatsApp предоставляет API через Business API Client (WABiz), который требует официальной верификации бизнес-аккаунта. Discord, напротив, имеет открытый API с относительно простой регистрацией приложения через Developer Portal.

Параметр WhatsApp API Discord API
Тип API RESTful + Webhooks RESTful + WebSockets
Авторизация OAuth 2.0 + API Key Bot Token
Лимиты запросов ~80 запросов/сек ~50 запросов/сек
Сложность интеграции Высокая Средняя
Документация Детализированная Обширная

Основные шаги интеграции с API WhatsApp:

  • Регистрация бизнес-аккаунта и получение доступа к API
  • Настройка webhook-сервера для приема уведомлений
  • Реализация механизмов отправки и получения сообщений
  • Обработка различных типов медиа-контента

Для Discord API процесс выглядит иначе:

  • Создание приложения в Discord Developer Portal
  • Генерация и настройка токена бота
  • Определение необходимых разрешений (permissions)
  • Подключение через WebSocket для отслеживания событий
  • Использование REST API для взаимодействия с серверами

Важно понимать, что оба API постоянно обновляются, и методы, которые работали год назад, могут требовать корректировки сегодня. Python, как язык с богатой экосистемой библиотек для работы с API, позволяет относительно безболезненно адаптироваться к этим изменениям.

Алексей Савин, Lead Backend Developer Столкнулся с необходимостью объединить коммуникации через WhatsApp и Discord, когда разрабатывал систему уведомлений для финтех-компании. Клиенты предпочитали WhatsApp, а команда разработки — Discord. Интеграция заняла неделю, но экономия времени была колоссальной — 6-7 часов ежедневно команда больше не тратила на перенос сообщений вручную. Ключом к успеху стало понимание асинхронной природы обоих API и правильная обработка очередей сообщений. Не пытайтесь синхронизировать всё напрямую — используйте промежуточное хранилище и механизмы очередей.

Пошаговый план для смены профессии

Настройка окружения для работы с API мессенджеров

Перед началом работы с API WhatsApp и Discord необходимо правильно настроить среду разработки. Python, благодаря своей гибкости и обширной экосистеме библиотек, идеально подходит для таких задач. 🛠️

Установка необходимых библиотек для работы с обоими API:

pip install requests aiohttp discord.py python-dotenv pywhatkit

Для изолированной разработки рекомендую использовать виртуальное окружение:

python -m venv venv
source venv/bin/activate # Linux/MacOS
venv\Scripts\activate.bat # Windows

Структура проекта для интеграции должна быть организована логически:

  • config/ — файлы конфигурации и переменные окружения
  • whatsapp/ — модули для работы с WhatsApp API
  • discord/ — компоненты для взаимодействия с Discord API
  • utils/ — общие утилиты и хелперы
  • integration/ — модули связи между платформами
  • main.py — точка входа приложения

Для безопасного хранения ключей API и токенов создайте файл .env в корне проекта:

# .env
WHATSAPP_API_KEY=your_whatsapp_api_key
WHATSAPP_PHONE_ID=your_whatsapp_phone_id
DISCORD_BOT_TOKEN=your_discord_bot_token
WEBHOOK_VERIFY_TOKEN=your_custom_verify_token

Загрузка переменных окружения в коде:

Python
Скопировать код
# config/settings.py
import os
from dotenv import load_dotenv

load_dotenv()

WHATSAPP_API_KEY = os.getenv("WHATSAPP_API_KEY")
WHATSAPP_PHONE_ID = os.getenv("WHATSAPP_PHONE_ID")
DISCORD_BOT_TOKEN = os.getenv("DISCORD_BOT_TOKEN")
WEBHOOK_VERIFY_TOKEN = os.getenv("WEBHOOK_VERIFY_TOKEN")

Для тестирования API локально потребуется настроить туннелирование с помощью ngrok или аналогичного сервиса, чтобы веб-хуки могли достичь вашего локального сервера:

# Установка ngrok
pip install pyngrok

# В коде для автоматического запуска
from pyngrok import ngrok
public_url = ngrok.connect(8000) # Порт вашего локального сервера
print(f"Public URL: {public_url}")

Настройка базовой обработки вебхуков с Flask:

Python
Скопировать код
# webhook_server.py
from flask import Flask, request, jsonify
from config.settings import WEBHOOK_VERIFY_TOKEN

app = Flask(__name__)

@app.route('/webhook', methods=['GET'])
def verify_webhook():
if request.args.get('hub.verify_token') == WEBHOOK_VERIFY_TOKEN:
return request.args.get('hub.challenge')
return 'Invalid verification token'

@app.route('/webhook', methods=['POST'])
def receive_message():
data = request.json
# Логика обработки входящего сообщения от WhatsApp
return jsonify({"status": "received"})

if __name__ == '__main__':
app.run(port=8000, debug=True)

Для работы с Discord API рекомендуется создать отдельный класс-клиент, который будет управлять соединением:

Python
Скопировать код
# discord/client.py
import discord
from discord.ext import commands
from config.settings import DISCORD_BOT_TOKEN

class DiscordClient(commands.Bot):
def __init__(self):
intents = discord.Intents.default()
intents.message_content = True
super().__init__(command_prefix="!", intents=intents)

async def on_ready(self):
print(f'Logged in as {self.user.name}')

async def setup_hook(self):
# Загрузка дополнительных команд и модулей
await self.load_extension("discord.commands")

client = DiscordClient()

Этап настройки Потенциальные проблемы Решения
Получение API ключей Задержка верификации бизнес-аккаунта WhatsApp Использовать тестовые номера или альтернативные библиотеки
Настройка вебхуков Локальный сервер недоступен из интернета Применение ngrok или размещение на облачном сервере
Авторизация бота Discord Недостаточные разрешения Пересмотреть и обновить scope разрешений в Developer Portal
Обработка событий Пропущенные уведомления при перезагрузке Реализация очередей сообщений (Redis/RabbitMQ)

Перед запуском в production настоятельно рекомендуется настроить логирование для отслеживания взаимодействия с API:

Python
Скопировать код
# utils/logging.py
import logging

def setup_logger():
logger = logging.getLogger('messenger_integration')
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)

logger.addHandler(handler)
return logger

logger = setup_logger()

Python-код для автоматизации сообщений в WhatsApp

Разработка функционала для автоматизации WhatsApp требует понимания специфики его API и правильного применения Python-библиотек. Рассмотрим основные механизмы отправки и обработки сообщений через WhatsApp Business API. 💬

Базовая функция для отправки текстовых сообщений:

Python
Скопировать код
# whatsapp/messaging.py
import requests
import json
from config.settings import WHATSAPP_API_KEY, WHATSAPP_PHONE_ID
from utils.logging import logger

def send_whatsapp_message(phone_number, message_text):
url = f"https://graph.facebook.com/v13.0/{WHATSAPP_PHONE_ID}/messages"

headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {WHATSAPP_API_KEY}"
}

data = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": phone_number,
"type": "text",
"text": {
"body": message_text
}
}

try:
response = requests.post(url, headers=headers, data=json.dumps(data))
response.raise_for_status()
logger.info(f"Message sent to {phone_number}")
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send message: {str(e)}")
return None

Для отправки медиа-файлов (изображений, документов, аудио) требуется другой формат запроса:

Python
Скопировать код
def send_whatsapp_media(phone_number, media_type, media_url, caption=None):
url = f"https://graph.facebook.com/v13.0/{WHATSAPP_PHONE_ID}/messages"

headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {WHATSAPP_API_KEY}"
}

media_object = {
"link": media_url
}

if caption:
media_object["caption"] = caption

data = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": phone_number,
"type": media_type,
media_type: media_object
}

try:
response = requests.post(url, headers=headers, data=json.dumps(data))
response.raise_for_status()
logger.info(f"{media_type.capitalize()} sent to {phone_number}")
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send {media_type}: {str(e)}")
return None

Интерактивные шаблоны сообщений — мощный инструмент для создания структурированных сообщений с кнопками:

Python
Скопировать код
def send_whatsapp_template(phone_number, template_name, language_code="en_US", components=None):
url = f"https://graph.facebook.com/v13.0/{WHATSAPP_PHONE_ID}/messages"

headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {WHATSAPP_API_KEY}"
}

data = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": phone_number,
"type": "template",
"template": {
"name": template_name,
"language": {
"code": language_code
}
}
}

if components:
data["template"]["components"] = components

try:
response = requests.post(url, headers=headers, data=json.dumps(data))
response.raise_for_status()
logger.info(f"Template sent to {phone_number}")
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send template: {str(e)}")
return None

Для обработки входящих сообщений необходим webhook-обработчик:

Python
Скопировать код
# whatsapp/webhook_handlers.py
from flask import request, jsonify
from utils.logging import logger
from integration.message_queue import process_incoming_message

def handle_whatsapp_webhook():
data = request.json

try:
# Проверяем, содержит ли webhook данные о сообщении
if 'entry' in data and data['entry'] and 'changes' in data['entry'][0]:
changes = data['entry'][0]['changes']

for change in changes:
if 'value' in change and 'messages' in change['value']:
messages = change['value']['messages']

for message in messages:
sender = message['from']
message_id = message['id']
timestamp = message['timestamp']

if 'text' in message:
message_text = message['text']['body']
logger.info(f"Received message from {sender}: {message_text}")

# Обработка сообщения и отправка в очередь для интеграции с Discord
process_incoming_message({
'platform': 'whatsapp',
'sender': sender,
'message': message_text,
'timestamp': timestamp,
'message_id': message_id
})

# Обработка других типов сообщений (медиа, локация и т.д.)
elif 'image' in message:
# Логика для изображений
pass
elif 'document' in message:
# Логика для документов
pass

return jsonify({"status": "success"}), 200
except Exception as e:
logger.error(f"Error processing webhook: {str(e)}")
return jsonify({"status": "error", "message": str(e)}), 500

Автоматизация часто повторяющихся задач с помощью планировщика:

Python
Скопировать код
# whatsapp/scheduler.py
import schedule
import time
import threading
from whatsapp.messaging import send_whatsapp_message
from utils.logging import logger

def schedule_recurring_message(phone_number, message, time_str, days=None):
def job():
logger.info(f"Sending scheduled message to {phone_number}")
send_whatsapp_message(phone_number, message)

if days:
for day in days:
schedule.every().day.at(time_str).do(job)
else:
schedule.every().day.at(time_str).do(job)

logger.info(f"Message scheduled for {phone_number} at {time_str}")

def run_scheduler():
while True:
schedule.run_pending()
time.sleep(60) # Проверка каждую минуту

def start_scheduler():
scheduler_thread = threading.Thread(target=run_scheduler)
scheduler_thread.daemon = True
scheduler_thread.start()

Для обработки большого количества сообщений рекомендуется реализовать очередь:

Python
Скопировать код
# whatsapp/message_queue.py
import queue
import threading
from whatsapp.messaging import send_whatsapp_message
from utils.logging import logger

# Очередь сообщений
message_queue = queue.Queue()

def enqueue_message(phone_number, message):
message_queue.put((phone_number, message))
logger.info(f"Message to {phone_number} added to queue")

def process_queue():
while True:
try:
phone_number, message = message_queue.get(timeout=1)
logger.info(f"Processing queued message for {phone_number}")
send_whatsapp_message(phone_number, message)
message_queue.task_done()
except queue.Empty:
# Очередь пуста, ждем новых сообщений
pass
except Exception as e:
logger.error(f"Error processing queued message: {str(e)}")

def start_queue_processor(num_workers=2):
for _ in range(num_workers):
worker = threading.Thread(target=process_queue)
worker.daemon = True
worker.start()
logger.info(f"{num_workers} message queue workers started")

Марина Котова, Python-разработчик Наш e-commerce проект столкнулся с лавиной запросов в службу поддержки после запуска нового продукта. Клиенты писали в WhatsApp, а нашему небольшому отделу поддержки приходилось обрабатывать сотни однотипных вопросов. Я разработала Python-скрипт для автоматического ответа на 80% типовых запросов. Ключевым моментом стала классификация сообщений с использованием NLP и SpaCy: система определяла намерение пользователя и выбирала соответствующий шаблон ответа. Интеграция с WhatsApp API сократила время ответа с нескольких часов до секунд. Нагрузка на службу поддержки снизилась на 68%, а удовлетворенность клиентов выросла на 22%.

Создание функциональных ботов для Discord с Python

Разработка ботов для Discord с использованием Python существенно упрощается благодаря библиотеке discord.py. Эта библиотека предоставляет высокоуровневый API для взаимодействия с Discord API, обработки событий и управления серверами. 🤖

Базовая структура бота Discord с обработкой команд:

Python
Скопировать код
# discord_bot.py
import discord
from discord.ext import commands
import asyncio
from config.settings import DISCORD_BOT_TOKEN
from utils.logging import logger

# Настройка intents для бота
intents = discord.Intents.default()
intents.message_content = True # Для чтения содержимого сообщений
intents.members = True # Для работы с участниками сервера

# Создание экземпляра бота
bot = commands.Bot(command_prefix='!', intents=intents)

@bot.event
async def on_ready():
logger.info(f'Bot logged in as {bot.user.name}')
# Установка статуса бота
activity = discord.Game(name="Мониторинг WhatsApp")
await bot.change_presence(activity=activity)

@bot.command(name="ping")
async def ping(ctx):
"""Простая команда для проверки работоспособности бота"""
await ctx.send(f"Pong! Задержка: {round(bot.latency * 1000)}ms")

@bot.command(name="whatsapp")
async def send_to_whatsapp(ctx, phone_number: str, *, message: str):
"""Отправляет сообщение в WhatsApp через бота"""
from whatsapp.messaging import send_whatsapp_message

# Проверка формата номера телефона
if not phone_number.isdigit() or len(phone_number) < 10:
await ctx.send("Некорректный номер телефона. Используйте формат с кодом страны, например: 79XXXXXXXXX")
return

try:
# Отправка сообщения в WhatsApp
result = send_whatsapp_message(phone_number, message)
if result:
await ctx.send(f"✅ Сообщение успешно отправлено на номер {phone_number}")
else:
await ctx.send("❌ Ошибка при отправке сообщения. Проверьте логи.")
except Exception as e:
await ctx.send(f"❌ Произошла ошибка: {str(e)}")
logger.error(f"Error sending WhatsApp message: {str(e)}")

@bot.event
async def on_message(message):
# Игнорировать сообщения от ботов
if message.author.bot:
return

# Обработка специальных команд или сообщений
# Например, можно добавить реакцию на определенные ключевые слова

# Важно: это позволяет обрабатывать команды вместе с обычными сообщениями
await bot.process_commands(message)

# Функция для запуска бота
def run_discord_bot():
bot.run(DISCORD_BOT_TOKEN)

Создание расширенных команд с аргументами и проверками:

Python
Скопировать код
# discord/commands.py
from discord.ext import commands
import discord
from whatsapp.messaging import send_whatsapp_message, send_whatsapp_media
from utils.logging import logger

class WhatsAppCommands(commands.Cog):
def __init__(self, bot):
self.bot = bot

@commands.command(name="status")
@commands.has_permissions(administrator=True)
async def status(self, ctx):
"""Показывает текущий статус интеграции с WhatsApp"""
embed = discord.Embed(
title="Статус интеграции",
description="Информация о текущем состоянии системы",
color=discord.Color.blue()
)

# Здесь можно добавить проверку доступности API WhatsApp
embed.add_field(name="WhatsApp API", value="✅ Подключено", inline=False)
embed.add_field(name="Обработано сообщений", value="1,234", inline=True)
embed.add_field(name="Активных пользователей", value="567", inline=True)

await ctx.send(embed=embed)

@commands.command(name="broadcast")
@commands.has_permissions(administrator=True)
async def broadcast(self, ctx, *, message: str):
"""Отправляет сообщение всем активным пользователям WhatsApp"""
# В реальном сценарии здесь должна быть логика получения списка активных номеров
phone_numbers = ["79XXXXXXXXX", "79YYYYYYYYY"] # Пример

sent_count = 0
failed_count = 0

# Создаем сообщение о начале рассылки
status_message = await ctx.send("Начинаем рассылку сообщений...")

for phone in phone_numbers:
try:
result = send_whatsapp_message(phone, message)
if result:
sent_count += 1
else:
failed_count += 1

# Обновляем статус каждые 5 отправленных сообщений
if (sent_count + failed_count) % 5 == 0:
await status_message.edit(content=f"Отправлено: {sent_count}, Ошибок: {failed_count}")

except Exception as e:
logger.error(f"Error in broadcast to {phone}: {str(e)}")
failed_count += 1

await status_message.edit(content=f"✅ Рассылка завершена! Отправлено: {sent_count}, Ошибок: {failed_count}")

async def setup(bot):
await bot.add_cog(WhatsAppCommands(bot))

Реализация системы слешкоманд (slash commands) для современного интерфейса Discord:

Python
Скопировать код
# discord/slash_commands.py
import discord
from discord import app_commands
from discord.ext import commands
from whatsapp.messaging import send_whatsapp_message
from utils.logging import logger

class WhatsAppSlashCommands(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.ctx_menu = app_commands.ContextMenu(
name='Отправить в WhatsApp',
callback=self.send_to_whatsapp_context,
)
self.bot.tree.add_command(self.ctx_menu)

@app_commands.command(name="whatsapp_send", description="Отправить сообщение в WhatsApp")
@app_commands.describe(
phone="Номер телефона получателя в формате с кодом страны",
message="Текст сообщения для отправки"
)
async def whatsapp_send(self, interaction: discord.Interaction, phone: str, message: str):
await interaction.response.defer(ephemeral=True)

try:
result = send_whatsapp_message(phone, message)
if result:
await interaction.followup.send(f"✅ Сообщение успешно отправлено на номер {phone}", ephemeral=True)
else:
await interaction.followup.send("❌ Ошибка при отправке сообщения", ephemeral=True)
except Exception as e:
logger.error(f"Error in slash command: {str(e)}")
await interaction.followup.send(f"❌ Произошла ошибка: {str(e)}", ephemeral=True)

async def send_to_whatsapp_context(self, interaction: discord.Interaction, message: discord.Message):
"""Контекстное меню для отправки сообщения в WhatsApp"""
# Открываем модальное окно для ввода номера телефона
modal = WhatsAppPhoneModal(message.content)
await interaction.response.send_modal(modal)

class WhatsAppPhoneModal(discord.ui.Modal, title="Отправка в WhatsApp"):
def __init__(self, message_content):
super().__init__()
self.message_content = message_content

self.phone = discord.ui.TextInput(
label="Номер телефона",
placeholder="79XXXXXXXXX",
required=True,
min_length=10,
max_length=15
)
self.add_item(self.phone)

async def on_submit(self, interaction: discord.Interaction):
await interaction.response.defer(ephemeral=True)

try:
result = send_whatsapp_message(self.phone.value, self.message_content)
if result:
await interaction.followup.send(f"✅ Сообщение успешно отправлено на номер {self.phone.value}", ephemeral=True)
else:
await interaction.followup.send("❌ Ошибка при отправке сообщения", ephemeral=True)
except Exception as e:
logger.error(f"Error in modal submission: {str(e)}")
await interaction.followup.send(f"❌ Произошла ошибка: {str(e)}", ephemeral=True)

async def setup(bot):
await bot.add_cog(WhatsAppSlashCommands(bot))

Создание специализированного канала для мониторинга сообщений WhatsApp в Discord:

Python
Скопировать код
# discord/monitoring.py
from discord.ext import commands, tasks
import discord
import datetime
from utils.logging import logger
from whatsapp.webhook_handlers import get_recent_messages

class WhatsAppMonitor(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.monitor_channel_id = None # ID канала нужно установить через команду
self.last_check_time = datetime.datetime.now()
self.check_new_messages.start()

def cog_unload(self):
self.check_new_messages.cancel()

@commands.command(name="set_monitor")
@commands.has_permissions(administrator=True)
async def set_monitor_channel(self, ctx, channel: discord.TextChannel = None):
"""Устанавливает текущий канал как канал мониторинга WhatsApp"""
if channel is None:
channel = ctx.channel

self.monitor_channel_id = channel.id
logger.info(f"Monitor channel set to {channel.name} (ID: {channel.id})")
await ctx.send(f"✅ Канал {channel.mention} установлен для мониторинга сообщений WhatsApp")

@tasks.loop(minutes=1)
async def check_new_messages(self):
"""Проверяет новые сообщения WhatsApp и отправляет их в канал мониторинга"""
if not self.monitor_channel_id:
return

try:
current_time = datetime.datetime.now()
new_messages = get_recent_messages(self.last_check_time)
self.last_check_time = current_time

if not new_messages:
return

channel = self.bot.get_channel(self.monitor_channel_id)
if not channel:
logger.error(f"Monitor channel with ID {self.monitor_channel_id} not found")
return

for msg in new_messages:
embed = discord.Embed(
title="Новое сообщение WhatsApp",
description=msg["message"],
color=discord.Color.green(),
timestamp=datetime.datetime.fromtimestamp(msg["timestamp"])
)
embed.set_author(name=f"От: {msg['sender']}")
embed.set_footer(text=f"Message ID: {msg['message_id']}")

await channel.send(embed=embed)

except Exception as e:
logger.error(f"Error in monitor task: {str(e)}")

@check_new_messages.before_loop
async def before_checking(self):
await self.bot.wait_until_ready()

async def setup(bot):
await bot.add_cog(WhatsAppMonitor(bot))

Практические кейсы объединения WhatsApp и Discord API

Интеграция WhatsApp и Discord может значительно улучшить коммуникационные процессы компании. Рассмотрим конкретные кейсы, где такая синергия даёт максимальный эффект. 🔄

Кейс 1: Создание единого центра поддержки клиентов

Python
Скопировать код
# integration/support_center.py
import asyncio
from discord.ext import commands
import discord
import re
from whatsapp.messaging import send_whatsapp_message
from config.settings import DISCORD_SUPPORT_CHANNEL_ID
from utils.logging import logger

class SupportCenterIntegration(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.active_tickets = {} # Хранение активных тикетов: {ticket_id: {customer_phone, thread_id}}

@commands.Cog.listener()
async def on_whatsapp_message(self, message_data):
"""Обрабатывает входящее сообщение из WhatsApp и создаёт тикет в Discord"""
phone = message_data['sender']
message_text = message_data['message']
timestamp = message_data['timestamp']

# Проверяем, существует ли уже тикет для этого номера
ticket_id = None
for tid, data in self.active_tickets.items():
if data['customer_phone'] == phone:
ticket_id = tid
break

support_channel = self.bot.get_channel(DISCORD_SUPPORT_CHANNEL_ID)
if not support_channel:
logger.error(f"Support channel not found: {DISCORD_SUPPORT_CHANNEL_ID}")
return

if ticket_id:
# Добавляем сообщение к существующему тикету
thread = await support_channel.fetch_thread(self.active_tickets[ticket_id]['thread_id'])

embed = discord.Embed(
description=message_text,
color=discord.Color.green(),
timestamp=datetime.datetime.fromtimestamp(timestamp)
)
embed.set_author(name=f"Клиент ({phone})")

await thread.send(embed=embed)
else:
# Создаём новый тикет
ticket_id = f"T-{int(timestamp)}"

embed = discord.Embed(
title=f"Новый запрос в поддержку: {ticket_id}",
description=message_text,
color=discord.Color.blue(),
timestamp=datetime.datetime.fromtimestamp(timestamp)
)
embed.add_field(name="Клиент", value=phone)

message = await support_channel.send(embed=embed)
thread = await message.create_thread(name=f"Запрос {ticket_id}")

# Сохраняем информацию о тикете
self.active_tickets[ticket_id] = {
'customer_phone': phone,
'thread_id': thread.id
}

# Отправляем автоответ клиенту
await self.send_auto_response(phone, ticket_id)

async def send_auto_response(self, phone, ticket_id):
"""Отправляет автоматический ответ клиенту в WhatsApp"""
response_message = f"Спасибо за обращение! Ваш запрос зарегистрирован под номером {ticket_id}. Специалист ответит вам в ближайшее время."

try:
send_whatsapp_message(phone, response_message)
logger.info(f"Auto-response sent to {phone} for ticket {ticket_id}")
except Exception as e:
logger.error(f"Error sending auto-response: {str(e)}")

@commands.command(name="reply")
async def reply_to_customer(self, ctx, *, message: str):
"""Отвечает клиенту из Discord-треда в WhatsApp"""
# Проверяем, что команда вызвана из треда
if not isinstance(ctx.channel, discord.Thread):
await ctx.send("❌ Эта команда должна использоваться в треде тикета")
return

# Ищем тикет по ID треда
thread_id = ctx.channel.id
ticket_id = None
customer_phone = None

for tid, data in self.active_tickets.items():
if data['thread_id'] == thread_id:
ticket_id = tid
customer_phone = data['customer_phone']
break

if not ticket_id or not customer_phone:
await ctx.send("❌ Не удалось найти информацию о тикете")
return

try:
# Отправляем ответ клиенту в WhatsApp
send_whatsapp_message(customer_phone, message)

# Отображаем ответ в треде Discord
embed = discord.Embed(
description=message,
color=discord.Color.blue(),
timestamp=datetime.datetime.now()
)
embed.set_author(name=f"Ответ от: {ctx.author.display_name}")

await ctx.send(embed=embed)
logger.info(f"Reply sent to {customer_phone} for ticket {ticket_id}")
except Exception as e:
logger.error(f"Error sending reply: {str(e)}")
await ctx.send(f"❌ Ошибка при отправке сообщения: {str(e)}")

@commands.command(name="close_ticket")
@commands.has_permissions(manage_messages=True)
async def close_ticket(self, ctx, ticket_id: str = None):
"""Закрывает тикет поддержки"""
if not isinstance(ctx.channel, discord.Thread):
await ctx.send("❌ Эта команда должна использоваться в треде тикета")
return

thread_id = ctx.channel.id

# Если ticket_id не указан, ищем по ID треда
if not ticket_id:
for tid, data in self.active_tickets.items():
if data['thread_id'] == thread_id:
ticket_id = tid
break

if not ticket_id or ticket_id not in self.active_tickets:
await ctx.send("❌ Указан неверный ID тикета")
return

customer_phone = self.active_tickets[ticket_id]['customer_phone']

# Отправляем уведомление клиенту
closing_message = f"Ваш запрос {ticket_id} закрыт. Спасибо за обращение в службу поддержки!"

try:
send_whatsapp_message(customer_phone, closing_message)

# Архивируем тред
await ctx.channel.edit(archived=True, locked=True)

# Удаляем тикет из активных
del self.active_tickets[ticket_id]

logger.info(f"Ticket {ticket_id} closed by {ctx.author.name}")
except Exception as e:
logger.error(f"Error closing ticket: {str(e)}")
await ctx.send(f"❌ Ошибка при закрытии тикета: {str(e)}")

Кейс 2: Система мониторинга и оповещения

Python
Скопировать код
# integration/monitoring_system.py
import asyncio
from discord.ext import commands, tasks
import discord
import json
import os
from datetime import datetime, timedelta
from utils.logging import logger

class MonitoringSystem(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.alerts_channel_id = None # ID канала для оповещений
self.whatsapp_stats = {
'messages_received': 0,
'messages_sent': 0,
'errors': 0,
'active_users': set()
}
self.stats_file = "data/whatsapp_stats.json"
self.load_stats()
self.update_stats_display.start()
self.save_stats_periodic.start()

def cog_unload(self):
self.update_stats_display.cancel()
self.save_stats_periodic.cancel()
self.save_stats()

def load_stats(self):
"""Загружает статистику из файла"""
try:
if os.path.exists(self.stats_file):
with open(self.stats_file, 'r') as f:
data = json.load(f)
self.whatsapp_stats['messages_received'] = data.get('messages_received', 0)
self.whatsapp_stats['messages_sent'] = data.get('messages_sent', 0)
self.whatsapp_stats['errors'] = data.get('errors', 0)
self.whatsapp_stats['active_users'] = set(data.get('active_users', []))
logger.info("WhatsApp stats loaded from file")
except Exception as e:
logger.error(f"Error loading stats: {str(e)}")

def save_stats(self):
"""Сохраняет статистику в файл"""
try:
os.makedirs(os.path.dirname(self.stats_file), exist_ok=True)
with open(self.stats_file, 'w') as f:
data = {
'messages_received': self.whatsapp_stats['messages_received'],
'messages_sent': self.whatsapp_stats['messages_sent'],
'errors': self.whatsapp_stats['errors'],
'active_users': list(self.whatsapp_stats['active_users'])
}
json.dump(data, f)
logger.info("WhatsApp stats saved to file")
except Exception as e:
logger.error(f"Error saving stats: {str(e)}")

@commands.Cog.listener()
async def on_whatsapp_message(self, message_data):
"""Обновляет статистику при получении сообщения из WhatsApp"""
self.whatsapp_stats['messages_received'] += 1
self.whatsapp_stats['active_users'].add(message_data['sender'])

@commands.Cog.listener()
async def on_whatsapp_send(self, phone, success):
"""Обновляет статистику при отправке сообщения в WhatsApp"""
if success:
self.whatsapp_stats['messages_sent'] += 1
else:
self.whatsapp_stats['errors'] += 1

@commands.command(name="set_alerts")
@commands.has_permissions(administrator=True)
async def set_alerts_channel(self, ctx, channel: discord.TextChannel = None):
"""Устанавливает канал для оповещений"""
if channel is None:
channel = ctx.channel

self.alerts_channel_id = channel.id
await ctx.send(f"✅ Канал {channel.mention} установлен для оповещений")

@tasks.loop(minutes=15)
async def update_stats_display(self):
"""Обновляет отображение статистики в Discord"""
if not self.alerts_channel_id:
return

channel = self.bot.get_channel(self.alerts_channel_id)
if not channel:
logger.error(f"Alerts channel with ID {self.alerts_channel_id} not found")
return

embed = discord.Embed(
title="Статистика WhatsApp интеграции",
color=discord.Color.blue(),
timestamp=datetime.now()
)

embed.add_field(name="Сообщений получено", value=str(self.whatsapp_stats['messages_received']), inline=True)
embed.add_field(name="Сообщений отправлено", value=str(self.whatsapp_stats['messages_sent']), inline=True)
embed.add_field(name="Ошибок", value=str(self.whatsapp_stats['errors']), inline=True)
embed.add_field(name="Активных пользователей", value=str(len(self.whatsapp_stats['active_users'])), inline=True)

# Проверка на ошибки
if self.whatsapp_stats['errors'] > 10:
embed.add_field(name="⚠️ Внимание", val

**Читайте также**
- [Функции с параметрами в Python: секретное оружие разработчика](/python/sozdanie-i-ispolzovanie-funkcij-s-parametrami-v-python/)
- [Примеры Python-кода: от основ до продвинутых техник с разбором](/python/primery-koda-na-python-i-ih-obuyasnenie/)
- [Выбор языка программирования: где Python действительно выигрывает](/python/python-protiv-drugih-yazykov-programmirovaniya/)
- [Интеграция Python и R-Studio: мощный тандем в анализе данных](/python/instrukciya-po-rabote-v-r-studio-s-python/)
- [Почему в Python нет операторов ++ и -- и какие альтернативы использовать](/python/inkrement-i-dekrement-v-python-kak-eto-rabotaet/)
- [Python: компилируемый или интерпретируемый язык, правда скрыта](/python/python-kompiliruemyj-ili-interpretiruemyj-yazyk/)
- [15 полезных Python-скриптов для автоматизации и работы с данными](/python/poleznye-skripty-na-python/)
- [Lambda-функции в Python: мощные однострочные условия для кода](/python/kak-ispolzovat-lambda-vyrazheniya-s-usloviem-v-python/)
- [Настройка Python в Visual Studio: полное руководство для разработчиков](/python/nastrojka-i-ispolzovanie-python-v-visual-studio/)
- [Массивы в Python: особенности, отличия, эффективное применение](/python/massivy-v-python-opredelenie-i-primery/)

Проверь как ты усвоил материалы статьи
Пройди тест и узнай насколько ты лучше других читателей
Какой библиотекой рекомендуется использовать для работы с API WhatsApp в Python?
1 / 5

Загрузка...