1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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
|
