aerc/contrib/ircbot/Sourcehut/plugin.py

import email.header
import email.utils
import json
import mailbox
from urllib.parse import quote
from urllib.request import urlopen
import re
import traceback

from supybot import callbacks, httpserver, ircmsgs, world
from supybot.ircutils import bold, italic, mircColor, underline


class Sourcehut(callbacks.Plugin):
    """
    Supybot plugin to receive Sourcehut webhooks
    """

    def __init__(self, irc):
        super().__init__(irc)
        httpserver.hook("sourcehut", SourcehutServerCallback(self))

    def die(self):
        httpserver.unhook("sourcehut")
        super().die()

    def announce(self, channel, message):
        libera = world.getIrc("libera")
        if libera is None:
            print("error: no irc libera")
            return
        if channel not in libera.state.channels:
            print(f"error: not in {channel} channel")
            return
        libera.sendMsg(ircmsgs.notice(channel, message))


def decode_header(header: str) -> str:
    if not header:
        return ""
    text = ""
    for chunk, encoding in email.header.decode_header(header):
        if isinstance(chunk, bytes):
            chunk = chunk.decode(encoding or "us-ascii")
        text += chunk
    return text


class SourcehutServerCallback(httpserver.SupyHTTPServerCallback):
    name = "Sourcehut"
    defaultResponse = "Bad request\n"

    def __init__(self, plugin: Sourcehut):
        super().__init__()
        self.plugin = plugin

    SUBJECT = "[PATCH {prefix} v{version}] {subject}"
    URL = "https://lists.sr.ht/{list[owner][canonicalName]}/{list[name]}"
    CHANS = {
        "#public-inbox": "##rjarry",
        "#aerc-devel": "#aerc",
    }

    def announce_patch(self, patchset):
        subject = self.SUBJECT.format(**patchset)
        url = self.URL.format(**patchset)
        if not url.startswith("https://lists.sr.ht/~rjarry/"):
            raise ValueError("unknown list")
        url += "/patches/{id}".format(**patchset)
        channel = f"#{patchset['list']['name']}"
        channel = self.CHANS.get(channel, channel)
        try:
            submitter = patchset["submitter"]["canonicalName"]
        except KeyError:
            try:
                submitter = patchset["submitter"]["name"]
            except KeyError:
                submitter = patchset["submitter"]["address"]
        msg = f"{mircColor('received', 'light gray')} {bold(subject)}"
        msg += f" from {italic(submitter)}: {underline(url)}"
        self.plugin.announce(channel, msg)

    def announce_apply(self, mail):
        channel = f"#{mail['list']['name']}"
        channel = self.CHANS.get(channel, channel)
        refs = []
        for header in mail['references']:
            refs += header.split()
        for ref in refs:
            url = self.URL.format(**mail) + quote(f"/{ref}")
            print(f"GET {url}/raw")
            with urlopen(f"{url}/raw") as u:
                msg = mailbox.Message(u.read())
            subject = re.sub(r"\s+", " ", decode_header(msg["subject"]))
            if not subject.startswith("[PATCH"):
                continue
            for name, addr in email.utils.getaddresses([decode_header(msg["from"])]):
                if name:
                    submitter = name
                else:
                    submitter = addr
                msg = f"{bold(mircColor('applied', 'green'))} {bold(subject)}"
                msg += f" from {italic(submitter)}: {underline(url)}"
                self.plugin.announce(channel, msg)
                return

    def doPost(self, handler, path, form=None):
        if hasattr(form, "decode"):
            form = form.decode("utf-8")
        print(f"POST {path} {form}")
        try:
            body = json.loads(form)
            hook = body["data"]["webhook"]
            if hook["event"] == "PATCHSET_RECEIVED":
                self.announce_patch(hook["patchset"])
                handler.send_response(200)
                handler.end_headers()
                handler.wfile.write(b"")
                return

            if hook["event"] == "EMAIL_RECEIVED":
                if hook["email"]["patchset_update"] == ["APPLIED"]:
                    self.announce_apply(hook["email"])
                handler.send_response(200)
                handler.end_headers()
                handler.wfile.write(b"")
                return

            raise ValueError(f"unsupported webhook: {hook}")

        except Exception as e:
            traceback.print_exception(e)
            handler.send_response(400)
            handler.end_headers()
            handler.wfile.write(b"Bad request\n")

    def log_message(self, format, *args):
        pass


Class = Sourcehut