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
|
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
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 = utils.irc.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(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).split("\n")[0].strip("\r")
def format(self) -> str:
tags, line = self._format()
if tags:
return "%s %s" % (tags, line)
else:
return line
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:
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:]
return valid, overflow
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")
|
