Verified Commit d7b27343 authored by Fabio Manganiello's avatar Fabio Manganiello
Browse files

[#203] Added IRC integration

parent 3ef602cc
Pipeline #99 passed with stages
in 2 minutes and 32 seconds
......@@ -7,10 +7,12 @@ Given the high speed of development in the first phase, changes are being report
### Added
- Added Mastodon integration.
- Added `mastodon` plugin.
- Added `chat.irc` plugin.
### Fixed
- Fixed `switchbot.status` method in case of virtual devices.
- Fixed `switchbot.status` method in case of virtual devices.
## [0.22.4] - 2021-10-19
......
......@@ -281,6 +281,12 @@ autodoc_mock_imports = ['googlesamples.assistant.grpc.audio_helpers',
'aiohttp',
'watchdog',
'pyngrok',
'irc',
'irc.bot',
'irc.strings',
'irc.client',
'irc.connection',
'irc.events',
]
sys.path.insert(0, os.path.abspath('../..'))
......
......@@ -32,6 +32,7 @@ Events
platypush/events/http.hook.rst
platypush/events/http.rss.rst
platypush/events/inotify.rst
platypush/events/irc.rst
platypush/events/joystick.rst
platypush/events/kafka.rst
platypush/events/light.rst
......
``platypush.message.event.irc``
===============================
.. automodule:: platypush.message.event.irc
:members:
``chat.irc``
============
.. automodule:: platypush.plugins.chat.irc
:members:
......@@ -23,6 +23,7 @@ Plugins
platypush/plugins/camera.gstreamer.rst
platypush/plugins/camera.ir.mlx90640.rst
platypush/plugins/camera.pi.rst
platypush/plugins/chat.irc.rst
platypush/plugins/chat.telegram.rst
platypush/plugins/clipboard.rst
platypush/plugins/config.rst
......
......@@ -39,7 +39,9 @@ class Event(Message):
self.disable_web_clients_notification = disable_web_clients_notification
for arg, value in self.args.items():
if arg != 'args':
if arg not in [
'id', 'args', 'origin', 'target', 'type', 'timestamp', 'disable_logging'
] and not arg.startswith('_'):
self.__setattr__(arg, value)
@classmethod
......
from abc import ABC
from base64 import b64encode
from typing import Optional
from platypush.message.event import Event
class IRCEvent(Event, ABC):
"""
IRC base event.
"""
def __init__(self, *args, server: Optional[str] = None, port: Optional[int] = None,
alias: Optional[str] = None, channel: Optional[str] = None, **kwargs):
super().__init__(*args, server=server, port=port, alias=alias, channel=channel, **kwargs)
class IRCChannelJoinEvent(IRCEvent):
"""
Event triggered upon account channel join.
"""
def __init__(self, *args, nick: str, **kwargs):
super().__init__(*args, nick=nick, **kwargs)
class IRCChannelKickEvent(IRCEvent):
"""
Event triggered upon account channel kick.
"""
def __init__(self, *args, target_nick: str, source_nick: Optional[str] = None, **kwargs):
super().__init__(*args, source_nick=source_nick, target_nick=target_nick, **kwargs)
class IRCModeEvent(IRCEvent):
"""
Event triggered when the IRC mode of a channel user changes.
"""
def __init__(
self, *args, mode: str, channel: Optional[str] = None,
source: Optional[str] = None,
target_: Optional[str] = None, **kwargs
):
super().__init__(*args, mode=mode, channel=channel, source=source, target_=target_, **kwargs)
class IRCPartEvent(IRCEvent):
"""
Event triggered when an IRC nick parts.
"""
def __init__(self, *args, nick: str, **kwargs):
super().__init__(*args, nick=nick, **kwargs)
class IRCQuitEvent(IRCEvent):
"""
Event triggered when an IRC nick quits.
"""
def __init__(self, *args, nick: str, **kwargs):
super().__init__(*args, nick=nick, **kwargs)
class IRCNickChangeEvent(IRCEvent):
"""
Event triggered when a IRC nick changes.
"""
def __init__(self, *args, before: str, after: str, **kwargs):
super().__init__(*args, before=before, after=after, **kwargs)
class IRCConnectEvent(IRCEvent):
"""
Event triggered upon server connection.
"""
class IRCDisconnectEvent(IRCEvent):
"""
Event triggered upon server disconnection.
"""
class IRCPrivateMessageEvent(IRCEvent):
"""
Event triggered when a private message is received.
"""
def __init__(self, *args, text: str, nick: str, mentions_me: bool = False, **kwargs):
super().__init__(*args, text=text, nick=nick, mentions_me=mentions_me, **kwargs)
class IRCPublicMessageEvent(IRCEvent):
"""
Event triggered when a public message is received.
"""
def __init__(self, *args, text: str, nick: str, mentions_me: bool = False, **kwargs):
super().__init__(*args, text=text, nick=nick, mentions_me=mentions_me, **kwargs)
class IRCDCCRequestEvent(IRCEvent):
"""
Event triggered when a DCC connection request is received.
"""
def __init__(self, *args, address: str, port: int, nick: str, **kwargs):
super().__init__(*args, address=address, port=port, nick=nick, **kwargs)
class IRCDCCMessageEvent(IRCEvent):
"""
Event triggered when a DCC message is received.
"""
def __init__(self, *args, address: str, body: bytes, **kwargs):
super().__init__(
*args, address=address, body=b64encode(body).decode(), **kwargs
)
class IRCCTCPMessageEvent(IRCEvent):
"""
Event triggered when a CTCP message is received.
"""
def __init__(self, *args, address: str, message: str, **kwargs):
super().__init__(*args, address=address, message=message, **kwargs)
class IRCDCCFileRequestEvent(IRCEvent):
"""
Event triggered when a DCC file send request is received.
"""
def __init__(
self, *args, nick: str, address: str, file: str,
port: int, size: Optional[int] = None, **kwargs
):
super().__init__(
*args, nick=nick, address=address, file=file, port=port,
size=size, **kwargs
)
class IRCDCCFileRecvCompletedEvent(IRCEvent):
"""
Event triggered when a DCC file transfer RECV is completed.
"""
def __init__(
self, *args, address: str, port: int, file: str,
size: Optional[int] = None, **kwargs
):
super().__init__(
*args, address=address, file=file,
port=port, size=size, **kwargs
)
class IRCDCCFileRecvCancelledEvent(IRCEvent):
"""
Event triggered when a DCC file transfer RECV is cancelled.
"""
def __init__(
self, *args, address: str, port: int, file: str,
error: str, **kwargs
):
super().__init__(
*args, address=address, file=file, port=port,
error=error, **kwargs
)
class IRCDCCFileSendCompletedEvent(IRCEvent):
"""
Event triggered when a DCC file transfer SEND is completed.
"""
def __init__(self, *args, address: str, port: int, file: str, **kwargs):
super().__init__(*args, address=address, file=file, port=port, **kwargs)
class IRCDCCFileSendCancelledEvent(IRCEvent):
"""
Event triggered when a DCC file transfer SEND is cancelled.
"""
def __init__(
self, *args, address: str, port: int, file: str,
error: str, **kwargs
):
super().__init__(
*args, address=address, file=file, port=port,
error=error, **kwargs
)
import os
from typing import Sequence, Dict, Tuple, Union, Optional
from platypush.plugins import RunnablePlugin, action
from platypush.schemas.irc import IRCServerSchema, IRCServerStatusSchema, IRCChannelSchema
from ._bot import IRCBot
from .. import ChatPlugin
class ChatIrcPlugin(RunnablePlugin, ChatPlugin):
"""
IRC integration.
This plugin allows you to easily create IRC bots with custom logic that reacts to IRC events
and interact with IRC sessions.
Triggers:
* :class:`platypush.message.event.irc.IRCChannelJoinEvent` when a user joins a channel.
* :class:`platypush.message.event.irc.IRCChannelKickEvent` when a user is kicked from a channel.
* :class:`platypush.message.event.irc.IRCModeEvent` when a user/channel mode change event occurs.
* :class:`platypush.message.event.irc.IRCPartEvent` when a user parts a channel.
* :class:`platypush.message.event.irc.IRCQuitEvent` when a user quits.
* :class:`platypush.message.event.irc.IRCNickChangeEvent` when a user nick changes.
* :class:`platypush.message.event.irc.IRCConnectEvent` when the bot connects to a server.
* :class:`platypush.message.event.irc.IRCDisconnectEvent` when the bot disconnects from a server.
* :class:`platypush.message.event.irc.IRCPrivateMessageEvent` when a private message is received.
* :class:`platypush.message.event.irc.IRCPublicMessageEvent` when a public message is received.
* :class:`platypush.message.event.irc.IRCDCCRequestEvent` when a DCC connection request is received.
* :class:`platypush.message.event.irc.IRCDCCMessageEvent` when a DCC message is received.
* :class:`platypush.message.event.irc.IRCCTCPMessageEvent` when a CTCP message is received.
* :class:`platypush.message.event.irc.IRCDCCFileRequestEvent` when a DCC file request is received.
* :class:`platypush.message.event.irc.IRCDCCFileRecvCompletedEvent` when a DCC file download is completed.
* :class:`platypush.message.event.irc.IRCDCCFileRecvCancelledEvent` when a DCC file download is cancelled.
* :class:`platypush.message.event.irc.IRCDCCFileSendCompletedEvent` when a DCC file upload is completed.
* :class:`platypush.message.event.irc.IRCDCCFileSendCancelledEvent` when a DCC file upload is cancelled.
Requires:
* **irc** (``pip install irc``)
"""
def __init__(self, servers: Sequence[dict], **kwargs):
"""
:param servers: List of servers/channels that the bot will automatically connect/join.
"""
super().__init__(**kwargs)
try:
self._bots: Dict[Tuple[str, int], IRCBot] = {
(server_conf['server'], server_conf['port']): IRCBot(**server_conf)
for server_conf in IRCServerSchema().load(servers, many=True)
}
except Exception as e:
self.logger.warning(f'Could not load IRC server configuration: {e}')
self.logger.exception(e)
raise e
@property
def _bots_by_server(self) -> Dict[str, IRCBot]:
return {
bot.server: bot
for srv, bot in self._bots.items()
}
@property
def _bots_by_server_and_port(self) -> Dict[Tuple[str, int], IRCBot]:
return {
(bot.server, bot.port): bot
for srv, bot in self._bots.items()
}
@property
def _bots_by_alias(self) -> Dict[str, IRCBot]:
return {
bot.alias: bot
for srv, bot in self._bots.items()
if bot.alias
}
def main(self):
self._connect()
self._should_stop.wait()
def _connect(self):
for srv, bot in self._bots.items():
self.logger.info(f'Connecting to IRC server {srv}')
bot.start()
def stop(self):
for srv, bot in self._bots.items():
self.logger.info(f'Disconnecting from IRC server {srv}')
try:
bot.stop(bot.stop_message or 'Application stopped')
except Exception as e:
self.logger.warning(f'Error while stopping connection to {srv}: {e}')
super().stop()
def _get_bot(self, server: Union[str, Tuple[str, int]]) -> IRCBot:
if isinstance(server, (tuple, list, set)):
bot = self._bots_by_server_and_port[tuple(server)]
else:
bot = self._bots_by_alias.get(server, self._bots_by_server.get(server))
assert bot, f'Bot connection to {server} not found'
return bot
@action
def send_file(
self, file: str, server: Union[str, Tuple[str, int]], nick: str, bind_address: Optional[str] = None
):
"""
Send a file to an IRC user over DCC connection.
Note that passive connections are currently not supported.
:param file: Path of the file that should be transferred.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param nick: Target IRC nick.
:param bind_address: DCC listen bind address (default: any).
"""
file = os.path.expanduser(file)
assert os.path.isfile(file), f'{file} is not a regular file'
bot = self._get_bot(server)
bot.dcc_file_transfer(file=file, nick=nick, bind_address=bind_address)
@action
def send_message(
self, text: str, server: Union[str, Tuple[str, int]], target: Union[str, Sequence[str]]
):
"""
Send a message to a channel or a nick.
:param text: Message content.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param target: Message target (nick or channel). If it's a list then the message will be sent
to multiple targets.
"""
bot = self._get_bot(server)
method = (
bot.connection.privmsg if isinstance(target, str)
else bot.connection.privmsg_many
)
method(target, text)
@action
def send_notice(
self, text: str, server: Union[str, Tuple[str, int]], target: str
):
"""
Send a notice to a channel or a nick.
:param text: Message content.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param target: Message target (nick or channel).
"""
bot = self._get_bot(server)
bot.connection.notice(target, text)
@action
def servers(self) -> Sequence[dict]:
"""
Get information about the connected servers.
:return: .. schema:: irc.IRCServerStatusSchema(many=True)
"""
bots = self._bots_by_server.values()
return IRCServerStatusSchema().dump(
{
'server': bot.server,
'port': bot.port,
'alias': bot.alias,
'real_name': bot.connection.get_server_name(),
'nickname': bot.connection.get_nickname(),
'is_connected': bot.connection.is_connected(),
'connected_channels': bot.channels.keys(),
}
for bot in bots
)
@action
def channel(self, server: Union[str, Tuple[str, int]], channel: str) -> dict:
"""
Get information about a connected channel.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param channel:
:return: .. schema:: irc.IRCChannelSchema
"""
bot = self._get_bot(server)
channel_name = channel
channel = bot.channels.get(channel)
assert channel, f'Not connected to channel {channel}'
return IRCChannelSchema().dump({
'is_invite_only': channel.is_invite_only(),
'is_moderated': channel.is_moderated(),
'is_protected': channel.is_protected(),
'is_secret': channel.is_secret(),
'name': channel_name,
'modes': channel.modes,
'opers': list(channel.opers()),
'owners': channel.owners(),
'users': list(channel.users()),
'voiced': list(channel.voiced()),
})
@action
def send_ctcp_message(
self, ctcp_type: str, body: str, server: Union[str, Tuple[str, int]], target: str
):
"""
Send a CTCP message to a target.
:param ctcp_type: CTCP message type.
:param body: Message content.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param target: Message target.
"""
bot = self._get_bot(server)
bot.connection.ctcp(ctcp_type, target, body)
@action
def send_ctcp_reply(
self, body: str, server: Union[str, Tuple[str, int]], target: str
):
"""
Send a CTCP REPLY command.
:param body: Message content.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param target: Message target.
"""
bot = self._get_bot(server)
bot.connection.ctcp_reply(target, body)
@action
def disconnect(self, server: Union[str, Tuple[str, int]], message: Optional[str] = None):
"""
Disconnect from a server.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param message: Disconnect message (default: configured ``stop_message``.
"""
bot = self._get_bot(server)
bot.connection.disconnect(message or bot.stop_message)
@action
def invite(
self, nick: str, channel: str, server: Union[str, Tuple[str, int]]
):
"""
Invite a nick to a channel.
:param nick: Target IRC nick.
:param channel: Target IRC channel.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
"""
bot = self._get_bot(server)
bot.connection.invite(nick, channel)
@action
def join(self, channel: str, server: Union[str, Tuple[str, int]]):
"""
Join a channel.
:param channel: Target IRC channel.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
"""
bot = self._get_bot(server)
bot.connection.join(channel)
@action
def kick(
self, nick: str, channel: str, server: Union[str, Tuple[str, int]], reason: Optional[str] = None
):
"""
Kick a nick from a channel.
:param nick: Target IRC nick.
:param channel: Target IRC channel.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param reason: Kick reason.
"""
bot = self._get_bot(server)
bot.connection.kick(channel, nick, reason)
@action
def mode(
self, target: str, command: str, server: Union[str, Tuple[str, int]]
):
"""
Send a MODE command on the selected target.
:param target: IRC target.
:param command: Mode command.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
"""
bot = self._get_bot(server)
bot.connection.mode(target, command)
@action
def set_nick(self, nick: str, server: Union[str, Tuple[str, int]]):
"""
Set the IRC nick.
:param nick: New IRC nick.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
"""
bot = self._get_bot(server)
bot.connection.nick(nick)
@action
def oper(self, nick: str, password: str, server: Union[str, Tuple[str, int]]):
"""
Send an OPER command.
:param nick: IRC nick.
:param password: Nick password.
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
"""
bot = self._get_bot(server)
bot.connection.oper(nick, password)
@action
def part(
self, channel: Union[str, Sequence[str]], server: Union[str, Tuple[str, int]],
message: Optional[str] = None
):
"""
Parts/exits a channel.
:param channel: IRC channel (or list of channels).
:param server: IRC server, identified either by ``alias`` or ``(server, port)`` tuple.
:param message: Optional part message (default: same as the bot's ``stop_message``).
"""
bot = self._get_bot(server)
channels = [channel] if isinstance(channel, str) else channel
bot.connection.part(channels=channels, message=message or bot.stop_message)
@action
def quit(
self, server: Union[str, Tuple[str, int]], message: Optional[str] = None
):
"""
Send a QUIT command.