-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserverBot.py
More file actions
executable file
·99 lines (79 loc) · 3.59 KB
/
serverBot.py
File metadata and controls
executable file
·99 lines (79 loc) · 3.59 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
# !/usr/bin/env python
import configparser, dingtalk_stream, logging, time, threading, datetime, os
from dingtalk_stream import AckMessage
from dingtalk_webhook.dingtalk_webhook import initWebhook, send_message
from serverOperator.serverOperator import initOperator, getInfoMessage, startUpServer, shutDownServer, checkServer
from utils.log import preiodic_check_log
def setup_logger():
logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter('%(asctime)s %(name)-8s %(levelname)-8s %(message)s [%(filename)s:%(lineno)d]'))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return logger
def define_options():
os.chdir(os.path.dirname(os.path.abspath(__file__)))
options = {}
config = configparser.ConfigParser()
config.read('conf/conf.ini')
for section in config.sections():
for key in config[section]:
if key in ['temperature_limit', 'power_limit']:
options[key] = int(config[section][key])
else:
options[key] = config[section][key]
return options
def periodicCheck():
global lastCheckTime, lastCheckMessage
while True:
status, message = checkServer()
if status == True:
send_message(message)
preiodic_check_log("./logs/periodic_check.log", 'status: {status}\nmessage: {message}'.format(status='ERROR' if status else 'OK', message=message))
lastCheckTime = datetime.datetime.now()
lastCheckMessage = message
time.sleep(300)
class ServerBotHandler(dingtalk_stream.ChatbotHandler):
global lastCheckTime, lastCheckMessage
def __init__(self, logger: logging.Logger = None):
super(dingtalk_stream.ChatbotHandler, self).__init__()
if logger:
self.logger = logger
async def process(self, callback: dingtalk_stream.CallbackMessage):
incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
incoming_message_str = incoming_message.text.content.strip()
if "help" in incoming_message_str:
response = """
Commands:
help: show this help message
status: check serverBot
info: show server's info
startup: start up server
shutdown: shut down server
""".strip()
elif "status" in incoming_message_str:
response = "ServerBot is running.\nLast check time: {lastCheckTime}\nLast check message: {lastCheckMessage}".format(lastCheckTime=lastCheckTime, lastCheckMessage=lastCheckMessage)
elif "info" in incoming_message_str:
response = getInfoMessage()
elif "startup" in incoming_message_str:
response = startUpServer()
elif "shutdown" in incoming_message_str:
response = shutDownServer()
else:
response = "Give me \"help\" to get commands."
self.reply_text(response, incoming_message)
return AckMessage.STATUS_OK, 'OK'
def main():
logger = setup_logger()
options = define_options()
credential = dingtalk_stream.Credential(options['client_id'], options['client_secret'])
initWebhook(options['webhook'], options['client_secret'], options['at_user_phone'])
initOperator(options['host'], options['user'], options['password'], options['temperature_limit'], options['power_limit'])
thread = threading.Thread(target=periodicCheck)
thread.start()
client = dingtalk_stream.DingTalkStreamClient(credential)
client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC, ServerBotHandler(logger))
client.start_forever()
if __name__ == '__main__':
main()