forget-cancellare-vecchi-toot/routes.py

264 lines
8.6 KiB
Python
Raw Normal View History

from flask import render_template, url_for, redirect, request, g, Response, jsonify
2017-08-03 21:37:00 +02:00
from datetime import datetime, timedelta
2017-07-27 00:35:53 +02:00
import lib.twitter
import lib.mastodon
2017-08-01 20:57:15 +02:00
import lib
2017-08-12 01:04:22 +02:00
from lib.auth import require_auth, require_auth_api
from lib import set_session_cookie
2017-08-12 01:04:22 +02:00
from lib import get_viewer_session, get_viewer
from model import Account, Session, Post, TwitterArchive, MastodonApp, MastodonInstance
2017-08-10 17:07:39 +02:00
from app import app, db, sentry, limiter
2017-07-30 13:53:14 +02:00
import tasks
from zipfile import BadZipFile
2017-08-09 11:32:32 +02:00
from twitter import TwitterError
2017-08-09 14:27:39 +02:00
from urllib.error import URLError
2017-08-09 11:43:16 +02:00
import version
import lib.version
2017-07-25 09:52:24 +02:00
@app.before_request
def load_viewer():
2017-08-10 17:07:39 +02:00
g.viewer = get_viewer_session()
if g.viewer and sentry:
sentry.user_context({
'id': g.viewer.account.id,
'username': g.viewer.account.screen_name,
'service': g.viewer.account.service
})
2017-08-09 11:43:16 +02:00
@app.context_processor
def inject_version():
return dict(
version=version.version,
repo_url=lib.version.url_for_version(version.version),
)
2017-08-09 11:43:16 +02:00
2017-08-11 22:20:34 +02:00
@app.context_processor
def inject_sentry():
if sentry:
client_dsn = app.config.get('SENTRY_DSN').split('@')
client_dsn[:1] = client_dsn[0].split(':')
client_dsn = ':'.join(client_dsn[0:2]) + '@' + client_dsn[3]
return dict(sentry_dsn=client_dsn)
return dict()
@app.after_request
def touch_viewer(resp):
2017-08-10 17:07:39 +02:00
if 'viewer' in g and g.viewer:
set_session_cookie(g.viewer, resp, app.config.get('HTTPS'))
g.viewer.touch()
db.session.commit()
return resp
2017-08-11 19:13:37 +02:00
lib.brotli.brotli(app)
2017-08-11 17:57:32 +02:00
2017-07-25 09:52:24 +02:00
@app.route('/')
def index():
if g.viewer:
return render_template('logged_in.html', scales=lib.interval.SCALES,
2017-08-07 16:26:25 +02:00
tweet_archive_failed = 'tweet_archive_failed' in request.args,
settings_error = 'settings_error' in request.args
)
else:
return render_template('index.html',
twitter_login_error = 'twitter_login_error' in request.args)
2017-07-25 09:52:24 +02:00
@app.route('/login/twitter')
2017-08-10 17:07:39 +02:00
@limiter.limit('3/minute')
2017-07-27 00:35:53 +02:00
def twitter_login_step1():
try:
return redirect(lib.twitter.get_login_url(
callback = url_for('twitter_login_step2', _external=True),
**app.config.get_namespace("TWITTER_")
))
except (TwitterError, URLError):
return redirect(url_for('index', twitter_login_error='', _anchor='log_in'))
2017-07-27 00:35:53 +02:00
@app.route('/login/twitter/callback')
2017-08-10 17:07:39 +02:00
@limiter.limit('3/minute')
2017-07-27 00:35:53 +02:00
def twitter_login_step2():
try:
oauth_token = request.args['oauth_token']
oauth_verifier = request.args['oauth_verifier']
token = lib.twitter.receive_verifier(oauth_token, oauth_verifier, **app.config.get_namespace("TWITTER_"))
2017-07-30 13:53:14 +02:00
session = Session(account_id = token.account_id)
db.session.add(session)
db.session.commit()
2017-07-30 13:53:14 +02:00
tasks.fetch_acc.s(token.account_id).apply_async(routing_key='high')
2017-07-30 13:53:14 +02:00
resp = Response(status=302, headers={"location": url_for('index')})
set_session_cookie(session, resp, app.config.get('HTTPS'))
return resp
except (TwitterError, URLError):
return redirect(url_for('index', twitter_login_error='', _anchor='log_in'))
@app.route('/upload_tweet_archive', methods=('POST',))
2017-08-10 17:07:39 +02:00
@limiter.limit('10/10 minutes')
@require_auth
def upload_tweet_archive():
2017-07-31 00:07:34 +02:00
ta = TwitterArchive(account = g.viewer.account,
body = request.files['file'].read())
db.session.add(ta)
db.session.commit()
try:
2017-08-12 20:32:51 +02:00
files = lib.twitter.chunk_twitter_archive(ta.id)
ta.chunks = len(files)
db.session.commit()
2017-07-31 00:07:34 +02:00
assert ta.chunks > 0
2017-08-12 20:32:51 +02:00
for filename in files:
2017-08-18 23:03:49 +02:00
tasks.import_twitter_archive_month.s(ta.id, filename).apply_async()
2017-08-12 20:32:51 +02:00
return redirect(url_for('index', _anchor='recent_archives'))
except (BadZipFile, AssertionError):
2017-08-07 15:44:21 +02:00
return redirect(url_for('index', tweet_archive_failed='', _anchor='tweet_archive_import'))
2017-08-03 16:05:28 +02:00
@app.route('/settings', methods=('POST',))
@require_auth
def settings():
viewer = get_viewer()
try:
for attr in lib.settings.attrs:
2017-08-07 16:26:25 +02:00
if attr in request.form:
setattr(viewer, attr, request.form[attr])
db.session.commit()
except ValueError:
return 400
2017-07-31 18:29:09 +02:00
2017-08-03 16:05:28 +02:00
return redirect(url_for('index', settings_saved=''))
@app.route('/disable', methods=('POST',))
@require_auth
def disable():
g.viewer.account.policy_enabled = False
db.session.commit()
2017-07-31 18:29:09 +02:00
2017-08-03 16:05:28 +02:00
return redirect(url_for('index'))
@app.route('/enable', methods=('POST',))
@require_auth
def enable():
2017-08-03 21:37:00 +02:00
risky = False
if not 'confirm' in request.form and not g.viewer.account.policy_enabled:
if g.viewer.account.policy_delete_every == timedelta(0):
approx = g.viewer.account.estimate_eligible_for_delete()
return render_template('warn.html', message=f"""You've set the time between deleting posts to 0. Every post that matches your expiration rules will be deleted within minutes.
{ ("That's about " + str(approx) + " posts.") if approx > 0 else "" }
Go ahead?""")
2017-08-14 20:58:22 +02:00
if g.viewer.account.next_delete < datetime.now() - timedelta(days=365):
2017-08-03 21:37:00 +02:00
return render_template('warn.html', message="""Once you enable Forget, posts that match your expiration rules will be deleted <b>permanently</b>. We can't bring them back. Make sure that you won't miss them.""")
if not g.viewer.account.policy_enabled:
2017-08-14 20:58:22 +02:00
g.viewer.account.next_delete = datetime.now() + g.viewer.account.policy_delete_every
2017-08-03 21:37:00 +02:00
2017-08-03 16:05:28 +02:00
g.viewer.account.policy_enabled = True
db.session.commit()
return redirect(url_for('index'))
2017-07-31 18:29:09 +02:00
@app.route('/logout')
@require_auth
def logout():
if(g.viewer):
db.session.delete(g.viewer)
db.session.commit()
g.viewer = None
return redirect(url_for('index'))
2017-08-12 01:04:22 +02:00
@app.route('/api/settings', methods=('PUT',))
@require_auth_api
def api_settings_put():
viewer = get_viewer()
data = request.json
updated = dict()
for key in lib.settings.attrs:
if key in data:
setattr(viewer, key, data[key])
updated[key] = data[key]
db.session.commit()
return jsonify(status='success', updated=updated)
2017-08-12 01:52:33 +02:00
@app.route('/api/viewer')
@require_auth_api
def api_viewer():
2017-08-12 01:52:33 +02:00
viewer = get_viewer()
return jsonify(
post_count=viewer.post_count(),
eligible_for_delete_estimate=viewer.estimate_eligible_for_delete(),
display_name=viewer.display_name,
screen_name=viewer.screen_name,
avatar_url=viewer.avatar_url,
id=viewer.id,
service=viewer.service,
)
@app.route('/api/viewer/timers')
@require_auth_api
def api_viewer_timers():
viewer = get_viewer()
return jsonify(
last_refresh=viewer.last_refresh,
last_refresh_rel=lib.interval.relnow(viewer.last_refresh),
last_fetch=viewer.last_fetch,
last_fetch_rel=lib.interval.relnow(viewer.last_fetch),
2017-08-14 20:58:22 +02:00
next_delete=viewer.next_delete,
next_delete_rel=lib.interval.relnow(viewer.next_delete),
2017-08-12 01:52:33 +02:00
)
@app.route('/login/mastodon', methods=('GET', 'POST'))
def mastodon_login_step1():
instances = MastodonInstance.query.filter(MastodonInstance.popularity > 1).order_by(db.desc(MastodonInstance.popularity)).limit(16)
if request.method == 'GET':
return render_template('mastodon_login.html', instances=instances, generic_error = 'error' in request.args)
if not 'instance_url' in request.form or not request.form['instance_url']:
return render_template('mastodon_login.html', instances=instances, address_error=True)
instance_url = request.form['instance_url'].split("@")[-1].lower()
callback = url_for('mastodon_login_step2', instance=instance_url, _external=True)
app = lib.mastodon.get_or_create_app(instance_url,
callback,
url_for('index', _external=True))
db.session.merge(app)
db.session.commit()
return redirect(lib.mastodon.login_url(app, callback))
@app.route('/login/mastodon/callback/<instance>')
def mastodon_login_step2(instance):
code = request.args.get('code', None)
app = MastodonApp.query.get(instance)
if not code or not app:
return redirect('mastodon_login_step1', error=True)
callback = url_for('mastodon_login_step2', instance=instance, _external=True)
token = lib.mastodon.receive_code(code, app, callback)
account = token.account
sess = Session(account = account)
db.session.add(sess)
i=MastodonInstance(instance=instance)
i=db.session.merge(i)
i.bump()
db.session.commit()
g.viewer = sess
return redirect(url_for('index'))