-
Notifications
You must be signed in to change notification settings - Fork 0
/
websocket_bot.py
75 lines (58 loc) · 1.9 KB
/
websocket_bot.py
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
import json
import ssl
import threading
import time
import websocket
import config
from utils.text_emoji import message_to_code
new_messages = []
ws = websocket.create_connection(f"wss://{config.IP}:5000", sslopt = {'cert_reqs': ssl.CERT_NONE})
class BackgroundBot:
def __init__(self):
self.local_history = ''
def process(self):
print('bot was started')
while 1:
_new_messages = self.check_new_message()
if _new_messages:
_message = _new_messages[-1]
if _message['type'] == 'new_message':
if _message['text'] == 'hahaha':
self.send_response(
{'type': 'drop_emoji', 'emoji': '😀'})
else:
self.send_response(
{'type': 'change_color', 'color': message_to_code(_message['text']), 'save': 1})
elif _message['type'] == 'new_sticker':
self.send_response(
{'type': 'change_color', 'color': message_to_code(_message['sticker']), 'save': 1})
time.sleep(0.1)
def send_response(self, data):
ws.send(json.dumps(data))
# ws
# send_messages.append(data)
def check_new_message(self):
global new_messages
if len(new_messages) != 0:
my_message = new_messages.copy()
new_messages = []
return my_message
def start_loader():
ws.send('{"type": "new_user"}')
while 1:
result = ws.recv()
print('new message', result)
try:
new_messages.append(json.loads(result))
except:
pass
def start_process():
BackgroundBot().process()
if __name__ == '__main__':
try:
threading.Thread(target = start_process).start()
start_loader()
except:
pass
finally:
ws.close()