From: Christian Heller Date: Sat, 6 Sep 2025 04:46:19 +0000 (+0200) Subject: Better encapsulate channel and user dictionaries code. X-Git-Url: https://plomlompom.com/repos/day?a=commitdiff_plain;h=c5b7b0364cfb808b6949da2f675a1c2ee8678b6e;p=ircplom Better encapsulate channel and user dictionaries code. --- diff --git a/ircplom/client.py b/ircplom/client.py index f8e4c1d..d57dabe 100644 --- a/ircplom/client.py +++ b/ircplom/client.py @@ -409,6 +409,27 @@ class _User(_NickUserHost): modes: str = '?' exit_msg: str = '' + def __init__(self, + names_channels_of_user: Callable, + remove_from_channels: Callable, + **kwargs) -> None: + self.names_channels = lambda: names_channels_of_user(self) + self._remove_from_channels = lambda name='': remove_from_channels(self, + name) + super().__init__(**kwargs) + + def part(self, channel_name: str, exit_msg: str) -> None: + 'First set .exit_msg, then remove from channel of channel_name.' + self.exit_msg = f'P{exit_msg}' + self._remove_from_channels(channel_name) + self.exit_msg = '' + + def quit(self, exit_msg: str) -> None: + 'First set .exit_msg, then remove from any channels.' + self.exit_msg = f'Q{exit_msg}' + self._remove_from_channels() + self.exit_msg = '' + @property def id_(self) -> str: 'To be set to key inside dictionary if placed into one.' @@ -484,11 +505,36 @@ class _UpdatingUsersDict(_UpdatingDict[_UpdatingUser]): return None return id_ + def purge(self) -> None: + 'Remove all not linked to by existing channels, except of our ID "me".' + for id_ in [id_ for id_, user in self._dict.items() + if id_ != 'me' and not user.names_channels()]: + del self[id_] + + +class _UpdatingChannelsDict(_UpdatingDict[_UpdatingChannel]): + + def _of_user(self, user: _User) -> dict[str, _UpdatingChannel]: + return {k: v for k, v in self._dict.items() + if user.id_ in v.user_ids.completed} + + def of_user(self, user: _User) -> tuple[str, ...]: + 'Return names of channels listing user as member.' + return tuple(self._of_user(user).keys()) + + def remove_user(self, user: _User, target: str) -> None: + 'Remove user from channel named "target", or all with user if empty.' + if target: + self[target].remove_user(user) + else: + for channel in self._of_user(user).values(): + channel.remove_user(user) + class _ClientDb(_UpdatingMixin, SharedClientDbFields): _keep_on_clear = set(IrcConnSetup.__annotations__.keys()) caps: _UpdatingDict[_UpdatingServerCapability] - channels: _UpdatingDict[_UpdatingChannel] + channels: _UpdatingChannelsDict isupport: _UpdatingDict[str] motd: _UpdatingCompletableStringsList users: _UpdatingUsersDict @@ -497,12 +543,18 @@ class _ClientDb(_UpdatingMixin, SharedClientDbFields): attr = super().__getattribute__(key) if key == 'isupport' and not attr._defaults: attr._defaults = ISUPPORT_DEFAULTS - elif key == 'channels' and attr._create_if_none is None: + elif key == 'channels' and attr._create_if_none is None\ + and super().__getattribute__('users' + )._create_if_none is not None: attr._create_if_none = { 'userid_for_nickuserhost': self.users.id_for_nickuserhost, 'get_membership_prefixes': self._get_membership_prefixes, - 'purge_users': self.purge_users} - elif key in {'users', 'caps'} and attr._create_if_none is None: + 'purge_users': self.users.purge} + elif key == 'users' and attr._create_if_none is None: + attr._create_if_none = { + 'names_channels_of_user': self.channels.of_user, + 'remove_from_channels': self.channels.remove_user} + elif key == 'caps' and attr._create_if_none is None: attr._create_if_none = {} return attr @@ -515,12 +567,6 @@ class _ClientDb(_UpdatingMixin, SharedClientDbFields): elif issubclass(value, str): setattr(self, key, '') - def purge_users(self) -> None: - 'Remove from .users all not linked to by existing channels, except us.' - for id_ in self.users.keys(): - if id_ != 'me' and not self.chans_of_user(id_): - del self.users[id_] - def is_nick(self, nick: str) -> bool: 'Tests name to match rules for nicknames.' if len(nick) == 0: @@ -540,11 +586,6 @@ class _ClientDb(_UpdatingMixin, SharedClientDbFields): assert toks[0][0] == '(' return toks[1] - def chans_of_user(self, user_id: str) -> dict[str, _UpdatingChannel]: - 'Return dictionary of channels user is in.' - return {k: self.channels[k] for k in self.channels.keys() - if user_id in self.channels[k].user_ids.completed} - class _CapsManager: @@ -785,18 +826,14 @@ class Client(ABC, ClientQueueMixin): else ret['channel'])} self._log(ret['message'], out=False, **kw) elif ret['_verb'] == 'PART': - ret['parter'].exit_msg = 'P' + ret.get('message', '') - ret['parter'].exit_msg = '' - self.db.channels[ret['channel']].remove_user(ret['parter']) + ret['parter'].part(ret['channel'], ret.get('message', '')) if ret['parter'] is self.db.users['me']: del self.db.channels[ret['channel']] - self.db.purge_users() + self.db.users.purge() elif ret['_verb'] == 'PING': self.send(IrcMessage(verb='PONG', params=(ret['reply'],))) elif ret['_verb'] == 'QUIT': - ret['quitter'].exit_msg = 'Q' + ret['message'] - for channel in self.db.chans_of_user(ret['quitter'].id_).values(): - channel.remove_user(ret['quitter']) + ret['quitter'].quit(ret['message']) ClientsDb = dict[str, Client]