CHANNEL = auto()
LIST = auto()
NICKNAME = auto()
- NICKNAME_ME = auto()
NONE = auto()
SERVER = auto()
USER_ADDRESS = auto()
- USER_ADDRESS_ME = auto()
+
+
+_MsgTokGuide = str | _MsgTok | tuple[_MsgTok, str]
class _MsgParseExpectation(NamedTuple):
- source: _MsgTok
+ source: _MsgTokGuide
verb: str
+ params: tuple[_MsgTokGuide, ...] = tuple()
len_min_params: int = 0
len_max_params: int = 0
- params: tuple[str | _MsgTok, ...] = tuple()
_EXPECTATIONS: tuple[_MsgParseExpectation, ...] = (
- _MsgParseExpectation(_MsgTok.SERVER, '005', 3, 15),
- _MsgParseExpectation(_MsgTok.SERVER, '353',
- params=(_MsgTok.NICKNAME_ME, '=', _MsgTok.CHANNEL,
- _MsgTok.LIST,)),
- _MsgParseExpectation(_MsgTok.SERVER, '366',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.CHANNEL,
- _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.SERVER, '372',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.SERVER, '376',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.SERVER, '396', 3),
- _MsgParseExpectation(_MsgTok.SERVER, '401', 3),
- _MsgParseExpectation(_MsgTok.SERVER, '432', 3),
- _MsgParseExpectation(_MsgTok.SERVER, '433', 3),
- _MsgParseExpectation(_MsgTok.SERVER, '900', 4),
- _MsgParseExpectation(_MsgTok.SERVER, '903',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.SERVER, '904',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.NONE, 'AUTHENTICATE', params=('+',)),
- _MsgParseExpectation(_MsgTok.SERVER, 'CAP', 3, 15),
- _MsgParseExpectation(_MsgTok.NONE, 'ERROR', params=(_MsgTok.ANY,)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS_ME, 'JOIN',
- params=(_MsgTok.CHANNEL,)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'JOIN',
- params=(_MsgTok.CHANNEL,)),
- _MsgParseExpectation(_MsgTok.NICKNAME_ME, 'MODE',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS_ME, 'MODE',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS_ME, 'NICK',
- params=(_MsgTok.NICKNAME,)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'NICK',
- params=(_MsgTok.NICKNAME,)),
- _MsgParseExpectation(_MsgTok.SERVER, 'NOTICE', params=('*', _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.SERVER, 'NOTICE',
- params=(_MsgTok.NICKNAME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'NOTICE',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS_ME, 'PART',
- params=(_MsgTok.CHANNEL,)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'PART',
- params=(_MsgTok.CHANNEL,)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'PART',
- params=(_MsgTok.CHANNEL, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.NONE, 'PING', params=(_MsgTok.ANY,)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'PRIVMSG',
- params=(_MsgTok.NICKNAME_ME, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'PRIVMSG',
- params=(_MsgTok.CHANNEL, _MsgTok.ANY)),
- _MsgParseExpectation(_MsgTok.USER_ADDRESS, 'QUIT', params=(_MsgTok.ANY,)),
+ _MsgParseExpectation(_MsgTok.SERVER, '005', tuple(), 3, 15),
+
+ _MsgParseExpectation(_MsgTok.SERVER,
+ '353',
+ (_MsgTok.NICKNAME,
+ '=',
+ (_MsgTok.CHANNEL, 'channel'),
+ (_MsgTok.LIST, 'names'))),
+ _MsgParseExpectation(_MsgTok.SERVER,
+ '366',
+ (_MsgTok.NICKNAME,
+ (_MsgTok.CHANNEL, 'channel'),
+ _MsgTok.ANY)),
+
+ _MsgParseExpectation(_MsgTok.SERVER,
+ '372',
+ (_MsgTok.NICKNAME,
+ (_MsgTok.ANY, 'line'))),
+ _MsgParseExpectation(_MsgTok.SERVER,
+ '376',
+ (_MsgTok.NICKNAME,
+ _MsgTok.ANY)),
+
+ _MsgParseExpectation(_MsgTok.SERVER, '396', tuple(), 3),
+ _MsgParseExpectation(_MsgTok.SERVER, '401', tuple(), 3),
+ _MsgParseExpectation(_MsgTok.SERVER, '432', tuple(), 3),
+ _MsgParseExpectation(_MsgTok.SERVER, '433', tuple(), 3),
+ _MsgParseExpectation(_MsgTok.SERVER, '900', tuple(), 4),
+
+ _MsgParseExpectation(_MsgTok.SERVER,
+ '903',
+ (_MsgTok.NICKNAME,
+ (_MsgTok.ANY, 'result'))),
+ _MsgParseExpectation(_MsgTok.SERVER,
+ '904',
+ (_MsgTok.NICKNAME,
+ (_MsgTok.ANY, 'result'))),
+
+ _MsgParseExpectation(_MsgTok.NONE,
+ 'AUTHENTICATE',
+ ('+',)),
+
+ _MsgParseExpectation(_MsgTok.SERVER, 'CAP', tuple(), 3, 15),
+
+ _MsgParseExpectation(_MsgTok.NONE,
+ 'ERROR',
+ ((_MsgTok.ANY, 'reason'),)),
+
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'joiner'),
+ 'JOIN',
+ ((_MsgTok.CHANNEL, 'channel'),)),
+
+ _MsgParseExpectation(_MsgTok.NICKNAME,
+ 'MODE',
+ (_MsgTok.NICKNAME,
+ (_MsgTok.ANY, 'mode'))),
+ _MsgParseExpectation(_MsgTok.USER_ADDRESS,
+ 'MODE',
+ (_MsgTok.NICKNAME,
+ (_MsgTok.ANY, 'mode'))),
+
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'named'),
+ 'NICK',
+ ((_MsgTok.NICKNAME, 'nickname'),)),
+
+ _MsgParseExpectation(_MsgTok.SERVER,
+ 'NOTICE',
+ ('*',
+ (_MsgTok.ANY, 'message'))),
+ _MsgParseExpectation(_MsgTok.SERVER,
+ 'NOTICE',
+ ((_MsgTok.NICKNAME, 'nickname'),
+ (_MsgTok.ANY, 'message'))),
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'sender'),
+ 'NOTICE',
+ ((_MsgTok.NICKNAME, 'nickname'),
+ (_MsgTok.ANY, 'message'))),
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'sender'),
+ 'NOTICE',
+ ((_MsgTok.CHANNEL, 'channel'),
+ (_MsgTok.ANY, 'message'))),
+
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'parter'),
+ 'PART',
+ ((_MsgTok.CHANNEL, 'channel'),)),
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'parter'),
+ 'PART',
+ ((_MsgTok.CHANNEL, 'channel'),
+ (_MsgTok.ANY, 'reason'))),
+
+ _MsgParseExpectation(_MsgTok.NONE,
+ 'PING',
+ ((_MsgTok.ANY, 'reply'),)),
+
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'sender'),
+ 'PRIVMSG',
+ ((_MsgTok.NICKNAME, 'nickname'),
+ (_MsgTok.ANY, 'message'))),
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'sender'),
+ 'PRIVMSG',
+ ((_MsgTok.CHANNEL, 'channel'),
+ (_MsgTok.ANY, 'message'))),
+
+ _MsgParseExpectation((_MsgTok.USER_ADDRESS, 'quitter'),
+ 'QUIT',
+ ((_MsgTok.ANY, 'message'),)),
)
continue
if (not ex.len_max_params) and len_p != ex.len_min_params:
continue
- to_return: dict[str, Any] = {'': ''}
+ to_return: dict[str, Any] = {'': ''} # non-emtpy so boolish True
ex_tok_fields = tuple([ex.source] + list(ex.params))
msg_tok_fields = tuple([msg.source] + list(msg.params))
if ex.params and len(ex_tok_fields) != len(msg_tok_fields):
passing = True
for idx, ex_tok in enumerate(ex_tok_fields):
passing = False
- msg_tok = msg_tok_fields[idx]
+ ex_tok, key = ((ex_tok[0], ex_tok[1])
+ if isinstance(ex_tok, tuple) else (ex_tok, ''))
+ val_to_ret: str | tuple[str, ...] | dict[str, str | _ChannelDb]
+ val_to_ret = msg_tok = msg_tok_fields[idx]
if ex_tok is _MsgTok.NONE and msg_tok != '':
break
if ex_tok is _MsgTok.SERVER\
and ('.' not in msg_tok or set('@!') & set(msg_tok)):
break
- key_nick = 'sender' if not idx else 'nickname'
if ex_tok is _MsgTok.CHANNEL:
if msg_tok[0] != '#':
break
- to_return |= {'ch_name': msg_tok,
- 'channel': self._db.chan(msg_tok)}
- elif ex_tok in {_MsgTok.NICKNAME, _MsgTok.NICKNAME_ME}:
+ val_to_ret = {'id': msg_tok, 'db': self._db.chan(msg_tok)}
+ elif ex_tok is _MsgTok.NICKNAME:
if msg_tok[0] in '~&@%+# *':
break
- to_return[key_nick] = msg_tok
- elif ex_tok in {_MsgTok.USER_ADDRESS, _MsgTok.USER_ADDRESS_ME}:
+ elif ex_tok is _MsgTok.USER_ADDRESS:
toks = msg_tok.split('!')
if len(toks) != 2:
break
toks = toks[0:1] + toks[1].split('@')
if len(toks) != 3:
break
- to_return[key_nick] = toks[0]
- elif ex_tok is _MsgTok.ANY:
- to_return['any'] = msg_tok
+ val_to_ret = tuple(toks)
elif ex_tok is _MsgTok.LIST:
- to_return['list'] = msg_tok.split()
- if ex_tok in {_MsgTok.NICKNAME_ME, _MsgTok.USER_ADDRESS_ME}:
- if to_return[key_nick] != self._db.nickname:
- break
- to_return[f'{key_nick}_me'] = to_return[key_nick]
+ val_to_ret = tuple(msg_tok.split())
+ if key:
+ to_return[key] = val_to_ret
passing = True
if passing:
return to_return
return
if self._match_msg(msg, '005'): # RPL_ISUPPORT
self._db.process_isupport(msg.params[1:-1])
+
elif (ret := self._match_msg(msg, '353')): # RPL_NAMREPLY
- for usr in ret['list']:
- ret['channel'].append_completable('users', usr.lstrip('~&@%+'))
- elif self._match_msg(msg, '366'): # RPL_ENDOFNAMES
- self._db.chan(msg.params[1]).declare_complete('users')
+ for name in ret['names']:
+ ret['channel']['db'].append_completable('users',
+ name.lstrip('~&@%+'))
+ elif (ret := self._match_msg(msg, '366')): # RPL_ENDOFNAMES
+ ret['channel']['db'].declare_complete('users')
+
elif (ret := self._match_msg(msg, '372')): # RPL_MOTD
- self._db.append_completable('motd', ret['any'])
+ self._db.append_completable('motd', ret['line'])
elif self._match_msg(msg, '376'): # RPL_ENDOFMOTD
self._db.declare_complete('motd')
+
elif self._match_msg(msg, '396'): # RPL_VISIBLEHOST
# '@'-split because <https://defs.ircdocs.horse/defs/numerics>
# claims: "<hostname> can also be in the form <user@hostname>"
self._db.nickname, remainder = msg.params[1].split('!', maxsplit=1)
self._db.username, self._db.client_host = remainder.split('@')
self._db.sasl_account = msg.params[2]
+
elif ((ret := self._match_msg(msg, '903')) # RPL_SASLSUCCESS
or (ret := self._match_msg(msg, '904'))): # ERR_SASLFAIL
- self._db.sasl_auth_state = ret['any']
+ self._db.sasl_auth_state = ret['result']
self._caps.end_negotiation()
+
elif self._match_msg(msg, 'AUTHENTICATE'):
auth = b64encode((self._db.nick_wanted + '\0'
+ self._db.nick_wanted + '\0'
+ self._db.password
).encode('utf-8')).decode('utf-8')
self.send(IrcMessage('AUTHENTICATE', (auth,)))
+
elif self._match_msg(msg, 'CAP'):
if (self._caps.process_msg(msg.params[1:])
and self._db.caps.has('sasl')
self.send(IrcMessage('AUTHENTICATE', ('PLAIN',)))
else:
self._caps.end_negotiation()
+
elif (ret := self._match_msg(msg, 'ERROR')):
- self._db.connection_state = ret['any']
+ self._db.connection_state = ret['reason']
self.close()
+
elif (ret := self._match_msg(msg, 'JOIN')):
- log_msg = f'{ret["sender"]} {msg.verb.lower()}s {ret["ch_name"]}'
- self._log(log_msg, scope=LogScope.CHAT, target=ret['ch_name'])
- if 'sender_me' not in ret:
- ret['channel'].append_completable('users', ret['sender'],
- stay_complete=True)
+ self._log(f'{ret["joiner"][0]} {msg.verb.lower()}s '
+ + f'{ret["channel"]["id"]}',
+ scope=LogScope.CHAT, target=ret['channel']['id'])
+ if ret['joiner'][0] != self._db.nickname:
+ ret['channel']['db'].append_completable('users',
+ ret['joiner'][0], True)
+
elif (ret := self._match_msg(msg, 'MODE')):
- self._db.user_modes = ret['any']
+ self._db.user_modes = ret['mode']
+
elif (ret := self._match_msg(msg, 'NICK')):
- if 'sender_me' in ret:
+ if ret['named'][0] == self._db.nickname:
self.set_nick(ret['nickname'], confirmed=True)
else:
- for nom, chan in self._db.chans_of_user(ret['sender']).items():
- chan.remove_completable('users', ret['sender'], True)
- chan.append_completable('users', ret['nickname'], True)
- self._log(f'{ret["sender"]} becomes {ret["nickname"]}',
- scope=LogScope.CHAT, target=nom)
+ for id_, ch in self._db.chans_of_user(ret['named'][0]).items():
+ ch.remove_completable('users', ret['named'][0], True)
+ ch.append_completable('users', ret['nickname'], True)
+ self._log(f'{ret["named"][0]} becomes {ret["nickname"]}',
+ scope=LogScope.CHAT, target=id_)
+
elif (ret := self._match_msg(msg, 'NOTICE'))\
or (ret := self._match_msg(msg, 'PRIVMSG')):
- kw = {'sender': ret['sender'], 'scope': LogScope.CHAT,
- 'target': ret.get('ch_name', ret['sender']),
- } if 'sender' in ret else {}
- if msg.verb == 'NOTICE':
- if 'nickname' in ret and 'nickname_me' not in ret:
- self.set_nick(ret['nickname'], confirmed=True)
- kw |= {'as_notice': True}
- self._log(ret['any'], out=False, **kw)
+ if 'nickname' in ret and ret['nickname'] != self._db.nickname:
+ self.set_nick(ret['nickname'], confirmed=True)
+ kw: dict[str, bool | str | LogScope] = {
+ 'as_notice': msg.verb == 'NOTICE'}
+ if 'sender' in ret: # not just server message
+ kw |= {'sender': ret['sender'][0], 'scope': LogScope.CHAT,
+ 'target': (ret['sender'][0] if 'nickname' in ret
+ else ret['channel']['id'])}
+ self._log(ret['message'], out=False, **kw)
+
elif (ret := self._match_msg(msg, 'PART')):
- log_msg = f'{ret["sender"]} {msg.verb.lower()}s {ret["ch_name"]}'
- log_msg += f': {ret["any"]}' if 'any' in ret else ''
- self._log(log_msg, scope=LogScope.CHAT, target=ret['ch_name'])
- if 'sender_me' in ret:
+ reason = f': ret["reason"]' if 'reason' in ret else ''
+ self._log(f'{ret["parter"][0]} {msg.verb.lower()}s '
+ + f'{ret["channel"]["id"]}{reason}',
+ scope=LogScope.CHAT, target=ret['channel']['id'])
+ if ret['parter'][0] == self._db.nickname:
self._db.del_chan(ret['ch_name'])
else:
- ret['channel'].remove_completable('users', ret['sender'], True)
+ ret['channel']['db'].remove_completable('users',
+ ret['parter'][0], True)
+
elif (ret := self._match_msg(msg, 'PING')):
- self.send(IrcMessage(verb='PONG', params=(ret['any'],)))
+ self.send(IrcMessage(verb='PONG', params=(ret['reply'],)))
+
elif (ret := self._match_msg(msg, 'QUIT')):
- for nom, chan in self._db.chans_of_user(ret['sender']).items():
- chan.remove_completable('users', ret['sender'], True)
- self._log(f'{ret["sender"]} quits: {ret["any"]}',
- LogScope.CHAT, target=nom)
+ for id_, chan in self._db.chans_of_user(ret['quitter'][0]).items():
+ chan.remove_completable('users', ret['quitter'][0], True)
+ self._log(f'{ret["quitter"][0]} quits: {ret["message"]}',
+ LogScope.CHAT, target=id_)
+
else:
self._log(f'PLEASE IMPLEMENT HANDLER FOR: {msg.raw}')
projects / ircplom / commitdiff