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
|
import datetime, typing
from src import IRCObject
# this should be 510 (RFC1459, 512 with \r\n) but a server BitBot uses is broken
LINE_CUTOFF = 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.command = command
self._args = args
self.args = IRCArgs(args)
self.source = source
self.tags = {} if tags == None else tags
def __repr__(self):
return "ParsedLine(%s)" % self.__str__()
def __str__(self):
return self.format()
def _tag_str(self, tags: typing.Dict[str, str]) -> str:
tag_pieces = []
for tag, value in tags.items():
if value:
tag_pieces.append("%s=%s" % (tag, value))
else:
tag_pieces.append(tag)
if tag_pieces:
return "@%s" % ",".join(tag_pieces)
return ""
def format(self) -> str:
pieces = []
if self.tags:
pieces.append(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 " ".join(pieces).split("\n")[0].strip("\r")
class SentLine(IRCObject.Object):
def __init__(self, events: "EventManager.EventHook",
send_time: datetime.datetime, hostmask: str, line: ParsedLine):
self.events = events
self.send_time = send_time
self._hostmask = hostmask
self.parsed_line = line
self.truncate_marker: typing.Optional[str] = None
def __repr__(self) -> str:
return "IRCLine.SentLine(%s)" % self.__str__()
def __str__(self) -> str:
return self.decoded_data()
def _char_limit(self) -> int:
return LINE_CUTOFF-len(":%s " % self._hostmask)
def _encode_truncate(self) -> typing.Tuple[bytes, str]:
line = self.parsed_line.format()
byte_max = self._char_limit()
encoded = b""
truncated = ""
truncate_marker = b""
if not self.truncate_marker == None:
truncate_marker = typing.cast(str, self.truncate_marker
).encode("utf8")
for i, character in enumerate(line):
encoded_character = character.encode("utf8")
new_len = len(encoded + encoded_character)
if truncate_marker and (byte_max-new_len) < len(truncate_marker):
encoded += truncate_marker
truncated = line[i:]
break
elif new_len > byte_max:
truncated = line[i:]
break
else:
encoded += encoded_character
return (encoded, truncated)
def _for_wire(self) -> bytes:
return self._encode_truncate()[0]
def for_wire(self) -> bytes:
return b"%s\r\n" % self._for_wire()
def decoded_data(self) -> str:
return self._for_wire().decode("utf8")
def truncated(self) -> str:
return self._encode_truncate()[1]
|
