71 lines
2.6 KiB
Python
71 lines
2.6 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 kibicara.platformapi import Censor, Spawner, Message
|
|
from kibicara.platforms.mastodon.model import MastodonAccount
|
|
|
|
from mastodon import Mastodon, MastodonError
|
|
from asyncio import gather
|
|
import sys
|
|
|
|
from logging import getLogger
|
|
|
|
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.account = Mastodon(
|
|
client_id=self.model.instance.client_id,
|
|
client_secret=self.model.instance.client_secret,
|
|
access_token=self.model.access_token,
|
|
)
|
|
|
|
async def run(self):
|
|
await gather(self.poll(), self.push())
|
|
|
|
async def poll(self):
|
|
"""Get new mentions and DMs from Mastodon"""
|
|
while True:
|
|
try:
|
|
notifications = self.account.notifications()
|
|
except MastodonError:
|
|
logger.warning(
|
|
"%s in hood %s" % (sys.exc_info()[0], self.model.hood.name)
|
|
)
|
|
continue
|
|
last_seen = int(self.model.last_seen)
|
|
for status in notifications:
|
|
status_id = int(status['status']['id'])
|
|
if status_id <= last_seen:
|
|
continue # toot was already processed in the past
|
|
if status_id > self.model.last_seen:
|
|
self.model.last_seen = status_id # save last_seen in database
|
|
text = status['status']['content']
|
|
# :TODO sanitize toot content; see ticketfrei2 for regex magic
|
|
logger.debug(
|
|
"Mastodon in %s received message: " % (self.model.hood.name,)
|
|
)
|
|
if status['status']['visibility'] == 'public':
|
|
await self.publish(Message(text, toot_id=status_id))
|
|
else:
|
|
await self.publish(Message(text))
|
|
|
|
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"):
|
|
await self.account.status_reblog(message.tood_id)
|
|
else:
|
|
await self.account.status_post(message.text)
|
|
|
|
|
|
spawner = Spawner(MastodonAccount, MastodonBot)
|