From: Christian Heller Date: Tue, 19 Aug 2025 05:50:28 +0000 (+0200) Subject: Allow more than one matching pattern per verb. X-Git-Url: https://plomlompom.com/repos/process_efforts?a=commitdiff_plain;h=15895541bc0e03e82a41ed5558056c66ffe9bec1;p=ircplom Allow more than one matching pattern per verb. --- diff --git a/ircplom/client.py b/ircplom/client.py index 0a15747..df36cac 100644 --- a/ircplom/client.py +++ b/ircplom/client.py @@ -46,37 +46,40 @@ class _MsgSource(Enum): class _MsgParseExpectation(NamedTuple): + verb: str len_min_params: int = 0 len_max_params: int = 0 params: tuple[str, ...] = tuple() source: Optional[_MsgSource] = None -_EXPECTATIONS: dict[str, _MsgParseExpectation] = { - '005': _MsgParseExpectation(3, 15, source=_MsgSource.SERVER), - '353': _MsgParseExpectation(4, source=_MsgSource.SERVER), - '366': _MsgParseExpectation(3, source=_MsgSource.SERVER), - '372': _MsgParseExpectation(2, source=_MsgSource.SERVER), - '376': _MsgParseExpectation(2, source=_MsgSource.SERVER), - '396': _MsgParseExpectation(3, source=_MsgSource.SERVER), - '401': _MsgParseExpectation(3, source=_MsgSource.SERVER), - '432': _MsgParseExpectation(3, source=_MsgSource.SERVER), - '433': _MsgParseExpectation(3, source=_MsgSource.SERVER), - '900': _MsgParseExpectation(4, source=_MsgSource.SERVER), - '903': _MsgParseExpectation(2, source=_MsgSource.SERVER), - '904': _MsgParseExpectation(2, source=_MsgSource.SERVER), - 'AUTHENTICATE': _MsgParseExpectation(params=('+',), source=_MsgSource.NONE), - 'CAP': _MsgParseExpectation(3, 15, source=_MsgSource.SERVER), - 'ERROR': _MsgParseExpectation(1, source=_MsgSource.NONE), - 'JOIN': _MsgParseExpectation(1, source=_MsgSource.USER_ADDRESS), - 'MODE': _MsgParseExpectation(2, source=_MsgSource.USER_ADDRESS), - 'NICK': _MsgParseExpectation(1, source=_MsgSource.USER_ADDRESS), - 'NOTICE': _MsgParseExpectation(2), - 'PART': _MsgParseExpectation(1, 2, source=_MsgSource.USER_ADDRESS), - 'PING': _MsgParseExpectation(1, source=_MsgSource.NONE), - 'PRIVMSG': _MsgParseExpectation(2), - 'QUIT': _MsgParseExpectation(1, source=_MsgSource.USER_ADDRESS), -} +_EXPECTATIONS: tuple[_MsgParseExpectation, ...] = ( + _MsgParseExpectation('005', 3, 15, source=_MsgSource.SERVER), + _MsgParseExpectation('353', 4, source=_MsgSource.SERVER), + _MsgParseExpectation('366', 3, source=_MsgSource.SERVER), + _MsgParseExpectation('372', 2, source=_MsgSource.SERVER), + _MsgParseExpectation('376', 2, source=_MsgSource.SERVER), + _MsgParseExpectation('396', 3, source=_MsgSource.SERVER), + _MsgParseExpectation('401', 3, source=_MsgSource.SERVER), + _MsgParseExpectation('432', 3, source=_MsgSource.SERVER), + _MsgParseExpectation('433', 3, source=_MsgSource.SERVER), + _MsgParseExpectation('900', 4, source=_MsgSource.SERVER), + _MsgParseExpectation('903', 2, source=_MsgSource.SERVER), + _MsgParseExpectation('904', 2, source=_MsgSource.SERVER), + _MsgParseExpectation('AUTHENTICATE', params=('+',), source=_MsgSource.NONE), + _MsgParseExpectation('CAP', 3, 15, source=_MsgSource.SERVER), + _MsgParseExpectation('ERROR', 1, source=_MsgSource.NONE), + _MsgParseExpectation('JOIN', 1, source=_MsgSource.USER_ADDRESS), + _MsgParseExpectation('MODE', 2, source=_MsgSource.USER_ADDRESS), + _MsgParseExpectation('NICK', 1, source=_MsgSource.USER_ADDRESS), + _MsgParseExpectation('NOTICE', 2, source=_MsgSource.USER_ADDRESS), + _MsgParseExpectation('NOTICE', 2, source=_MsgSource.SERVER), + _MsgParseExpectation('PART', 1, 2, source=_MsgSource.USER_ADDRESS), + _MsgParseExpectation('PING', 1, source=_MsgSource.NONE), + _MsgParseExpectation('PRIVMSG', 2, source=_MsgSource.USER_ADDRESS), + _MsgParseExpectation('PRIVMSG', 2, source=_MsgSource.SERVER), + _MsgParseExpectation('QUIT', 1, source=_MsgSource.USER_ADDRESS), +) class _IrcMsg(IrcMessage): @@ -560,39 +563,40 @@ class Client(ABC, ClientQueueMixin): self.close() def _match_msg(self, msg: _IrcMsg, verb: str): - 'Test .verb, .params.' + 'Test .source, .verb, .params.' to_return: dict[str, Optional[str]] = {'_': None} if not msg.verb == verb: return False - expect = _EXPECTATIONS[verb] - if expect.source is _MsgSource.NONE: - if msg.source != '': - return False - elif expect.source is _MsgSource.SERVER: - if ('!' in msg.source - or '@' in msg.source - or '.' not in msg.source - or self._db.hostname.split('.')[-2:] - != msg.source.split('.')[-2:]): - return False - elif expect.source is _MsgSource.USER_ADDRESS: - toks = msg.source.split('!') - if len(toks) != 2: - return False - toks = toks[0:1] + toks[1].split('@') - if len(toks) != 3: - return False - to_return['nickname'] = toks[0] - if expect.params and msg.params != expect.params: - return False - n_len_params = len(msg.params) - if expect.len_max_params: - if not (expect.len_min_params <= n_len_params - <= expect.len_max_params): - return False - elif n_len_params != expect.len_min_params: - return False - return to_return + for expect in [x for x in _EXPECTATIONS if x.verb == msg.verb]: + if expect.source is _MsgSource.NONE: + if msg.source != '': + continue + elif expect.source is _MsgSource.SERVER: + if ('!' in msg.source + or '@' in msg.source + or '.' not in msg.source + or self._db.hostname.split('.')[-2:] + != msg.source.split('.')[-2:]): + continue + elif expect.source is _MsgSource.USER_ADDRESS: + toks = msg.source.split('!') + if len(toks) != 2: + continue + toks = toks[0:1] + toks[1].split('@') + if len(toks) != 3: + continue + to_return['nickname'] = toks[0] + if expect.params and msg.params != expect.params: + continue + n_len_params = len(msg.params) + if expect.len_max_params: + if not (expect.len_min_params <= n_len_params + <= expect.len_max_params): + continue + elif n_len_params != expect.len_min_params: + continue + return to_return + return False def handle_msg(self, msg: _IrcMsg) -> None: 'Log msg.raw, then process incoming msg into appropriate client steps.' @@ -670,17 +674,15 @@ class Client(ABC, ClientQueueMixin): elif (ret := self._match_msg(msg, 'NICK'))\ and ret['nickname'] == self._db.nickname: self.set_nick(msg.params[0], confirmed=True) - elif self._match_msg(msg, 'NOTICE')\ + elif (ret := self._match_msg(msg, 'NOTICE'))\ and (msg.params[0] != '*' or not self._db.nickname): - kw: dict[str, str | LogScope] = {} - if '!' in msg.source: - kw |= {'sender': msg.nick_from_source, 'scope': LogScope.CHAT} + kw = {'sender': ret['nickname'], 'scope': LogScope.CHAT + } if 'nickname' in ret else {} self._log(msg.params[-1], out=False, target=msg.params[0], as_notice=True, **kw) - elif self._match_msg(msg, 'PRIVMSG') and msg.params[0] != '*': - kw = {} - if '!' in msg.source: - kw |= {'sender': msg.nick_from_source, 'scope': LogScope.CHAT} + elif (ret := self._match_msg(msg, 'PRIVMSG')) and msg.params[0] != '*': + kw = {'sender': ret['nickname'], 'scope': LogScope.CHAT + } if 'nickname' in ret else {} self._log(msg.params[-1], out=False, target=msg.params[0], **kw) elif (ret := self._match_msg(msg, 'PART')): channel = msg.params[0]