nxy/bot/plugins/ctcp.py

# -*- coding: utf-8 -*-
import time

import irc3
from docopt import Dict as DocOptDict
from irc3.plugins.command import command
from irc3.utils import IrcString

from . import Plugin


@irc3.plugin
class CTCP(Plugin):
    requires = ['irc3.plugins.async',
                'irc3.plugins.command']

    # noinspection PyMethodMayBeStatic
    def _ctcp(self, name: str, nick: str, reply: str):
        return '\x02[{}]\x0F {}: {}'.format(name.upper(), nick, reply)

    async def ctcp(self, name: str, mask: IrcString, args: DocOptDict):
        nick = args.get('<nick>') or mask.nick
        name = name.upper()
        data = await self.bot.ctcp_async(nick, name)

        if not data or data['timeout']:
            reply = 'timeout'
        elif not data['success']:
            reply = 'Error: {}'.format(data['reply'])
        else:
            reply = data['reply']

        return self._ctcp(name, nick, reply)

    @command
    async def ping(self, mask: IrcString, target: IrcString,
                   args: DocOptDict):
        """Sends ping via CTCP to user and sends the time needed

        %%ping [<nick>]
        """
        nick = args.get('<nick>') or mask.nick
        data = await self.bot.ctcp_async(nick, 'PING {}'.format(time.time()))

        if not data or data['timeout']:
            reply = 'timeout'
        elif not data['success']:
            reply = 'Error: {}'.format(data['reply'])
        else:
            delta = time.time() - float(data['reply'])
            if delta < 1.0:
                delta *= 1000
                unit = 'ms'
            else:
                unit = 's'
            reply = '{0:.3f} {1}'.format(delta, unit)

        return self._ctcp('PING', nick, reply)

    @command
    async def finger(self, mask: IrcString, target: IrcString,
                     args: DocOptDict):
        """Gets the client response for finger nick user via CTCP

        %%finger [<nick>]
        """
        return await self.ctcp('FINGER', mask, args)

    @command
    async def time(self, mask: IrcString, target: IrcString,
                   args: DocOptDict):
        """Gets the client time from nick via CTCP

        %%time [<nick>]
        """
        return await self.ctcp('TIME', mask, args)

    @command
    async def ver(self, mask: IrcString, target: IrcString, args: DocOptDict):
        """Gets the client version from nick via CTCP

        %%ver [<nick>]
        """
        return await self.ctcp('VERSION', mask, args)