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
|
import datetime
from functools import partial
from typing import NamedTuple
from quarg.quassel.types import MessageType
class User(NamedTuple):
nick: str
host: str
prefix: str
@classmethod
def from_sender(cls, sender, prefixes=''):
nick, *host = sender.split('!', 1)
prefix = prefixes[0] if prefixes else ''
return cls(nick, host[0] if host else '', prefix)
def __repr__(self):
return f'{self.prefix}{self.nick}'
class Message(NamedTuple):
type: MessageType
time: datetime.datetime
buffer: str
user: User
message: str
@classmethod
def from_backlog(cls, backlog):
return cls(MessageType(backlog.type), backlog.time, backlog.buffer.buffername,
User.from_sender(backlog.sender.sender, backlog.senderprefixes), backlog.message)
def format_from(backlog_row):
message = Message.from_backlog(backlog_row)
formatter = FORMATTERS[message.type]
timestamp = message.time.isoformat(sep=' ', timespec='seconds')
return f'{timestamp}\t{message.buffer}\t{formatter(message)}'
def format_part(msg):
if msg.message:
return f'<-- {msg.user} has left {msg.buffer} ({msg.message})'
return f'<-- {msg.user} has left {msg.buffer}'
def format_quit(msg):
if msg.message:
return f'<-- {msg.user} has quit ({msg.message})'
return f'<-- {msg.user} has quit'
def format_kick(msg):
target, *kickmsg = msg.message.split(' ', 1)
if kickmsg:
return f'<-* {msg.user} has kicked {target} from {msg.buffer} ({kickmsg[0]})'
return f'<-* {msg.user} has kicked {target} from {msg.buffer}'
def format_kill(msg):
# pylint: disable=line-too-long
# As of 2021-04-24 not even Quassel implements printing this message [1].
# They do have a symbol [2] for it, however, so use that along with the message
# [1] https://github.com/quassel/quassel/blob/285215315e6f2420724532323a4b1bccae156cb1/src/uisupport/uistyle.cpp#L950
# [2] https://github.com/quassel/quassel/blob/285215315e6f2420724532323a4b1bccae156cb1/src/uisupport/uistyle.cpp#L1079-L1080
return f'<-x {msg.message}'
def format_generic(msg):
return f'* {msg.message}'
def parse_netsplit(splitmsg):
# splitmsg contains user!host separated by #:# ...
elements = splitmsg.split('#:#')
# ... however, the last element contains the split servers instead
servers = elements.pop().split(' ', 1)
# TODO: This takes ages if the netsplit was large
users = [User.from_sender(e) for e in elements]
return users, servers
def format_netsplit_join(msg):
users, (srv_left, srv_right) = parse_netsplit(msg.message)
have_joined = ', '.join(user.nick for user in users)
return f'=> Netsplit between {srv_left} and {srv_right} ended. Users joined: {have_joined}'
def format_netsplit_quit(msg):
users, (srv_left, srv_right) = parse_netsplit(msg.message)
have_quit = ', '.join(user.nick for user in users)
return f'<= Netsplit between {srv_left} and {srv_right}. Users quit: {have_quit}'
def fmt(string):
return partial(lambda string, msg: string.format(**msg._asdict()), string)
FORMATTERS = {
MessageType.PRIVMSG: fmt('<{user}> {message}'),
MessageType.NOTICE: fmt('[{user}] {message}'),
MessageType.ACTION: fmt('-*- {user} {message}'),
MessageType.NICK: fmt('<-> {user} is now known as {message}'),
MessageType.MODE: fmt('*** Mode {message} by {user}'),
MessageType.JOIN: fmt('--> {user} ({user.host}) has joined {buffer}'),
MessageType.PART: format_part,
MessageType.QUIT: format_quit,
MessageType.KICK: format_kick,
MessageType.KILL: format_kill,
MessageType.SERVER: format_generic,
MessageType.INFO: format_generic,
MessageType.ERROR: format_generic,
MessageType.DAYCHANGE: format_generic,
MessageType.TOPIC: format_generic,
MessageType.NETSPLIT_JOIN: format_netsplit_join,
MessageType.NETSPLIT_QUIT: format_netsplit_quit,
MessageType.INVITE: format_generic,
}
|