nonsensebot/app/modules/misskey_preview.py
2024-08-11 15:36:07 +01:00

67 lines
2.7 KiB
Python

"""
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"]["username"]
if not (text := data.get("content")):
self.log.warning("No text for post %s", post_id)
continue
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['content']!r}"}
)
sent.append(post_id)