init: pristine aerc 0.20.0 source

contrib/ircbot/Sourcehut/plugin.py

@@ -0,0 +1,141 @@
+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