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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
|
import datetime, typing, uuid
from src import IRCObject, utils
# this should be 510 (RFC1459, 512 with \r\n) but a server BitBot uses is broken
LINE_MAX = 470
class IRCArgs(object):
def __init__(self, args: typing.List[str]):
self._args = args
def get(self, index: int) -> typing.Optional[str]:
if index < 0:
if len(self._args) > (abs(index)-1):
return self._args[index]
elif len(self._args) > index:
return self._args[index]
return None
def __repr__(self):
return "IRCArgs(%s)" % self._args
def __len__(self) -> int:
return len(self._args)
def __getitem__(self, index: int) -> str:
return self._args[index]
def __setitem__(self, index: int, value: str):
self._args[index] = value
class Hostmask(object):
def __init__(self, nickname: str, username: str, hostname: str,
hostmask: str):
self.nickname = nickname
self.username = username
self.hostname = hostname
self.hostmask = hostmask
def __repr__(self):
return "Hostmask(%s)" % self.__str__()
def __str__(self):
return self.hostmask
def parse_hostmask(hostmask: str) -> Hostmask:
nickname, _, username = hostmask.partition("!")
username, _, hostname = username.partition("@")
return Hostmask(nickname, username, hostname, hostmask)
MESSAGE_TAG_ESCAPED = [r"\:", r"\s", r"\\", r"\r", r"\n"]
MESSAGE_TAG_UNESCAPED = [";", " ", "\\", "\r", "\n"]
def message_tag_escape(s):
return utils.irc.multi_replace(s, MESSAGE_TAG_UNESCAPED,
MESSAGE_TAG_ESCAPED)
def message_tag_unescape(s):
unescaped = utils.irc.multi_replace(s, MESSAGE_TAG_ESCAPED,
MESSAGE_TAG_UNESCAPED)
return unescaped.replace("\\", "")
class ParsedLine(object):
def __init__(self, command: str, args: typing.List[str],
source: Hostmask=None,
tags: typing.Dict[str, str]=None):
self.id = str(uuid.uuid4())
self.command = command
self._args = args
self.args = IRCArgs(args)
self.source = source
self.tags = tags or {} # type: typing.Dict[str, str]
self._valid = True
self._assured = False
def __repr__(self):
return "ParsedLine(%s)" % self.__str__()
def __str__(self):
return self.format()
def valid(self) -> bool:
return self._valid
def invalidate(self):
self._valid = False
def assured(self) -> bool:
return self._assured
def assure(self):
self._assured = True
def add_tag(self, tag: str, value: str=None):
self.tags[tag] = value or ""
def has_tag(self, tag: str) -> bool:
return "tag" in self.tags
def get_tag(self, tag: str) -> typing.Optional[str]:
return self.tags[tag]
def _tag_str(self, tags: typing.Dict[str, str]) -> str:
tag_pieces = []
for tag, value in tags.items():
if value:
value_escaped = message_tag_escape(value)
tag_pieces.append("%s=%s" % (tag, value_escaped))
else:
tag_pieces.append(tag)
if tag_pieces:
return "@%s" % ";".join(tag_pieces)
return ""
def _format(self) -> typing.Tuple[str, str]:
pieces = []
tags = ""
if self.tags:
tags = self._tag_str(self.tags)
if self.source:
pieces.append(":%s" % str(self.source))
pieces.append(self.command.upper())
if self.args:
for i, arg in enumerate(self._args):
if arg and i == len(self._args)-1 and (
" " in arg or arg[0] == ":"):
pieces.append(":%s" % arg)
else:
pieces.append(arg)
return tags, " ".join(pieces).replace("\r", "")
def format(self) -> str:
tags, line = self._format()
line, _ = self._newline_truncate(line)
if tags:
return "%s %s" % (tags, line)
else:
return line
def _newline_truncate(self, line: str) -> typing.Tuple[str, str]:
line, sep, overflow = line.partition("\n")
return [line, overflow]
def _line_max(self, hostmask: str, margin: int) -> int:
return LINE_MAX-len((":%s " % hostmask).encode("utf8"))-margin
def truncate(self, hostmask: str, margin: int=0) -> typing.Tuple[str, str]:
valid_bytes = b""
valid_index = -1
line_max = self._line_max(hostmask, margin)
tags_formatted, line_formatted = self._format()
for i, char in enumerate(line_formatted):
encoded_char = char.encode("utf8")
if (len(valid_bytes)+len(encoded_char) > line_max
or encoded_char == b"\n"):
break
else:
valid_bytes += encoded_char
valid_index = i
valid_index += 1
valid = line_formatted[:valid_index]
if tags_formatted:
valid = "%s %s" % (tags_formatted, valid)
overflow = line_formatted[valid_index:]
if overflow and overflow[0] == "\n":
overflow = overflow[1:]
return valid, overflow
def parse_line(line: str) -> ParsedLine:
tags = {} # type: typing.Dict[str, typing.Any]
source = None # type: typing.Optional[Hostmask]
command = None
if line[0] == "@":
tags_prefix, line = line[1:].split(" ", 1)
for tag in filter(None, tags_prefix.split(";")):
tag, sep, value = tag.partition("=")
if value:
tags[tag] = message_tag_unescape(value)
else:
tags[tag] = None
line, trailing_separator, trailing_split = line.partition(" :")
trailing = None # type: typing.Optional[str]
if trailing_separator:
trailing = trailing_split
if line[0] == ":":
source_str, line = line[1:].split(" ", 1)
source = parse_hostmask(source_str)
command, sep, line = line.partition(" ")
args = [] # type: typing.List[str]
if line:
# this is so that `args` is empty if `line` is empty
args = line.split(" ")
if not trailing == None:
args.append(typing.cast(str, trailing))
return ParsedLine(command, args, source, tags)
class SentLine(IRCObject.Object):
def __init__(self, events: "EventManager.Events",
send_time: datetime.datetime, hostmask: str, line: ParsedLine):
self.events = events
self.send_time = send_time
self._hostmask = hostmask
self.parsed_line = line
def __repr__(self) -> str:
return "IRCLine.SentLine(%s)" % self.__str__()
def __str__(self) -> str:
return self._for_wire()
def _for_wire(self) -> str:
return self.parsed_line.truncate(self._hostmask)[0]
def for_wire(self) -> bytes:
return b"%s\r\n" % self._for_wire().encode("utf8")
class IRCBatch(object):
def __init__(self, identifier: str, batch_type: str, args: typing.List[str],
tags: typing.Dict[str, str]=None, source: Hostmask=None):
self.identifier = identifier
self.type = batch_type
self.args = args
self.tags = tags or {}
self.source = source
self._lines = [] # type: typing.List[ParsedLine]
def add_line(self, line: ParsedLine):
self._lines.append(line)
def get_lines(self) -> typing.List[ParsedLine]:
return self._lines
class IRCSendBatch(IRCBatch):
def __init__(self, batch_type: str, args: typing.List[str],
tags: typing.Dict[str, str]=None):
IRCBatch.__init__(self, str(uuid.uuid4()), batch_type, args, tags)
def get_lines(self) -> typing.List[ParsedLine]:
lines = []
for line in self._lines:
line.add_tag("batch", self.identifier)
lines.append(line)
lines.insert(0, ParsedLine("BATCH",
["+%s" % self.identifier, self.type]))
lines.append(ParsedLine("BATCH", ["-%s" % self.identifier]))
return lines
|
