nonsensebot/app/modules/fediverse_preview.py

"""
This module takes fediverse links (e.g.) and provides a preview in a reply.

By default, this will only work with fedi.transgender.ing, but if you set `[fediverse_preview.urls]` in your config,
you can add more. This module will use the Mastodon protocol, however it is designed to be used with misskey.
"""

import logging
import niobot
import httpx
import typing
import textwrap
from urllib.parse import urlparse

if typing.TYPE_CHECKING:
    from ..main import NonsenseBot


class FediversePreviewModule(niobot.Module):
    bot: "NonsenseBot"
    log = logging.getLogger(__name__)

    @niobot.event("message")
    async def on_message(self, room: niobot.MatrixRoom, event: niobot.RoomMessage):
        config = self.bot.cfg.get("fediverse_preview", {})
        supported_prefixes = config.get("urls", ["https://fedi.transgender.ing"])
        ignore = config.get("ignore", [])
        if not isinstance(event, niobot.RoomMessageText):
            return
        if (
            event.sender in ignore
            or room.room_id in ignore
            or event.sender == self.bot.user_id
        ):
            return

        sent = []
        to_get: set[str] = set()
        for item in event.body.split():
            if not item.startswith(tuple(supported_prefixes)):
                return
            parsed = urlparse(item)
            post_id = parsed.path.split("/")[-1]
            if post_id in sent:
                self.log.info("Already sent post %s", post_id)
                continue
            elif len(to_get) >= 5:
                self.log.info("Already sent 5 posts, stopping")
                break
            to_get.add("https://%s/api/v1/statuses/%s" % (parsed.netloc, post_id))
        if to_get:
            async with httpx.AsyncClient(
                    headers={
                        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:129.0) Gecko/20100101 Firefox/129.0"
                    }
            ) as client:
                for url in to_get:
                    resp = await client.get(
                        url
                    )
                    if resp.status_code != 200:
                        self.log.error("Got HTTP %d from %s", resp.status_code, resp.url)
                        continue
                    data = resp.json()
                    self.log.info("Got data: %r", data)
                    username = data["account"]["username"]
                    if not (text := data.get("content")):
                        self.log.warning("No text for post %s", url)
                        continue

                    self.log.info("Detected fediverse post %s: %r", url, text)
                    rendered = await self.bot._markdown_to_html(text)
                    parsed = urlparse(url)
                    text_body = "<blockquote>%s</blockquote>" % rendered
                    body = '<a href="%s">@%s:</a><br>%s' % (
                        "https://%s/@%s" % (parsed.netloc, username),
                        username,
                        text_body,
                    )
                    if data.get("media_attachments"):
                        body += "<p>{:,} attachments</p>".format(
                            len(data["media_attachments"])
                        )
                    self.log.info("Sending fediverse post %s", url)
                    await self.bot.send_message(
                        room,
                        body,
                        reply_to=event,
                        content_type="html.raw",
                        override={"body": f"@{username}: {data['content']!r}"},
                    )
                    sent.append(post_id)