A Python Mastodon bot that's geared towards instance admins and their needs
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

119 lines
4.3 KiB

def curate(config):
"""Curate (fav/boost) popular posts on local and/or federated timelines."""
# Setup Mastodon API
mastodon = Mastodon(
client_id=config['config']['client_cred_file'],
access_token=config['config']['user_cred_file'],
api_base_url=config['config']['api_base_url']
)
mastodon.public_stream(CurationBot(config))
subparsers.add_parser('curation', help='Run the curation bot (service)')
# Deal with curation command
if args.command == 'curation':
# Verify toot cache file folder exists
cache_file = pathlib.Path(config['curation']['cache_file'])
if not cache_file.parent.exists(): # pylint: disable=no-member
# Warn if curation cache file doesn't exist
sys.exit('curation cache_file directory does not exist')
if not cache_file.exists():
print('warning: curation cache_file file will be created')
curate(config)
class CurationBot(StreamListener):
"""Implementation of the Mastodon.py StreamListener class for curation bot purposes."""
def __init__(self, bot_config):
StreamListener.__init__(self)
self.bot_config = bot_config
# Get access to cache
self.conn = sqlite3.connect(self.bot_config['curation']['cache_file'])
self.cursor = self.conn.cursor()
# Ensure cache table has been created
self.cursor.execute("""\
CREATE TABLE IF NOT EXISTS toot_cache (
toot_id INTEGER PRIMARY KEY,
federated BOOL,
username VARCHAR(2048),
is_reply BOOL,
is_boost BOOL,
toot_timestamp TIMESTAMP,
favorites INTEGER,
boosts INTEGER
);
""")
self.conn.commit()
def __del__(self):
# Cleanup connection to sqlite database for rss cache
self.conn.close()
def on_update(self, status):
"""A new status has appeared!
'status' is the parsed JSON dictionary describing the
status.
"""
if status['visibility'] == 'public':
toot_id = status['id']
federated = '@' in status['account']['acct']
username = status['account']['acct']
is_reply = status['in_reply_to_id'] is not None
is_boost = status['reblog'] is not None
timestamp = status['created_at']
favorites = status['favourites_count']
boosts = status['reblogs_count']
# TODO: Add support for tags? - Will need many to 1 relationship and a cross table
# tags = status['tags']
# Ensure a toot isn't cached twice for some odd reason
self.cursor.execute(
'select count(1) as found from toot_cache where toot_id = ?',
(toot_id, ))
if self.cursor.fetchone()[0] > 0:
return
# Cache toot
self.cursor.execute(
'insert into toot_cache values (?, ?, ?, ?, ?, ?, ?, ?)',
(toot_id, federated, username, is_reply, is_boost, timestamp,
favorites, boosts))
self.conn.commit()
def on_notification(self, notification):
"""A new notification.
'notification' is the parsed JSON dictionary describing the
notification.
"""
# We don't care if notifications come through our bot / curation account
# Leave handling notifications/folow up to the admins and e-mail notifications
pass
def on_delete(self, status_id):
"""A status has been deleted.
status_id is the status' integer ID.
"""
# Remove the status from the toot_cache if we see a delete
self.cursor.execute('delete from toot_cache where toot_id = ?',
(status_id, ))
self.conn.commit()
def handle_heartbeat(self):
"""The server has sent us a keep-alive message.
This callback may be useful to carry out periodic housekeeping
tasks, or just to confirm that the connection is still
open.
"""
print('!!!!!!!!heartbeat!!!!!!!!')
print(
' we should probably update statuses and whatnot here or at least do some '
'housekeeping for old toots')