[PYTHON] So erstellen Sie mit Flask einen BOT für Cisco Webex-Teams

Als ich kürzlich ein System mit Cisco Webex Teams entwickelte, hatte ich die Möglichkeit, einen Chatbot für Webex Teams zu erstellen, und ich werde mir selbst eine Notiz hinterlassen.

Zur Demonstration dieses Mal wird ngrok auch zum Veröffentlichen des Servers verwendet.

Der Fluss ist diesmal wie folgt. · Erstellen Sie ein Cisco Webex Developers-Konto ・ Erwerb des Zugriffstokens ・ Installation verschiedener Python-Bibliotheken ・ Vorbereitung von ngrok ·Implementierung ・ Funktionsprüfung

Die diesmal verwendete Entwicklungsumgebung ist wie folgt. ・ Himbeer Pi3 B + ・ Python3 ・ Ngrok · Flasche

Holen Sie sich ein Konto bei Cisco Webex Developers

Greifen Sie zunächst auf diese Site zu. https://developer.webex.com/

Klicken Sie oben rechts auf Anmelden, um mit der Registrierung fortzufahren.

Zugriffstoken erhalten

https://developer.webex.com/docs/api/getting-started Gehen Sie zur Einstiegsseite und holen Sie sich einen Token

Während der Demo-Phase müssen Sie diesen Token alle 12 Stunden wiederholen.

Klicken Sie auf My Webex Teams Apps.

Wählen Sie einen Bot.

Klicken Sie nach dem Ausfüllen der verschiedenen Formulare auf Addbot.

Installation verschiedener Python-Bibliotheken

terminal

$ pip3 install requests

terminal

$ pip3 install flask

Ngrok vorbereiten

https://qiita.com/kaba/items/82de7e53d99ad9c74cc0 Ich habe auf diese Seite verwiesen.

Implementierung

Dieses Mal werde ich als Beispiel versuchen, einen Bot zu erstellen, der nur Hallo Welt zurückgibt! Wenn ich mit Hi und dem Bot spreche. Fügen Sie den implementierten Code ein.

Ersetzen Sie zunächst das zuvor erhaltene Token durch YourToken.

token = 'YourToken'

Definieren Sie eine Funktion zum Posten einer Nachricht.

def post_message(room_id,txt,token):
    global ms_flag
    if ms_flag == True:
        ms_flag = False
        m = MultipartEncoder({'roomId': room_id,'text': txt})
        r = requests.post('https://api.ciscospark.com/v1/messages', data=m,headers={'Authorization': 'Bearer %s' % token,'Content-Type': m.content_type})
    else:
        time.sleep(5)
        ms_flag = True

Zu diesem Zeitpunkt wird es je nach Verhalten beim Erkennen der Nachricht mehrmals verarbeitet. Sobald diese Funktion aufgerufen wird, vergeht sie für einen bestimmten Zeitraum. Wenn jemand von diesem Verhalten weiß, lehre mich bitte.

Hier ist es der Teil, der beim Zugriff auf / panel die Datei mainpage.html anzeigt. Es ist eine gute Idee, eine Erklärung zu schreiben, wie man es benutzt.

@app.route("/panel",methods=['GET'])    
def main_page():
    return render_template("mainpage.html")

Sammeln Sie Nachrichten-ID, Personen-ID, E-Mail und Raum-ID aus den empfangenen Nachrichteninformationen

@app.route("/",methods=['POST'])   
def handle_message():
    json = request.json
    
    message_id = json["data"]["id"]
    user_id = json["data"]["personId"]
    email = json["data"]["personEmail"]
    room_id = json["data"]["roomId"]
    bot_id = "yourbotid"

    print(message_id, file = sys.stdout)
    print(user_id, file=sys.stdout)
    print(email, file=sys.stdout)
    print(room_id, file=sys.stdout)

Hier prüfen wir, ob die Benutzer-ID, die die Nachricht erhalten hat, der Bot selbst ist. Außerdem wird die zuvor definierte post_message aufgerufen, sodass bei Empfang von Hi hallo world! Zurückgegeben wird.

    if user_id != bot_id:
        global token
        header = {"Authorization": "Bearer %s" % token}
        get_rooms_url = "https://api.ciscospark.com/v1/messages/" + message_id
        api_response = requests.get(get_rooms_url, headers=header, verify=False)
        response_json = api_response.json()
        message = response_json["text"]
        print(message, file= sys.stdout)
        if message == "Hi" or message == "bot Hi":
            post_message(room_id,"hello world!",token)
       
        return "Success"
    else:
        return "Pass"

Da sich die IP-Adresse von ngrok bei jedem Start ändert, wird auch die IP-Adresse gelesen.

term_output_json = os.popen('curl http://127.0.0.1:4040/api/tunnels').read() 
tunnel_info = json.loads(term_output_json)
public_url = tunnel_info['tunnels'][0]['public_url']

Fehlerbehandlung.

if api_response.status_code != 200:
    print('Webhook registration Error !')
    exit(0)

Das Endergebnis ist wie folgt.

main.py

from __future__ import print_function 
import requests
import sys
import json
import os
import time
from flask import *
from requests_toolbelt.multipart.encoder import MultipartEncoder
import functools

token = 'YourToken'
ms_flag = True

app = Flask(__name__)

def post_message(room_id,txt,token):
    global ms_flag
    if ms_flag == True:
        ms_flag = False
        m = MultipartEncoder({'roomId': room_id,'text': txt})
        r = requests.post('https://api.ciscospark.com/v1/messages', data=m,headers={'Authorization': 'Bearer %s' % token,'Content-Type': m.content_type})
    else:
        time.sleep(5)
        ms_flag = True

@app.route("/panel",methods=['GET'])    
def main_page():
    return render_template("mainpage.html")


@app.route("/",methods=['POST'])   
def handle_message():
    json = request.json
    
    message_id = json["data"]["id"]
    user_id    = json["data"]["personId"]
    email      = json["data"]["personEmail"]
    room_id    = json["data"]["roomId"]
    bot_id     = "yourbotid"

    print(message_id, file = sys.stdout)
    print(user_id, file=sys.stdout)
    print(email, file=sys.stdout)
    print(room_id, file=sys.stdout)


    if user_id != bot_id:
        global token
        header = {"Authorization": "Bearer %s" % token}
        get_rooms_url = "https://api.ciscospark.com/v1/messages/" + message_id
        api_response = requests.get(get_rooms_url, headers=header, verify=False)
        response_json = api_response.json()
        message = response_json["text"]
        print(message, file= sys.stdout)
        if message == "Hi" or message == "bot Hi":
            post_message(room_id,"hello world!",token)
       
        return "Success"
    else:
        return "Pass"


term_output_json = os.popen('curl http://127.0.0.1:4040/api/tunnels').read() 
tunnel_info = json.loads(term_output_json)
public_url = tunnel_info['tunnels'][0]['public_url']

#Webhook-Aufzeichnung
header = {"Authorization": "Bearer %s" % token, "content-type": "application/json"}
requests.packages.urllib3.disable_warnings() #Entfernen Sie die SSL-Warnung
post_message_url = "https://api.ciscospark.com/v1/webhooks"

payload = {
    "resource": "messages",
    "event": "all",
    "targetUrl": public_url,
    "name": "BotDemoWebHook"
}

api_response = requests.post(post_message_url, json=payload, headers=header, verify=False) #Webhook-Aufzeichnung

if api_response.status_code != 200:
    print('Webhook registration Error !')
    exit(0)

if __name__ == '__main__':
    app.run(host='localhost', use_reloader=True, debug=True)

Funktionsprüfung

terminal

$ ngrok http 5000

terminal

$ python3 main.py