101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
|
|
# Copyright (C) 2020 by Cathy Hu <cathy.hu@fau.de>
|
|
# Copyright (C) 2020 by Martin Rey <martin.rey@mailbox.org>
|
|
#
|
|
# SPDX-License-Identifier: 0BSD
|
|
|
|
from asyncio import get_event_loop, sleep
|
|
from kibicara.platformapi import Censor, Spawner, Message
|
|
from kibicara.platforms.mastodon.model import MastodonAccount
|
|
from logging import getLogger
|
|
|
|
from mastodon import Mastodon, MastodonError
|
|
from asyncio import gather
|
|
import re
|
|
|
|
logger = getLogger(__name__)
|
|
|
|
|
|
class MastodonBot(Censor):
|
|
def __init__(self, mastodon_account_model):
|
|
super().__init__(mastodon_account_model.hood)
|
|
self.model = mastodon_account_model
|
|
self.enabled = self.model.enabled
|
|
self.polling_interval_sec = 60
|
|
|
|
@classmethod
|
|
async def destroy_hood(cls, hood):
|
|
"""Removes all its database entries."""
|
|
for mastodon in await MastodonAccount.objects.filter(hood=hood).all():
|
|
await mastodon.delete()
|
|
|
|
async def run(self):
|
|
try:
|
|
await self.model.instance.load()
|
|
self.account = Mastodon(
|
|
client_id=self.model.instance.client_id,
|
|
client_secret=self.model.instance.client_secret,
|
|
api_base_url=self.model.instance.name,
|
|
access_token=self.model.access_token,
|
|
)
|
|
account_details = await get_event_loop().run_in_executor(
|
|
None, self.account.account_verify_credentials
|
|
)
|
|
if username := account_details.get("username"):
|
|
await self.model.update(username=username)
|
|
await gather(self.poll(), self.push())
|
|
except Exception as e:
|
|
logger.debug("Bot {0} threw an Error: {1}".format(self.model.hood.name, e))
|
|
finally:
|
|
logger.debug("Bot {0} stopped.".format(self.model.hood.name))
|
|
|
|
async def poll(self):
|
|
"""Get new mentions and DMs from Mastodon"""
|
|
while True:
|
|
try:
|
|
notifications = await get_event_loop().run_in_executor(
|
|
None, self.account.notifications
|
|
)
|
|
except MastodonError as e:
|
|
logger.warning("%s in hood %s" % (e, self.model.hood.name))
|
|
continue
|
|
for status in notifications:
|
|
try:
|
|
status_id = int(status["status"]["id"])
|
|
except KeyError:
|
|
continue # ignore notifications which don't have a status
|
|
text = re.sub(r"<[^>]*>", "", status["status"]["content"])
|
|
text = re.sub(
|
|
"(?<=^|(?<=[^a-zA-Z0-9-_.]))@([A-Za-z]+[A-Za-z0-9-_]+)", "", text
|
|
)
|
|
logger.debug(
|
|
"Mastodon in %s received toot #%s: %s"
|
|
% (self.model.hood.name, status_id, text)
|
|
)
|
|
if status["status"]["visibility"] == "public":
|
|
await self.publish(Message(text, toot_id=status_id))
|
|
else:
|
|
await self.publish(Message(text))
|
|
await get_event_loop().run_in_executor(
|
|
None, self.account.notifications_dismiss, status["id"]
|
|
)
|
|
await sleep(self.polling_interval_sec)
|
|
|
|
async def push(self):
|
|
"""Push new Ticketfrei reports to Mastodon; if source is mastodon, boost it."""
|
|
while True:
|
|
message = await self.receive()
|
|
if hasattr(message, "tood_id"):
|
|
logger.debug("Boosting post %s: %s" % (message.tood_id, message.text))
|
|
await get_event_loop().run_in_executor(
|
|
None, self.account.status_reblog, message.tood_id
|
|
)
|
|
else:
|
|
logger.debug("Posting message: %s" % (message.text,))
|
|
await get_event_loop().run_in_executor(
|
|
None, self.account.status_post, message.text
|
|
)
|
|
|
|
|
|
spawner = Spawner(MastodonAccount, MastodonBot)
|