aboutsummaryrefslogtreecommitdiffhomepage
path: root/apioforum/webhooks.py
diff options
context:
space:
mode:
authorubq323 <ubq323@ubq323.website>2022-01-11 21:05:23 +0000
committerubq323 <ubq323@ubq323.website>2022-01-12 01:44:16 +0000
commit53361d4c4880e76a6f1d1b3ebe009422dd5438fc (patch)
tree36dee3257229d34d75a1fbae2ba9f6c72a3ba9ad /apioforum/webhooks.py
parent22dbccb40a5444550ef4ead954aa295d51c4eb24 (diff)
refactor webhooks codewebhooks
Diffstat (limited to 'apioforum/webhooks.py')
-rw-r--r--apioforum/webhooks.py241
1 files changed, 130 insertions, 111 deletions
diff --git a/apioforum/webhooks.py b/apioforum/webhooks.py
index 2407039..01f539e 100644
--- a/apioforum/webhooks.py
+++ b/apioforum/webhooks.py
@@ -1,4 +1,5 @@
import urllib
+import abc
import json
from .db import get_db
from flask import url_for, flash
@@ -10,125 +11,143 @@ def abridge_post(text):
else:
return text
-def send_discord_webhook(url,payload):
- headers = {
- "User-Agent":"apioforum (https://f.gh0.pw/apioforum/, v0.0)",
- "Content-Type":"application/json",
- }
- req = urllib.request.Request(
- url,
- json.dumps(payload).encode("utf-8"),
- headers
- )
- # todo: read response and things
- urllib.request.urlopen(req)
- #try:
- # res = urllib.request.urlopen(req)
- #except urllib.error.HTTPError as e:
- # print(f"error {e.code} {e.read()}")
- #else:
- # print(f"succ {res.read()}")
-
-# todo: object oriented thing for different kinds of webhook (not just discord)
-
-def f(name,value):
- # discord embed field
- return {"name":name,"value":value,"inline":True}
+webhook_types = {}
+def webhook_type(t):
+ def inner(cls):
+ webhook_types[t] = cls
+ return cls
+ return inner
-def discord_on_new_thread(wh_url, thread):
- db = get_db()
- forum = db.execute("select * from forums where id = ?",(thread['forum'],)).fetchone()
- username = thread['creator']
- userpage = url_for('user.view_user',username=username,_external=True)
+class WebhookType(abc.ABC):
+ def __init__(self, url):
+ self.url = url
- forumpage = url_for('forum.view_forum',forum_id=forum['id'],_external=True)
+ @abc.abstractmethod
+ def on_new_thread(self,thread):
+ pass
+ @abc.abstractmethod
+ def on_new_post(self,post):
+ pass
- post = db.execute("select * from posts where thread = ? order by id asc limit 1",(thread['id'],)).fetchone()
-
-
- payload = {
- "username":"apioforum",
- "avatar_url":"https://d.gh0.pw/lib/exe/fetch.php?media=wiki:logo.png",
- "embeds":[
- {
- "title":thread['title'],
- "description":abridge_post(post['content']),
- "url": url_for('thread.view_thread',thread_id=thread['id'],_external=True),
- "color": 0xff00ff,
- "fields":[
- f('author',f"[{username}]({userpage})"),
- f('forum',f"[{forum['name']}]({forumpage})"),
- ],
- "footer":{
- "text":thread['created'].isoformat(' '),
- },
- },
- ],
- }
- send_discord_webhook(wh_url,payload)
-
-def discord_on_new_post(wh_url, post):
- from .thread import post_jump
+def get_webhooks(forum_id):
db = get_db()
+ # todo inheritance (if needed)
+ webhooks = db.execute("select * from webhooks where webhooks.forum = ?;",(forum_id,)).fetchall()
- thread = db.execute("select * from threads where id = ?",(post['thread'],)).fetchone()
- threadpage = url_for('thread.view_thread',thread_id=thread['id'],_external=True)
-
- forum = db.execute("select * from forums where id = ?",(thread['forum'],)).fetchone()
- forumpage = url_for('forum.view_forum',forum_id=forum['id'],_external=True)
-
- username = post['author']
- userpage = url_for('user.view_user',username=username,_external=True)
-
- payload = {
- "username":"apioforum",
- "avatar_url":"https://d.gh0.pw/lib/exe/fetch.php?media=wiki:logo.png",
- "embeds":[
- {
- "title":"re: "+thread['title'],
- "description":abridge_post(post['content']),
- "url": post_jump(post['id'],external=True),
- "color": 0x00ffff,
- "fields":[
- f('author',f"[{username}]({userpage})"),
- f('thread',f"[{thread['title']}]({threadpage})"),
- f('forum',f"[{forum['name']}]({forumpage})"),
- ],
- "footer":{
- "text":post['created'].isoformat(' '),
- },
- },
- ],
- }
- send_discord_webhook(wh_url,payload)
-
-def _do_webhooks(forum_id,thing,fn):
- db = get_db()
- # todo inheritance
- webhooks = db.execute("""
- WITH RECURSIVE fs AS
- (SELECT * FROM forums WHERE id = ?
- UNION ALL
- SELECT forums.* FROM forums, fs WHERE fs.parent=forums.id)
- SELECT * from webhooks
- WHERE
- webhooks.forum = ?
- OR
- webhooks.inherits AND webhooks.forum IN (SELECT id FROM fS);
-
- """,(forum_id,forum_id)).fetchall()
for wh in webhooks:
+ wh_type = wh['type']
+ if wh_type not in webhook_types:
+ print(f"unknown webhook type {wh_type}")
+ continue
wh_url = wh['url']
- try:
- fn(wh_url,thing)
- except:
- pass # handle probably
- flash(f"wh {wh['id']}")
+ wo = webhook_types[wh_type](wh_url)
+ yield wo
def do_webhooks_thread(forum_id,thread):
- _do_webhooks(forum_id,thread,discord_on_new_thread)
-
+ for wh in get_webhooks(forum_id):
+ wh.on_new_thread(thread)
def do_webhooks_post(forum_id,post):
- _do_webhooks(forum_id,post,discord_on_new_post)
-
+ for wh in get_webhooks(forum_id):
+ wh.on_new_post(post)
+
+@webhook_type("fake")
+class FakeWebhook(WebhookType):
+ def on_new_post(self, post):
+ print(f'fake wh {self.url} post {post["id"]}')
+ def on_new_thread(self, thread):
+ print(f'fake wh {self.url} thread {thread["id"]}')
+
+@webhook_type("discord")
+class DiscordWebhook(WebhookType):
+ def send(self,payload):
+ headers = {
+ "User-Agent":"apioforum (https://g.gh0.pw/apioforum, v0.0)",
+ "Content-Type":"application/json",
+ }
+ req = urllib.request.Request(
+ self.url,
+ json.dumps(payload).encode("utf-8"),
+ headers
+ )
+ # todo: read response and things
+ urllib.request.urlopen(req)
+ #try:
+ # res = urllib.request.urlopen(req)
+ #except urllib.error.HTTPError as e:
+ # print(f"error {e.code} {e.read()}")
+ #else:
+ # print(f"succ {res.read()}")
+
+ @staticmethod
+ def field(name,value):
+ return {"name":name,"value":value,"inline":True}
+
+ def on_new_thread(self,thread):
+ f = self.field
+ db = get_db()
+ forum = db.execute("select * from forums where id = ?",(thread['forum'],)).fetchone()
+ username = thread['creator']
+ userpage = url_for('user.view_user',username=username,_external=True)
+
+ forumpage = url_for('forum.view_forum',forum_id=forum['id'],_external=True)
+
+ post = db.execute("select * from posts where thread = ? order by id asc limit 1",(thread['id'],)).fetchone()
+
+ payload = {
+ "username":"apioforum",
+ "avatar_url":"https://d.gh0.pw/lib/exe/fetch.php?media=wiki:logo.png",
+ "embeds":[
+ {
+ "title":"new thread: "+thread['title'],
+ "description":abridge_post(post['content']),
+ "url": url_for('thread.view_thread',thread_id=thread['id'],_external=True),
+ "color": 0xff00ff,
+ "fields":[
+ f('author',f"[{username}]({userpage})"),
+ f('forum',f"[{forum['name']}]({forumpage})"),
+ ],
+ "footer":{
+ "text":thread['created'].isoformat(' '),
+ },
+ },
+ ],
+ }
+ self.send(payload)
+
+ def on_new_post(self,post):
+ from .thread import post_jump
+ f = self.field
+ db = get_db()
+
+ thread = db.execute("select * from threads where id = ?",(post['thread'],)).fetchone()
+ threadpage = url_for('thread.view_thread',thread_id=thread['id'],_external=True)
+
+ forum = db.execute("select * from forums where id = ?",(thread['forum'],)).fetchone()
+ forumpage = url_for('forum.view_forum',forum_id=forum['id'],_external=True)
+
+ username = post['author']
+ userpage = url_for('user.view_user',username=username,_external=True)
+
+ payload = {
+ "username":"apioforum",
+ "avatar_url":"https://d.gh0.pw/lib/exe/fetch.php?media=wiki:logo.png",
+ "embeds":[
+ {
+ "title":"re: "+thread['title'],
+ "description":abridge_post(post['content']),
+ "url": post_jump(post['id'],external=True),
+ "color": 0x00ffff,
+ "fields":[
+ f('author',f"[{username}]({userpage})"),
+ f('thread',f"[{thread['title']}]({threadpage})"),
+ f('forum',f"[{forum['name']}]({forumpage})"),
+ ],
+ "footer":{
+ "text":post['created'].isoformat(' '),
+ },
+ },
+ ],
+ }
+ self.send(payload)
+