ticketfrei3/kibicara/platforms/twitter/bot.py

147 lines
5.8 KiB
Python
Raw Normal View History

2020-07-05 18:34:41 +00:00
# Copyright (C) 2020 by Cathy Hu <cathy.hu@fau.de>
#
# SPDX-License-Identifier: 0BSD
2020-07-17 21:48:16 +00:00
from asyncio import gather, sleep, CancelledError
2020-07-05 18:34:41 +00:00
from kibicara.config import config
from kibicara.platformapi import Censor, Message, Spawner
from kibicara.platforms.twitter.model import Twitter
from logging import getLogger
from peony import PeonyClient, exceptions
2020-07-05 18:34:41 +00:00
logger = getLogger(__name__)
class TwitterBot(Censor):
def __init__(self, twitter_model):
super().__init__(twitter_model.hood)
self.twitter_model = twitter_model
self.enabled = self.twitter_model.enabled
2020-07-05 18:34:41 +00:00
self.polling_interval_sec = 60
2020-07-06 09:11:08 +00:00
self.mentions_since_id = self.twitter_model.mentions_since_id
self.dms_since_id = self.twitter_model.dms_since_id
2020-07-05 18:34:41 +00:00
async def run(self):
try:
if not self.twitter_model.verified:
raise ValueError('Oauth Handshake not completed')
self.client = PeonyClient(
consumer_key=config['twitter']['consumer_key'],
consumer_secret=config['twitter']['consumer_secret'],
access_token=self.twitter_model.access_token,
access_token_secret=self.twitter_model.access_token_secret,
)
if self.twitter_model.mentions_since_id is None:
logger.debug('since_id is None in model, fetch newest mention id')
await self._poll_mentions()
if self.twitter_model.dms_since_id is None:
logger.debug('since_id is None in model, fetch newest dm id')
await self._poll_direct_messages()
logger.debug('Starting Twitter bot: %s' % self.twitter_model.__dict__)
await gather(self.poll(), self.push())
except CancelledError:
logger.debug(f'Bot {self.twitter_model.hood.name} received Cancellation.')
except exceptions.Unauthorized:
logger.debug(f'Bot {self.twitter_model.hood.name} has invalid auth token.')
await self.twitter_model.update(enabled=False)
self.enabled = self.twitter_model.enabled
except (KeyError, ValueError, exceptions.NotAuthenticated):
logger.warning('Missing consumer_keys for Twitter in your configuration.')
await self.twitter_model.update(enabled=False)
self.enabled = self.twitter_model.enabled
finally:
logger.debug(f'Bot {self.twitter_model.hood.name} stopped.')
2020-07-05 18:34:41 +00:00
async def poll(self):
while True:
dms = await self._poll_direct_messages()
logger.debug(
'Polled dms (%s): %s' % (self.twitter_model.hood.name, str(dms))
)
mentions = await self._poll_mentions()
logger.debug(
'Polled mentions (%s): %s'
% (self.twitter_model.hood.name, str(mentions))
)
2020-07-05 18:34:41 +00:00
await self.twitter_model.update(
dms_since_id=self.dms_since_id, mentions_since_id=self.mentions_since_id
)
for message in dms:
2020-07-05 18:34:41 +00:00
await self.publish(Message(message))
for message_id, message in mentions:
await self.publish(Message(message, twitter_mention_id=message_id))
2020-07-05 18:34:41 +00:00
await sleep(self.polling_interval_sec)
async def _poll_direct_messages(self):
dms = await self.client.api.direct_messages.events.list.get()
dms = dms.events
# TODO check for next_cursor (see twitter api)
dms_filtered = []
if dms:
for dm in dms:
2020-07-06 09:11:08 +00:00
if int(dm.id) == self.dms_since_id:
2020-07-05 18:34:41 +00:00
break
dms_filtered.append(dm)
2020-07-06 09:11:08 +00:00
self.dms_since_id = int(dms[0].id)
2020-07-05 18:34:41 +00:00
messages = []
for dm in dms_filtered:
filtered_text = await self._filter_text(
dm.message_create.message_data.entities,
dm.message_create.message_data.text,
)
if not filtered_text:
continue
messages.append(filtered_text)
return messages
async def _poll_mentions(self):
mentions = await self.client.api.statuses.mentions_timeline.get(
since_id=self.mentions_since_id
)
if mentions:
self.mentions_since_id = mentions[0].id
messages = []
for mention in mentions:
filtered_text = await self._filter_text(mention.entities, mention.text)
if not filtered_text:
continue
messages.append((mention.id, filtered_text))
2020-07-05 18:34:41 +00:00
return messages
async def _filter_text(self, entities, text):
remove_indices = set()
2020-07-05 18:34:41 +00:00
for user in entities.user_mentions:
remove_indices.update(range(user.indices[0], user.indices[1] + 1))
2020-07-05 18:34:41 +00:00
for url in entities.urls:
remove_indices.update(range(url.indices[0], url.indices[1] + 1))
2020-07-05 18:34:41 +00:00
for symbol in entities.symbols:
remove_indices.update(range(symbol.indices[0], symbol.indices[1] + 1))
2020-07-05 18:34:41 +00:00
filtered_text = ""
for index, character in enumerate(text):
if index not in remove_indices:
filtered_text += character
filtered_text = filtered_text.strip()
return filtered_text
async def push(self):
while True:
message = await self.receive()
logger.debug(
'Received message from censor (%s): %s'
% (self.twitter_model.hood.name, message.text)
)
if hasattr(message, 'twitter_mention_id'):
await self._retweet(message.twitter_mention_id)
else:
await self._post_tweet(message.text)
2020-07-05 18:34:41 +00:00
async def _post_tweet(self, message):
return await self.client.api.statuses.update.post(status=message)
async def _retweet(self, message_id):
return await self.client.api.statuses.retweet.post(id=message_id)
spawner = Spawner(Twitter, TwitterBot)