Интеграция API WhatsApp и Discord с Python: мощная автоматизация
Для кого эта статья:
- Разработчики и программисты, интересующиеся интеграцией 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
Загрузка переменных окружения в коде:
# 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:
# 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 рекомендуется создать отдельный класс-клиент, который будет управлять соединением:
# 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:
# 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. 💬
Базовая функция для отправки текстовых сообщений:
# 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
Для отправки медиа-файлов (изображений, документов, аудио) требуется другой формат запроса:
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
Интерактивные шаблоны сообщений — мощный инструмент для создания структурированных сообщений с кнопками:
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-обработчик:
# 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
Автоматизация часто повторяющихся задач с помощью планировщика:
# 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()
Для обработки большого количества сообщений рекомендуется реализовать очередь:
# 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 с обработкой команд:
# 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)
Создание расширенных команд с аргументами и проверками:
# 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:
# 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:
# 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: Создание единого центра поддержки клиентов
# 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: Система мониторинга и оповещения
# 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/)