nonsensebot/app/modules/misskey_preview.py

"""
This module takes misskey links (e.g.) and provides a preview in a reply.
"""
import logging
import niobot
import httpx
import typing
import textwrap
from urllib.parse import urlparse
if typing.TYPE_CHECKING:
    from ..main import TortoiseIntegratedBot


class MisskeyPreviewModule(niobot.Module):
    bot: "TortoiseIntegratedBot"
    log = logging.getLogger(__name__)

    @niobot.event("message")
    async def on_message(self, room: niobot.MatrixRoom, event: niobot.RoomMessage):
        supported_prefixes = self.bot.cfg.get("misskey_preview", {})
        supported_prefixes = supported_prefixes.get("urls", ["https://fedi.transgender.ing"])
        if not isinstance(event, niobot.RoomMessageText):
            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"]["fqn"]
                if not data.get("text"):
                    self.log.warning("No text for post %s", post_id)
                    continue

                text = textwrap.shorten(data["text"], width=1000)
                self.log.info("Detected misskey 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,
                )
                self.log.info("Sending misskey post %s", post_id)
                await self.bot.send_message(
                    room,
                    body,
                    reply_to=event,
                    content_type="html.raw",
                    override={"body": f"@{username}: {data['text']!r}"}
                )
                sent.append(post_id)