-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmsg_parser.py
86 lines (75 loc) · 2.7 KB
/
msg_parser.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
76
77
78
79
80
81
82
83
84
85
86
from enum import Enum
from typing import TypedDict
from exceptions import ParserException
class Action(Enum):
MESSAGE = "message"
GROUP_CREATE = "group_create"
GROUP_DELETE = "group_delete"
GROUP_ADD_USER = "group_add_user"
GROUP_REMOVE_USER = "group_remove_user"
GROUP_RENAME = "group_rename"
class ParsedMsg(TypedDict):
action: Action
args: dict
def parse(msg: str) -> ParsedMsg:
split_msg = msg.split(None, 1)
# parse messages of format "@<recipient> message"
if(str(split_msg[0]).startswith("@")):
recipient_uid = split_msg[0][1:]
if(len(split_msg)>1):
content = split_msg[1]
else:
content = ""
return {
"action": Action.MESSAGE,
"args": {'uid': recipient_uid, 'msg': content}
}
# parse group commands. All group commands start with group <name> ...
if(str(split_msg[0]) == "group"):
split_msg = msg.split(" ")
if(len(split_msg) <= 2):
raise ParserException(f"Unkown command {msg}")
if(split_msg[2] == "create"):
return {
"action": Action.GROUP_CREATE,
"args": {"group_uid": split_msg[1]}
}
if(split_msg[2] == "delete"):
return {
"action": Action.GROUP_DELETE,
"args": {"group_uid": split_msg[1]}
}
if(split_msg[2] == "rename"):
if(len(split_msg) != 4):
raise ParserException(f"Unkown command {msg}")
return {
"action": Action.GROUP_RENAME,
"args": {
"group_uid": split_msg[1],
"new_uid": split_msg[3]
}
}
# parse group add or remove user(s) commands of form group <name> add <user> ...
if(split_msg[2] == "add" or split_msg[2] == "remove"):
if(len(split_msg) <=3):
raise ParserException(f"Missing user in command {msg}")
uids = []
for client_uid in split_msg[3:]:
uids.append(client_uid)
if(split_msg[2] == "add"):
return {
"action": Action.GROUP_ADD_USER,
"args": {
"group_uid": split_msg[1],
"client_uids": uids
}
}
else:
return {
"action": Action.GROUP_REMOVE_USER,
"args": {
"group_uid": split_msg[1],
"client_uids": uids
}
}
raise ParserException(f"Unknown command: {msg}")