forked from rvinothrajendran/BotTutorialSample
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
138 lines (111 loc) · 4.57 KB
/
Copy pathapp.py
File metadata and controls
138 lines (111 loc) · 4.57 KB
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
from re import A
import sys
from time import time
import traceback
from datetime import datetime
from aiohttp import web
from aiohttp.web import Request, Response, json_response
from botbuilder.core import (
BotFrameworkAdapterSettings,
TurnContext,
BotFrameworkAdapter,
turn_context,
)
from botbuilder.core.integration import aiohttp_error_middleware
from botbuilder.schema import Activity, ActivityTypes,ConversationReference
from typing import Dict
from bot import MyBot
from config import DefaultConfig
import threading
import datetime
import time
import asyncio
CONFIG = DefaultConfig()
# Create adapter.
# See https://aka.ms/about-bot-adapter to learn more about how bots work.
SETTINGS = BotFrameworkAdapterSettings(CONFIG.APP_ID, CONFIG.APP_PASSWORD)
ADAPTER = BotFrameworkAdapter(SETTINGS)
CONVERSATIONS_REFERENCE : Dict[str,ConversationReference] = dict()
# Catch-all for errors.
async def on_error(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] unhandled error: {error}", file=sys.stderr)
traceback.print_exc()
# Send a message to the user
await context.send_activity("The bot encountered an error or bug.")
await context.send_activity(
"To continue to run this bot, please fix the bot source code."
)
# Send a trace activity if we're talking to the Bot Framework Emulator
if context.activity.channel_id == "emulator":
# Create a trace activity that contains the error object
trace_activity = Activity(
label="TurnError",
name="on_turn_error Trace",
timestamp=datetime.utcnow(),
type=ActivityTypes.trace,
value=f"{error}",
value_type="https://www.botframework.com/schemas/error",
)
# Send a trace activity, which will be displayed in Bot Framework Emulator
await context.send_activity(trace_activity)
ADAPTER.on_turn_error = on_error
# Create the Bot
BOT = MyBot(CONVERSATIONS_REFERENCE)
# Listen for incoming requests on /api/messages
async def messages(req: Request) -> Response:
# Main bot message handler.
if "application/json" in req.headers["Content-Type"]:
body = await req.json()
else:
return Response(status=415)
activity = Activity().deserialize(body)
auth_header = req.headers["Authorization"] if "Authorization" in req.headers else ""
try:
response = await ADAPTER.process_activity(activity, auth_header, BOT.on_turn)
if response:
return json_response(data=response.body, status=response.status)
return Response(status=201)
except Exception as exception:
raise exception
async def send_proactive_message():
while True:
time.sleep(5)
currenttime = time.localtime()
result = time.strftime("%I:%M:%S %p", currenttime)
if CONVERSATIONS_REFERENCE.keys():
for conversation in CONVERSATIONS_REFERENCE.values():
await ADAPTER.continue_conversation(
conversation,
lambda turn_context : turn_context.send_activity(f"It is now {result}."),
CONFIG.APP_ID,
)
def target_callback():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(send_proactive_message())
async def send_proactive_external_message():
if CONVERSATIONS_REFERENCE.keys():
for conversation in CONVERSATIONS_REFERENCE.values():
await ADAPTER.continue_conversation(
conversation,
lambda turn_context : turn_context.send_activity(f"Hey , Am External Request"),
CONFIG.APP_ID,
)
async def notify(req : Request) -> Response:
await send_proactive_external_message()
return Response(status=200,text="Hey I sent message to all the user")
APP = web.Application(middlewares=[aiohttp_error_middleware])
APP.router.add_post("/api/messages", messages)
APP.router.add_get("/api/notify",notify)
if __name__ == "__main__":
#_thread = threading.Thread(target=target_callback)
#_thread.start()
try:
web.run_app(APP, host="localhost", port=CONFIG.PORT)
except Exception as error:
raise error