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 = []
        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 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(sent) >= 5:
                    self.log.info("Already sent 5 posts, stopping")
                    break
                resp = await client.get(
                    "https://%s/api/v1/statuses/%s" % (parsed.netloc, post_id)
                )
                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", post_id)
                    continue

                self.log.info("Detected fediverse post %s: %r", post_id, text)
                rendered = await self.bot._markdown_to_html(text)
                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", post_id)
                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)