forget-cancellare-vecchi-toot/lib/mastodon.py

182 lines
5.1 KiB
Python
Raw Normal View History

from mastodon import Mastodon
2017-08-19 13:11:16 +02:00
from mastodon.Mastodon import MastodonAPIError
from model import MastodonApp, Account, OAuthToken, Post
from requests import head
from app import db
2017-08-19 13:11:16 +02:00
from math import inf
import iso8601
2017-08-29 14:46:32 +02:00
def get_or_create_app(instance_url, callback, website):
instance_url = instance_url
app = MastodonApp.query.get(instance_url)
try:
head('https://{}'.format(instance_url)).raise_for_status()
proto = 'https'
except Exception:
head('http://{}'.format(instance_url)).raise_for_status()
proto = 'http'
if not app:
2017-08-29 14:46:32 +02:00
client_id, client_secret = Mastodon.create_app(
'forget',
scopes=('read', 'write'),
api_base_url='{}://{}'.format(proto, instance_url),
redirect_uris=callback,
website=website,
)
app = MastodonApp()
app.instance = instance_url
app.client_id = client_id
app.client_secret = client_secret
app.protocol = proto
return app
2017-08-29 14:46:32 +02:00
def anonymous_api(app):
2017-08-29 14:46:32 +02:00
return Mastodon(
app.client_id,
client_secret=app.client_secret,
api_base_url='{}://{}'.format(app.protocol, app.instance),
)
2017-08-29 14:46:32 +02:00
def login_url(app, callback):
return anonymous_api(app).auth_request_url(
redirect_uris=callback,
scopes=('read', 'write',)
)
2017-08-29 14:46:32 +02:00
def receive_code(code, app, callback):
api = anonymous_api(app)
access_token = api.log_in(
code=code,
scopes=('read', 'write'),
redirect_uri=callback,
)
remote_acc = api.account_verify_credentials()
2017-08-19 13:11:16 +02:00
acc = account_from_api_object(remote_acc, app.instance)
acc = db.session.merge(acc)
2017-08-29 14:46:32 +02:00
token = OAuthToken(token=access_token)
2017-08-19 13:11:16 +02:00
token = db.session.merge(token)
token.account = acc
return token
def get_api_for_acc(account):
app = MastodonApp.query.get(account.mastodon_instance)
for token in account.tokens:
2017-08-29 14:46:32 +02:00
api = Mastodon(
app.client_id,
client_secret=app.client_secret,
api_base_url='{}://{}'.format(app.protocol, app.instance),
access_token=token.token,
ratelimit_method='throw',
)
2017-08-19 13:11:16 +02:00
# api.verify_credentials()
# doesnt error even if the token is revoked lol
# https://github.com/tootsuite/mastodon/issues/4637
# so we have to do this:
tl = api.timeline()
if 'error' in tl:
db.session.delete(token)
continue
return api
2017-08-19 13:11:16 +02:00
account.force_log_out()
def fetch_acc(acc, cursor=None):
api = get_api_for_acc(acc)
if not api:
print('no access, aborting')
return None
2017-08-29 14:46:32 +02:00
newacc = account_from_api_object(
api.account_verify_credentials(), acc.mastodon_instance)
2017-08-19 13:11:16 +02:00
acc = db.session.merge(newacc)
2017-08-29 14:46:32 +02:00
kwargs = dict(limit=40)
2017-08-19 13:11:16 +02:00
if cursor:
kwargs.update(cursor)
if 'max_id' not in kwargs:
2017-08-29 14:46:32 +02:00
most_recent_post = (
Post.query.with_parent(acc)
.order_by(db.desc(Post.created_at)).first())
2017-08-19 13:11:16 +02:00
if most_recent_post:
kwargs['since_id'] = most_recent_post.mastodon_id
statuses = api.account_statuses(acc.mastodon_id, **kwargs)
if statuses:
kwargs['max_id'] = +inf
for status in statuses:
post = post_from_api_object(status, acc.mastodon_instance)
db.session.merge(post)
kwargs['max_id'] = min(kwargs['max_id'], status['id'])
else:
kwargs = None
db.session.commit()
return kwargs
2017-08-29 14:46:32 +02:00
2017-08-19 13:11:16 +02:00
def post_from_api_object(obj, instance):
return Post(
2017-08-29 14:46:32 +02:00
mastodon_instance=instance,
mastodon_id=obj['id'],
favourite=obj['favourited'],
has_media=('media_attachments' in obj
and bool(obj['media_attachments'])),
created_at=iso8601.parse_date(obj['created_at']),
author_id=account_from_api_object(obj['account'], instance).id,
direct=obj['visibility'] == 'direct',
2017-08-19 13:11:16 +02:00
)
2017-08-29 14:46:32 +02:00
2017-08-19 13:11:16 +02:00
def account_from_api_object(obj, instance):
return Account(
2017-08-29 14:46:32 +02:00
mastodon_instance=instance,
mastodon_id=obj['id'],
2017-08-29 16:57:30 +02:00
screen_name='{}@{}'.format(obj['username'], instance),
2017-08-29 14:46:32 +02:00
display_name=obj['display_name'],
avatar_url=obj['avatar'],
reported_post_count=obj['statuses_count'],
2017-08-19 13:11:16 +02:00
)
2017-08-29 14:46:32 +02:00
2017-08-19 13:11:16 +02:00
def refresh_posts(posts):
acc = posts[0].author
api = get_api_for_acc(acc)
if not api:
raise Exception('no access')
new_posts = list()
for post in posts:
try:
status = api.status(post.mastodon_id)
2017-08-29 14:46:32 +02:00
new_post = db.session.merge(
post_from_api_object(status, post.mastodon_instance))
2017-08-19 13:11:16 +02:00
new_posts.append(new_post)
except MastodonAPIError as e:
if str(e) == 'Endpoint not found.':
db.session.delete(post)
else:
raise e
2017-08-19 13:11:16 +02:00
return new_posts
2017-08-29 14:46:32 +02:00
2017-08-19 13:11:16 +02:00
def delete(post):
api = get_api_for_acc(post.author)
api.status_delete(post.mastodon_id)
db.session.delete(post)