mostr-zulip-bot/mostr.py

from typing import Dict

from pynostr.event import Event
from pynostr.relay_manager import RelayManager
from pynostr.key import PrivateKey
from pynostr.key import PublicKey

from zulip_bots.lib import AbstractBotHandler

def parse_pubkey(key_str: str) -> PublicKey:
    try:
        return PublicKey.from_hex(key_str)
    except (ValueError, TypeError):
        pass  # try to load as npub if hex fails

    try:
        return PublicKey.from_npub(key_str)
    except (ValueError, TypeError):
        return None


class MostrHandler:
    """A Zulip bot for Mostr"""

    def usage(self) -> str:
        return "Send me a message to create a task!"

    def initialize(self, bot_handler: AbstractBotHandler) -> None:
        self.config_info = bot_handler.get_config_info('mostr')
        self.relay = self.config_info['relay']
        self.relays = RelayManager()
        self.relays.add_relay(self.relay)
        self.key = PrivateKey.from_nsec(self.config_info['key']).hex()
        print('Connecting to relay', self.relay)

    # TODO React to all group messages
    def handle_message(self, message: Dict[str, str], bot_handler: AbstractBotHandler) -> None:
        username = message.get('sender_full_name')
        print('Received', message, 'from', username)

        # Tags
        if message.get('type') == 'private' and len(message.get('display_recipient')) < 3:
            context = f"s{message.get('sender_id')}"
        else:
            context = f"r{message.get('recipient_id')}"
        try:
            hashtags = list(filter(None, bot_handler.storage.get(context).split("#")))
        except KeyError:
            hashtags = []

        print('Context', context, 'contains', hashtags)

        if message['content'].startswith("#"):
            hashtags.extend(filter(None, message['content'].split('#')))
            bot_handler.send_reply(message, f"Active Hashtags: {hashtags}")
            if hashtags:
                result = bot_handler.storage.put(context, '#'.join(set(hashtags)))
            return

        if message['content'].startswith("-"):
            to_remove = message['content'].replace('-', '#').split('#')
            hashtags = [t for t in hashtags if t not in to_remove]
            bot_handler.send_reply(message, f"Active Hashtags: {hashtags}")
            bot_handler.storage.put(context, '#'.join(hashtags))
            return

        # Initial Feedback
        if len(message['content']) < 4:
            if message.get('type') == 'private':
                bot_handler.send_reply(message, "That task name seems a bit short!")
            else:
                bot_handler.react(message, 'question')
            return
        bot_handler.react(message, 'working_on_it')

        # Assignee Public Key
        pubkey = None
        user = message.get('sender_id') or message.get('sender_email')
        if user:
            # TODO Need proper verification, which is not done by pynostr
            # A public key is generally also a valid private key, apparently
            # try:
            #    privkey = PrivateKey.from_hex(message['content'])
            #    bot_handler.send_reply(message, f"Whoa that looks like a private key! Please delete it swiftly. This is the corresponding private key: {privkey.public_key}")
            #    return
            # except:
            #    pass

            # Parse pubkey from storage or message
            key = f"pubkey_{user}"
            if message['content']:
                pubkey = parse_pubkey(message['content'])
            if pubkey:
                bot_handler.storage.put(key, pubkey.npub)
                bot_handler.send_reply(message, "New pubkey set!")
                return
            try:
                pubkey = PublicKey.from_npub(bot_handler.storage.get(key))
            except (KeyError, IndexError):
                if message.get('type') == 'private':
                    bot_handler.send_reply(message, "Please send me your public key for proper attribution")

        # auto-assign via username matching, extract hashtags

        tags = list(map(lambda tag: ['t', tag], hashtags or ['zulip']))
        if pubkey:
            tags.append(['p', pubkey.hex()])
        event = Event(content=message['content'], kind=1621, tags=tags)
        event.sign(self.key)

        print('Publishing', event)
        self.relays.publish_event(event)
        self.relays.run_sync()

        if message.get('type') == 'private':
            bot_handler.send_reply(message, f"Sent `{event}` to {self.relay}")
        else:
            if hasattr(bot_handler, '_client'):
                bot_handler._client.remove_reaction({'message_id': message['id'], 'emoji_name': 'working_on_it'})
            bot_handler.react(message, 'check')

    def __del__(self):
        if hasattr(self, 'relays'):
            self.relays.close_connections()
            print('Terminated Mostr Bot')

handler_class = MostrHandler