Spaces:
Build error
Build error
File size: 3,507 Bytes
0827183 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import asyncio
import sys
from types import MethodType
from flask import Flask, Response, request
from botbuilder.core import (
BotFrameworkAdapter,
BotFrameworkAdapterSettings,
MessageFactory,
TurnContext,
)
from botbuilder.schema import Activity, InputHints
from .default_config import DefaultConfig
from .my_bot import MyBot
class BotApp:
"""A Flask echo bot."""
def __init__(self):
# Create the loop and Flask app
self.loop = asyncio.get_event_loop()
self.flask = Flask(__name__, instance_relative_config=True)
self.flask.config.from_object(DefaultConfig)
# Create adapter.
# See https://aka.ms/about-bot-adapter to learn more about how bots work.
self.settings = BotFrameworkAdapterSettings(
self.flask.config["APP_ID"], self.flask.config["APP_PASSWORD"]
)
self.adapter = BotFrameworkAdapter(self.settings)
# Catch-all for errors.
async def on_error(adapter, context: TurnContext, error: Exception):
# This check writes out errors to console log .vs. app insights.
# NOTE: In production environment, you should consider logging this to Azure
# application insights.
print(f"\n [on_turn_error]: {error}", file=sys.stderr)
# Send a message to the user
error_message_text = "Sorry, it looks like something went wrong."
error_message = MessageFactory.text(
error_message_text, error_message_text, InputHints.expecting_input
)
await context.send_activity(error_message)
# pylint: disable=protected-access
if adapter._conversation_state:
# If state was defined, clear it.
await adapter._conversation_state.delete(context)
self.adapter.on_turn_error = MethodType(on_error, self.adapter)
# Create the main dialog
self.bot = MyBot()
def messages(self) -> Response:
"""Main bot message handler that listens for incoming requests."""
if "application/json" in request.headers["Content-Type"]:
body = request.json
else:
return Response(status=415)
activity = Activity().deserialize(body)
auth_header = (
request.headers["Authorization"]
if "Authorization" in request.headers
else ""
)
async def aux_func(turn_context):
await self.bot.on_turn(turn_context)
try:
task = self.loop.create_task(
self.adapter.process_activity(activity, auth_header, aux_func)
)
self.loop.run_until_complete(task)
return Response(status=201)
except Exception as exception:
raise exception
@staticmethod
def test() -> Response:
"""
For test only - verify if the flask app works locally - e.g. with:
```bash
curl http://127.0.0.1:3978/api/test
```
You shall get:
```
test
```
"""
return Response(status=200, response="test\n")
def run(self, host=None) -> None:
try:
self.flask.run(
host=host, debug=False, port=self.flask.config["PORT"]
) # nosec debug
except Exception as exception:
raise exception
|