class _Channel:
user_ids: _CompletableStringsList
+ topic: tuple[str, str] = ('', '')
def __init__(self,
get_id_for_nick: Callable,
else:
key, data = _Dict.key_val_from_eq_str(item)
self._db.isupport[key] = data
+ elif ret['verb'] == '332': # RPL_TOPIC
+ self._db.channels[ret['channel']].topic = (ret['topic'], '')
+ elif ret['verb'] == '333': # RPL_TOPICWHOTIME
+ self._db.channels[ret['channel']].topic = (
+ self._db.channels[ret['channel']].topic[0],
+ str(ret['author']))
elif ret['verb'] == '353': # RPL_NAMREPLY
self._db.channels[ret['channel']].add_from_namreply(ret['names'])
elif ret['verb'] == '366': # RPL_ENDOFNAMES
ch.remove_nick(ret['quitter'])
self._log(f'{ret["quitter"]} quits: {ret["message"]}',
LogScope.CHAT, target=ch_name)
+ elif ret['verb'] == 'TOPIC':
+ self._db.channels[ret['channel']].topic = (ret['topic'],
+ str(ret['author']))
ClientsDb = dict[str, Client]
class _UpdatingChannel(_UpdatingNode):
user_ids: tuple[str, ...] = tuple()
+ topic: tuple[str, str] = ('', '')
log_scopes = {tuple(): LogScope.CHAT}
def set_and_check_for_change(self, update: _Update
) -> Optional[tuple[LogScope, Any]]:
+ assert isinstance(update.value, tuple)
+ msg: str | dict[str, tuple[str, ...]] = {}
if update.path == ('user_ids',):
- assert isinstance(update.value, tuple)
- d: dict[str, tuple[str, ...]] = {}
+ msg = {}
if not self.user_ids:
- d['nicks:residents'] = tuple(update.value)
+ msg['nicks:residents'] = tuple(update.value)
else:
- d['nuhs:joining'] = tuple(id_ for id_ in update.value
- if id_ not in self.user_ids)
- d['nuhs:parting'] = tuple(id_ for id_ in self.user_ids
- if id_ not in update.value)
- if super().set_and_check_for_change(update):
- return (self._scope(update.path), d)
- return None
- return super().set_and_check_for_change(update)
+ msg['nuhs:joining'] = tuple(id_ for id_ in update.value
+ if id_ not in self.user_ids)
+ msg['nuhs:parting'] = tuple(id_ for id_ in self.user_ids
+ if id_ not in update.value)
+ elif update.path == ('topic',):
+ if '' in update.value:
+ return None
+ msg = f'raw:topic set by {update.value[1]} to: {update.value[0]}'
+ if super().set_and_check_for_change(update):
+ return (self._scope(update.path), msg)
+ return None
class _UpdatingNickUserHost(_UpdatingNode, NickUserHost):
(str(nuh) if transform == 'nuhs' else nuh.nick)
for nuh in nuhs])
self.log(f'{verb}: {item}', **log_kwargs)
+ elif isinstance(value, str) and value.startswith('raw:'):
+ self.log(value.split(':', maxsplit=1)[1], **log_kwargs)
else:
announcement = f'{log_path} set to:'
if isinstance(value, tuple):
# joining/leaving
MSG_EXPECTATIONS += [
+ _MsgParseExpectation(MsgTok.SERVER,
+ '332', # RPL_TOPIC
+ ((MsgTok.NICKNAME, 'set_me_attr:nick'),
+ (MsgTok.CHANNEL, ':channel'),
+ (MsgTok.ANY, ':topic'))),
+ _MsgParseExpectation(MsgTok.SERVER,
+ '333', # RPL_TOPICWHOTIME
+ ((MsgTok.NICKNAME, 'set_me_attr:nick'),
+ (MsgTok.CHANNEL, ':channel'),
+ (MsgTok.NICK_USER_HOST, ':author'),
+ (MsgTok.ANY, ':timestamp'))),
_MsgParseExpectation(MsgTok.SERVER,
'353', # RPL_NAMREPLY
((MsgTok.NICKNAME, 'set_me_attr:nick'),
_MsgParseExpectation((MsgTok.NICK_USER_HOST, ':parter'),
'PART',
((MsgTok.CHANNEL, ':channel'),)),
+ _MsgParseExpectation((MsgTok.NICK_USER_HOST, ':author'),
+ 'TOPIC',
+ ((MsgTok.CHANNEL, ':channel'),
+ (MsgTok.ANY, ':topic'))),
]
# messaging