بناء واتس اب شات بوت باستخدام Python

من السهل جدًا إنشاء chatbot باستخدام واجهة Ultramsg API ، يمكنك إنشاء روبوت محادثة لخدمة العملاء وبرمجة روبوت ذكي للإستعلام من قاعدة البيانات واستقبال البيانات من العملاء ومعالجتها من خلال خطوات بسيطة باستخدام لغة Python.

مهام الـ WhatsApp bot

إخراج قائمة الأوامر.
ناتج وقت الخادم للروبوت الذي يعمل على.
إرسال الصورة إلى رقم الهاتف أو المجموعة.
إرسال ملف صوتي.
إرسال تسجيل صوتي ppt.
إرسال ملف فيديو.
إرسال جهة اتصال.

الخطوة 1: تثبيت flask

نحتاج إلى نشر الخادم باستخدام إطار عمل FLASK.
يسمح FLASK بالاستجابة بشكل ملائم للطلبات الواردة ومعالجتها.

pip install flask

الخطوة الثانية: تثبيت ngrok

لأغراض التجربة بالسيرفر المحلي سنحتاج لخدمة tunneling و في هذا الدرس سنستخدم ngrok يمكنك تحميل ngrok من هنا .

الخطوة 3: إنشاء تطبيق flask جديد

سننشئ الملف: app.py ونكتب الكود التالي بداخله

from flask import Flask, request, jsonify
from ultrabot import ultraChatBot
import json

app = Flask(__name__)

@app.route('/', methods=['POST'])
def home():
    if request.method == 'POST':
        bot = ultraChatBot(request.json)
        return bot.Processingـincomingـmessages()

if(__name__) == '__main__':
    app.run()

الخطوة 4: معالجة الرسائل الواردة

سننشئ الملف: ultrabot.py ونكتب الكود التالي بداخله

import json
import requests
import datetime

class ultraChatBot():    
    def __init__(self, json):
        self.json = json
        self.dict_messages = json['data']
        self.ultraAPIUrl = 'https://api.ultramsg.com/{{instance_id}}/'
        self.token = '{{token}}'

   
    def send_requests(self, type, data):
        url = f"{self.ultraAPIUrl}{type}?token={self.token}"
        headers = {'Content-type': 'application/json'}
        answer = requests.post(url, data=json.dumps(data), headers=headers)
        return answer.json()

    def send_message(self, chatID, text):
        data = {"to" : chatID,
                "body" : text}  
        answer = self.send_requests('messages/chat', data)
        return answer

    def send_image(self, chatID):
        data = {"to" : chatID,
                "image" : "https://file-example.s3-accelerate.amazonaws.com/images/test.jpeg"}  
        answer = self.send_requests('messages/image', data)
        return answer

    def send_video(self, chatID):
        data = {"to" : chatID,
                "video" : "https://file-example.s3-accelerate.amazonaws.com/video/test.mp4"}  
        answer = self.send_requests('messages/video', data)
        return answer

    def send_audio(self, chatID):
        data = {"to" : chatID,
                "audio" : "https://file-example.s3-accelerate.amazonaws.com/audio/2.mp3"}  
        answer = self.send_requests('messages/audio', data)
        return answer


    def send_voice(self, chatID):
        data = {"to" : chatID,
                "audio" : "https://file-example.s3-accelerate.amazonaws.com/voice/oog_example.ogg"}  
        answer = self.send_requests('messages/voice', data)
        return answer

    def send_contact(self, chatID):
        data = {"to" : chatID,
                "contact" : "[email protected]"}  
        answer = self.send_requests('messages/contact', data)
        return answer


    def time(self, chatID):
        t = datetime.datetime.now()
        time = t.strftime('%Y-%m-%d %H:%M:%S')
        return self.send_message(chatID, time)


    def welcome(self,chatID, noWelcome = False):
        welcome_string = ''
        if (noWelcome == False):
            welcome_string = "Hi , welcome to WhatsApp chatbot using Python\n"
        else:
            welcome_string = """wrong command
Please type one of these commands:
*hi* : Saluting
*time* : show server time
*image* : I will send you a picture
*video* : I will send you a Video
*audio* : I will send you a audio file
*voice* : I will send you a ppt audio recording
*contact* : I will send you a contact
"""
        return self.send_message(chatID, welcome_string)


    def Processingـincomingـmessages(self):
        if self.dict_messages != []:
            message =self.dict_messages
            text = message['body'].split()
            if not message['fromMe']:
                chatID  = message['from'] 
                if text[0].lower() == 'hi':
                    return self.welcome(chatID)
                elif text[0].lower() == 'time':
                    return self.time(chatID)
                elif text[0].lower() == 'image':
                    return self.send_image(chatID)
                elif text[0].lower() == 'video':
                    return self.send_video(chatID)
                elif text[0].lower() == 'audio':
                    return self.send_audio(chatID)
                elif text[0].lower() == 'voice':
                    return self.send_voice(chatID)
                elif text[0].lower() == 'contact':
                    return self.send_contact(chatID)
                else:
                    return self.welcome(chatID, True)
            else: return 'NoCommand'

الخطوة 5: ابدأ مشروع WhatsApp Chatbot

تشغيل خادم FLASK

flask run

تشغيل ngrok

تشغيل ngrok لنظام التشغيل Windows:

ngrok http 5000

تشغيل ngrok لنظام التشغيل Mac:

./ngrok http 5000

الخطوة 6: تعيين URL Webhook في إعدادات الجهاز

بعد تشغيل ngrok ، يجب أن ترى على سبيل المثال:

https://7647-115-83-121-164.ngrok.io

الصق عنوان URL الخاص بك في إعدادات الجهاز ، كما في الصورة التالية:

استقبال رسائل WhatsApp باستخدام Python و webhook-ultramsg

وظائف Chatbot المستخدمة في الكود

send_message

تستخدم لإرسال رسائل نصية WhatsApp

def send_message(self, chatID, text):
    data = {"to" : chatID,
        "body" : text}  
    answer = self.send_requests('messages/chat', data)
    return answer

ChatID – معرف الدردشة حيث يجب إرسال الرسالة له ، على سبيل المثال [email protected] .
نص – نص الرسالة.

time

يرسل وقت الخادم الحالي.

def time(self, chatID):
    t = datetime.datetime.now()
    time = t.strftime('%Y-%m-%d %H:%M:%S')
    return self.send_message(chatID, time)

ChatID – معرف الدردشة حيث يجب إرسال الرسالة له ، على سبيل المثال [email protected] .

send_image

أرسل صورة إلى رقم الهاتف أو المجموعة

def send_image(self, chatID):
    data = {"to" : chatID,
        "image" : "https://file-example.s3-accelerate.amazonaws.com/images/test.jpeg"}  
    answer = self.send_requests('messages/image', data)
    return answer

send_video

أرسل مقطع فيديو إلى رقم الهاتف أو المجموعة

def send_video(self, chatID):
    data = {"to" : chatID,
        "video" : "https://file-example.s3-accelerate.amazonaws.com/video/test.mp4"}  
    answer = self.send_requests('messages/video', data)
    return answer

send_audio

أرسل ملفًا صوتيًا إلى رقم الهاتف أو المجموعة

def send_audio(self, chatID):
    data = {"to" : chatID,
        "audio" : "https://file-example.s3-accelerate.amazonaws.com/audio/2.mp3"}  
    answer = self.send_requests('messages/audio', data)
    return answer

send_voice

أرسل تسجيل صوتي ppt إلى رقم الهاتف أو المجموعة

def send_voice(self, chatID):
    data = {"to" : chatID,
        "audio" : "https://file-example.s3-accelerate.amazonaws.com/voice/oog_example.ogg"}  
    answer = self.send_requests('messages/voice', data)
    return answer

send_contact

إرسال جهة اتصال واحدة أو قائمة جهات اتصال إلى رقم الهاتف أو المجموعة

def send_contact(self, chatID):
    data = {"to" : chatID,
        "contact" : "[email protected]"}  
    answer = self.send_requests('messages/contact', data)
    return answer

Processingـincomingـmessages

def Processingـincomingـmessages(self):
    if self.dict_messages != []:
        message =self.dict_messages
        text = message['body'].split()
        if not message['fromMe']:
        chatID  = message['from'] 
        if text[0].lower() == 'hi':
            return self.welcome(chatID)
        elif text[0].lower() == 'time':
            return self.time(chatID)
        elif text[0].lower() == 'image':
            return self.send_image(chatID)
        elif text[0].lower() == 'video':
            return self.send_video(chatID)
        elif text[0].lower() == 'audio':
            return self.send_audio(chatID)
        elif text[0].lower() == 'voice':
            return self.send_voice(chatID)
        elif text[0].lower() == 'contact':
            return self.send_contact(chatID)
        else:
            return self.welcome(chatID, True)
        else: return 'NoCommand'

يمكنك مشاهدة الخطوات السابقة في هذا الفيديو:

روابط مفيدة

كل الكود متاح على GitHub .
إرسال رسائل WhatsApp API باستخدام Python.
كيفية ضبط WebHooks باستخدام منصة Ultramsg
كيفية تلقي رسائل WhatsApp باستخدام Python و webhook.