Source code for pydle.features.isupport

## isupport.py
# ISUPPORT (server-side IRC extension indication) support.
# See: http://tools.ietf.org/html/draft-hardy-irc-isupport-00
import collections
import pydle.protocol
from pydle.features import rfc1459

__all__ = [ 'ISUPPORTSupport' ]


FEATURE_DISABLED_PREFIX = '-'
BAN_EXCEPT_MODE = 'e'
INVITE_EXCEPT_MODE = 'I'


[docs]class ISUPPORTSupport(rfc1459.RFC1459Support): """ ISUPPORT support. """ ## Internal overrides. def _reset_attributes(self): super()._reset_attributes() self._isupport = {} self._extban_types = [] self._extban_prefix = None def _create_channel(self, channel): """ Create channel with optional ban and invite exception lists. """ super()._create_channel(channel) if 'EXCEPTS' in self._isupport: self.channels[channel]['exceptlist'] = None if 'INVEX' in self._isupport: self.channels[channel]['inviteexceptlist'] = None ## Command handlers. async def on_raw_005(self, message): """ ISUPPORT indication. """ isupport = {} # Parse response. # Strip target (first argument) and 'are supported by this server' (last argument). for feature in message.params[1:-1]: if feature.startswith(FEATURE_DISABLED_PREFIX): value = False elif '=' in feature: feature, value = feature.split('=', 1) else: value = True isupport[feature.upper()] = value # Update internal dict first. self._isupport.update(isupport) # And have callbacks update other internals. for entry, value in isupport.items(): if value != False: # A value of True technically means there was no value supplied; correct this for callbacks. if value == True: value = None method = 'on_isupport_' + pydle.protocol.identifierify(entry) if hasattr(self, method): await getattr(self, method)(value) ## ISUPPORT handlers. async def on_isupport_awaylen(self, value): """ Away message length limit. """ self._away_message_length_limit = int(value) async def on_isupport_casemapping(self, value): """ IRC case mapping for nickname and channel name comparisons. """ if value in rfc1459.protocol.CASE_MAPPINGS: self._case_mapping = value self.channels = rfc1459.parsing.NormalizingDict(self.channels, case_mapping=value) self.users = rfc1459.parsing.NormalizingDict(self.users, case_mapping=value) async def on_isupport_channellen(self, value): """ Channel name length limit. """ self._channel_length_limit = int(value) async def on_isupport_chanlimit(self, value): """ Simultaneous channel limits for user. """ self._channel_limits = {} for entry in value.split(','): types, limit = entry.split(':') # Assign limit to channel type group and add lookup entry for type. self._channel_limits[frozenset(types)] = int(limit) for prefix in types: self._channel_limit_groups[prefix] = frozenset(types) async def on_isupport_chanmodes(self, value): """ Valid channel modes and their behaviour. """ list, param, param_set, noparams = [ set(modes) for modes in value.split(',')[:4] ] self._channel_modes.update(set(value.replace(',', ''))) # The reason we have to do it like this is because other ISUPPORTs (e.g. PREFIX) may update these values as well. if not rfc1459.protocol.BEHAVIOUR_LIST in self._channel_modes_behaviour: self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_LIST] = set() self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_LIST].update(list) if not rfc1459.protocol.BEHAVIOUR_PARAMETER in self._channel_modes_behaviour: self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_PARAMETER] = set() self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_PARAMETER].update(param) if not rfc1459.protocol.BEHAVIOUR_PARAMETER_ON_SET in self._channel_modes_behaviour: self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_PARAMETER_ON_SET] = set() self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_PARAMETER_ON_SET].update(param_set) if not rfc1459.protocol.BEHAVIOUR_NO_PARAMETER in self._channel_modes_behaviour: self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_NO_PARAMETER] = set() self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_NO_PARAMETER].update(noparams) async def on_isupport_chantypes(self, value): """ Channel name prefix symbols. """ if not value: value = '' self._channel_prefixes = set(value) async def on_isupport_excepts(self, value): """ Server allows ban exceptions. """ if not value: value = BAN_EXCEPT_MODE self._channel_modes.add(value) self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_LIST].add(value) async def on_isupport_extban(self, value): """ Extended ban prefixes. """ self._extban_prefix, types = value.split(',') self._extban_types = set(types) async def on_isupport_invex(self, value): """ Server allows invite exceptions. """ if not value: value = INVITE_EXCEPT_MODE self._channel_modes.add(value) self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_LIST].add(value) async def on_isupport_maxbans(self, value): """ Maximum entries in ban list. Replaced by MAXLIST. """ if 'MAXLIST' not in self._isupport: if not self._list_limits: self._list_limits = {} self._list_limits['b'] = int(value) async def on_isupport_maxchannels(self, value): """ Old version of CHANLIMIT. """ if 'CHANTYPES' in self._isupport and 'CHANLIMIT' not in self._isupport: self._channel_limits = {} prefixes = self._isupport['CHANTYPES'] # Assume the limit is for all types of channels. Make a single group for all types. self._channel_limits[frozenset(prefixes)] = int(value) for prefix in prefixes: self._channel_limit_groups[prefix] = frozenset(prefixes) async def on_isupport_maxlist(self, value): """ Limits on channel modes involving lists. """ self._list_limits = {} for entry in value.split(','): modes, limit = entry.split(':') # Assign limit to mode group and add lookup entry for mode. self._list_limits[frozenset(modes)] = int(limit) for mode in modes: self._list_limit_groups[mode] = frozenset(modes) async def on_isupport_maxpara(self, value): """ Limits to parameters given to command. """ self._command_parameter_limit = int(value) async def on_isupport_modes(self, value): """ Maximum number of variable modes to change in a single MODE command. """ self._mode_limit = int(value) async def on_isupport_namesx(self, value): """ Let the server know we do in fact support NAMESX. Effectively the same as CAP multi-prefix. """ await self.rawmsg('PROTOCTL', 'NAMESX') async def on_isupport_network(self, value): """ IRC network name. """ self.network = value async def on_isupport_nicklen(self, value): """ Nickname length limit. """ self._nickname_length_limit = int(value) async def on_isupport_prefix(self, value): """ Nickname prefixes on channels and their associated modes. """ if not value: # No prefixes support. self._nickname_prefixes = collections.OrderedDict() return modes, prefixes = value.lstrip('(').split(')', 1) # Update valid channel modes and their behaviour as CHANMODES doesn't include PREFIX modes. self._channel_modes.update(set(modes)) if not rfc1459.protocol.BEHAVIOUR_PARAMETER in self._channel_modes_behaviour: self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_PARAMETER] = set() self._channel_modes_behaviour[rfc1459.protocol.BEHAVIOUR_PARAMETER].update(set(modes)) self._nickname_prefixes = collections.OrderedDict() for mode, prefix in zip(modes, prefixes): self._nickname_prefixes[prefix] = mode async def on_isupport_statusmsg(self, value): """ Support for messaging every member on a channel with given status or higher. """ self._status_message_prefixes.update(value) async def on_isupport_targmax(self, value): """ The maximum number of targets certain types of commands can affect. """ if not value: return for entry in value.split(','): command, limit = entry.split(':', 1) if not limit: continue self._target_limits[command] = int(limit) async def on_isupport_topiclen(self, value): """ Channel topic length limit. """ self._topic_length_limit = int(value) async def on_isupport_wallchops(self, value): """ Support for messaging every opped member or higher on a channel. Replaced by STATUSMSG. """ for prefix, mode in self._nickname_prefixes.items(): if mode == 'o': break else: prefix = '@' self._status_message_prefixes.add(prefix) async def on_isupport_wallvoices(self, value): """ Support for messaging every voiced member or higher on a channel. Replaced by STATUSMSG. """ for prefix, mode in self._nickname_prefixes.items(): if mode == 'v': break else: prefix = '+' self._status_message_prefixes.add(prefix)