[PYTHON] Comment créer un BOT Cisco Webex Teams à l'aide de Flask

Récemment, lors du développement d'un système utilisant Cisco Webex Teams, j'ai eu l'opportunité de créer un chatbot pour Webex Teams, et je laisserai une note pour moi-même.

Aussi, pour la démonstration cette fois, ngrok sera utilisé pour publier le serveur.

Le flux cette fois est le suivant. · Obtenez un compte Cisco Webex Developers ・ Acquisition de jeton d'accès ・ Installation de diverses bibliothèques Python ・ Préparation de ngrok ·La mise en oeuvre ・ Contrôle de fonctionnement

L'environnement de développement utilisé cette fois est le suivant. ・ Raspberry Pi3 B + ・ Python3 ・ Ngrok ・ Flacon

Obtenez un compte Cisco Webex Developers

Tout d'abord, accédez à ce site. https://developer.webex.com/

Cliquez sur S'inscrire dans le coin supérieur droit pour procéder à l'enregistrement.

Obtenez un jeton d'accès

https://developer.webex.com/docs/api/getting-started Accédez à la page de démarrage et obtenez un jeton

Pendant la phase de démonstration, vous devrez reprendre ce jeton toutes les 12 heures.

Cliquez sur Mes applications Webex Teams.

Sélectionnez un bot.

Après avoir rempli les différents formulaires, cliquez sur Addbot.

Installation de diverses bibliothèques Python

terminal

$ pip3 install requests

terminal

$ pip3 install flask

Préparer ngrok

https://qiita.com/kaba/items/82de7e53d99ad9c74cc0 J'ai fait référence à ce site.

la mise en oeuvre

Cette fois, à titre d'exemple, je vais essayer de créer un bot qui ne renvoie que bonjour le monde! En parlant à Hi et au bot. Collez le code implémenté.

Tout d'abord, remplacez le jeton que vous avez obtenu précédemment dans YourToken.

token = 'YourToken'

Définissez une fonction pour publier un message.

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

À ce stade, elle est gérée plusieurs fois en fonction du comportement lors de la détection du message, donc une fois cette fonction appelée, elle passe pendant un certain temps. Si quelqu'un connaît ce comportement, apprenez-moi s'il vous plaît.

Ici, c'est la partie qui affiche mainpage.html lors de l'accès à / panel. C'est une bonne idée d'écrire une explication sur la façon de l'utiliser.

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

Collectez l'identifiant du message, l'identifiant de la personne, l'e-mail et l'identifiant de la salle à partir des informations du message reçu

@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)

Ici, nous cherchons à savoir si l'ID utilisateur qui a reçu le message est le bot lui-même. De plus, le post_message défini précédemment est appelé de sorte que lorsque Hi est reçu, bonjour le monde! Est renvoyé.

    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"

De plus, comme l'adresse IP de ngrok change à chaque fois qu'il est démarré, il lit l'adresse IP.

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']

La gestion des erreurs.

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

Le résultat final est le suivant.

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']

#Enregistrement Webhook
header = {"Authorization": "Bearer %s" % token, "content-type": "application/json"}
requests.packages.urllib3.disable_warnings() #Supprimer l'avertissement SSL
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) #enregistrement de webhook

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

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

Contrôle de fonctionnement

terminal

$ ngrok http 5000

terminal

$ python3 main.py