92 lines
3.6 KiB
Python
92 lines
3.6 KiB
Python
"""
|
|
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)
|