122 lines
3.5 KiB
Python
122 lines
3.5 KiB
Python
|
import json
|
||
|
|
import base64
|
||
|
|
import giteapy
|
||
|
|
import os
|
||
|
|
|
||
|
|
from flask import request, Blueprint
|
||
|
|
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
webhook_bp = Blueprint("webmention", __name__)
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
@webhook_bp.route("/webmentions", methods=['POST'])
|
||
|
|
def webmention_hook():
|
||
|
|
"""Accept web mention webhook request"""
|
||
|
|
|
||
|
|
|
||
|
|
body = request.get_json()
|
||
|
|
|
||
|
|
print(f"Incoming webmention {body}")
|
||
|
|
|
||
|
|
# webmention should always have a json body
|
||
|
|
if body is None:
|
||
|
|
return {"error":"invalid_request"}, 400
|
||
|
|
|
||
|
|
# assert that secret matches
|
||
|
|
if body.get('secret') != os.environ.get("WEBMENTION_SECRET", "changeme"):
|
||
|
|
return {"error":"invalid_secret"}, 401
|
||
|
|
|
||
|
|
# get existing mentions from gitea
|
||
|
|
from microcosm import get_api_client
|
||
|
|
|
||
|
|
api = get_api_client()
|
||
|
|
|
||
|
|
old_sha = None
|
||
|
|
|
||
|
|
|
||
|
|
try:
|
||
|
|
mentions_meta = api.repo_get_contents(os.environ.get('GITEA_REPO_OWNER'),
|
||
|
|
os.environ.get('GITEA_REPO_NAME'), os.environ.get('WEBMENTIONS_JSON_FILE'))
|
||
|
|
|
||
|
|
old_sha = mentions_meta.sha
|
||
|
|
|
||
|
|
mentions = json.loads(base64.decodebytes(mentions_meta.content.encode('utf8')))
|
||
|
|
|
||
|
|
except giteapy.rest.ApiException as e:
|
||
|
|
if e.status == 404:
|
||
|
|
print("no mentions yet, create new mentions file")
|
||
|
|
mentions = {}
|
||
|
|
|
||
|
|
# parse target url and get path on server
|
||
|
|
url_path = urlparse(body.get('target')).path
|
||
|
|
|
||
|
|
# create mention entry if needed
|
||
|
|
if url_path not in mentions:
|
||
|
|
mentions[url_path] = []
|
||
|
|
|
||
|
|
# if the mention is already processed, do nothing.
|
||
|
|
for entry in mentions[url_path]:
|
||
|
|
if entry['source'] == body['source']:
|
||
|
|
return {"message": "mention already processed, no action taken."}
|
||
|
|
|
||
|
|
activity_types = {
|
||
|
|
"in-reply-to": "reply",
|
||
|
|
"like-of": "like",
|
||
|
|
"repost-of": "repost",
|
||
|
|
"bookmark-of": "bookmark",
|
||
|
|
"mention-of": "mention",
|
||
|
|
"rsvp":"rsvp"
|
||
|
|
}
|
||
|
|
|
||
|
|
# format the mention like the data from webmention.io api
|
||
|
|
new_mention = {
|
||
|
|
"id": body['post']['wm-id'],
|
||
|
|
"source": body['source'],
|
||
|
|
"target": body['target'],
|
||
|
|
"activity":{
|
||
|
|
"type": activity_types[body['post']['wm-property']]
|
||
|
|
},
|
||
|
|
"verified_date": body.get('wm-received', datetime.now().isoformat()),
|
||
|
|
"data":{
|
||
|
|
"author": body['post']['author'],
|
||
|
|
"content": body['post'].get('content',{}).get('html', None),
|
||
|
|
"published": body['post'].get('published')
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
# append the new mention
|
||
|
|
mentions[url_path].append(new_mention)
|
||
|
|
|
||
|
|
content = base64.encodestring(json.dumps(mentions, indent=2).encode("utf8")).decode("utf8")
|
||
|
|
|
||
|
|
# store to repo
|
||
|
|
if old_sha:
|
||
|
|
print(f"Update {os.environ.get('WEBMENTIONS_JSON_FILE')}")
|
||
|
|
body = giteapy.UpdateFileOptions(content=content, sha=old_sha)
|
||
|
|
|
||
|
|
|
||
|
|
try:
|
||
|
|
r = api.repo_update_file(os.environ.get(
|
||
|
|
'GITEA_REPO_OWNER'), os.environ.get('GITEA_REPO_NAME'), os.environ.get('WEBMENTIONS_JSON_FILE'), body)
|
||
|
|
|
||
|
|
return {"message":"ok"}
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
return {"error": str(e)}, 500
|
||
|
|
|
||
|
|
else:
|
||
|
|
print(f"Create {os.environ.get('WEBMENTIONS_JSON_FILE')}")
|
||
|
|
body = giteapy.CreateFileOptions(content=content)
|
||
|
|
|
||
|
|
try:
|
||
|
|
r = api.repo_create_file(os.environ.get(
|
||
|
|
'GITEA_REPO_OWNER'), os.environ.get('GITEA_REPO_NAME'), os.environ.get('WEBMENTIONS_JSON_FILE'), body)
|
||
|
|
|
||
|
|
return {"message":"ok"}
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
return {"error": str(e)}, 500
|