pymicrocosm/src/microcosm/webmentions.py

import json
import base64
import giteapy
import os

import clientapi_forgejo as forgejo

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 forgejo.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.b64encode(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 = forgejo.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