mirror of
https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git
synced 2024-11-15 01:05:42 +00:00
379 lines
12 KiB
Python
379 lines
12 KiB
Python
from datetime import datetime, timedelta
|
|
from flask import (
|
|
abort,
|
|
current_app,
|
|
flash,
|
|
Markup,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
send_from_directory,
|
|
url_for
|
|
)
|
|
from flask_login import current_user, login_required
|
|
from threading import Thread
|
|
import jwt
|
|
import os
|
|
from app import db, hashids
|
|
from app.models import (
|
|
Corpus,
|
|
CorpusFile,
|
|
CorpusFollowerAssociation,
|
|
CorpusFollowPermission,
|
|
CorpusStatus,
|
|
User
|
|
)
|
|
from . import bp
|
|
from .forms import (
|
|
CreateCorpusFileForm,
|
|
CreateCorpusForm,
|
|
UpdateCorpusFileForm
|
|
)
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/enable_is_public', methods=['POST'])
|
|
@login_required
|
|
def enable_corpus_is_public(corpus_id):
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
corpus.is_public = True
|
|
db.session.commit()
|
|
return '', 204
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/disable_is_public', methods=['POST'])
|
|
@login_required
|
|
def disable_corpus_is_public(corpus_id):
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
corpus.is_public = False
|
|
db.session.commit()
|
|
return '', 204
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/follow/<token>')
|
|
@login_required
|
|
def follow_corpus(corpus_id, token):
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
current_app.config['SECRET_KEY'],
|
|
algorithms=['HS256'],
|
|
issuer=current_app.config['SERVER_NAME'],
|
|
options={'require': ['iat', 'iss', 'sub']}
|
|
)
|
|
except jwt.PyJWTError:
|
|
return False
|
|
return redirect(url_for('.corpus', corpus_id=corpus_id))
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/unfollow', methods=['GET', 'POST'])
|
|
@login_required
|
|
def unfollow_corpus(corpus_id):
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
user_hashid = request.args.get('user_id')
|
|
if user_hashid is None:
|
|
user = current_user
|
|
elif current_user.is_administrator():
|
|
user_id = hashids.decode(user_hashid)
|
|
user = User.query.get_or_404(user_id)
|
|
else:
|
|
abort(403)
|
|
if user.is_following_corpus(corpus):
|
|
user.unfollow_corpus(corpus)
|
|
db.session.commit()
|
|
flash(f'You are not following {corpus.title} anymore', category='corpus')
|
|
return '', 204
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/followers/<hashid:user_id>/permissions/add', methods=['POST'])
|
|
def add_permission(corpus_id, user_id, permission):
|
|
corpus_follow_association = CorpusFollowerAssociation.query.filter_by(followed_corpus_id=corpus_id, following_user_id=user_id).first_or_404()
|
|
permission = request.args.get('permission', type=int)
|
|
corpus = corpus_follow_association.followed_corpus
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
if permission is None or permission not in iter(CorpusFollowPermission):
|
|
abort(400)
|
|
corpus_follow_association.add_permission(permission)
|
|
db.session.commit()
|
|
return '', 204
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/followers/<hashid:user_id>/permissions/remove', methods=['POST'])
|
|
def remove_permission(corpus_id, user_id, permission):
|
|
corpus_follow_association = CorpusFollowerAssociation.query.filter_by(followed_corpus_id=corpus_id, following_user_id=user_id).first_or_404()
|
|
permission = request.args.get('permission')
|
|
corpus = corpus_follow_association.followed_corpus
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
if permission is None or permission not in iter(CorpusFollowPermission):
|
|
abort(400)
|
|
corpus_follow_association.remove_permission(permission)
|
|
db.session.commit()
|
|
return '', 204
|
|
|
|
|
|
@bp.route('/public')
|
|
@login_required
|
|
def public_corpora():
|
|
corpora = [
|
|
c.to_json_serializeable()
|
|
for c in Corpus.query.filter(Corpus.is_public == True).all()
|
|
]
|
|
return render_template(
|
|
'corpora/public_corpora.html.j2',
|
|
corpora=corpora,
|
|
title='Corpora'
|
|
)
|
|
|
|
|
|
@bp.route('/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
def create_corpus():
|
|
form = CreateCorpusForm()
|
|
if form.validate_on_submit():
|
|
try:
|
|
corpus = Corpus.create(
|
|
title=form.title.data,
|
|
description=form.description.data,
|
|
user=current_user
|
|
)
|
|
except OSError:
|
|
abort(500)
|
|
db.session.commit()
|
|
message = Markup(
|
|
f'Corpus "<a href="{corpus.url}">{corpus.title}</a>" created'
|
|
)
|
|
flash(message, 'corpus')
|
|
return redirect(corpus.url)
|
|
return render_template(
|
|
'corpora/create_corpus.html.j2',
|
|
form=form,
|
|
title='Create corpus'
|
|
)
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>')
|
|
@login_required
|
|
def corpus(corpus_id):
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
exp_date = (datetime.utcnow() + timedelta(days=7)).strftime('%b %d, %Y')
|
|
if corpus.user == current_user or current_user.is_administrator():
|
|
return render_template(
|
|
'corpora/corpus.html.j2',
|
|
corpus=corpus,
|
|
exp_date=exp_date,
|
|
title='Corpus'
|
|
)
|
|
if current_user.is_following_corpus(corpus) or corpus.is_public:
|
|
corpus_files = [x.to_json_serializeable() for x in corpus.files]
|
|
return render_template(
|
|
'corpora/public_corpus.html.j2',
|
|
corpus=corpus,
|
|
corpus_files=corpus_files,
|
|
title='Corpus'
|
|
)
|
|
abort(403)
|
|
|
|
@bp.route('/<hashid:corpus_id>/generate-corpus-share-link', methods=['GET', 'POST'])
|
|
@login_required
|
|
def generate_corpus_share_link(corpus_id):
|
|
data = request.get_json('data')
|
|
permission = data['permission']
|
|
expiration = data['expiration']
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
now = datetime.utcnow()
|
|
payload = {
|
|
'exp': expiration,
|
|
'iat': now,
|
|
'iss': current_app.config['SERVER_NAME'],
|
|
'sub': permission
|
|
}
|
|
token = jwt.encode(
|
|
payload,
|
|
current_app.config['SECRET_KEY'],
|
|
algorithm='HS256'
|
|
)
|
|
link = url_for('corpora.follow_corpus', corpus_id=corpus_id, token=token, _external=True)
|
|
return link
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>', methods=['DELETE'])
|
|
@login_required
|
|
def delete_corpus(corpus_id):
|
|
def _delete_corpus(app, corpus_id):
|
|
with app.app_context():
|
|
corpus = Corpus.query.get(corpus_id)
|
|
corpus.delete()
|
|
db.session.commit()
|
|
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
thread = Thread(
|
|
target=_delete_corpus,
|
|
args=(current_app._get_current_object(), corpus_id)
|
|
)
|
|
thread.start()
|
|
return {}, 202
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/analyse')
|
|
@login_required
|
|
def analyse_corpus(corpus_id):
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
if not (corpus.user == current_user
|
|
or current_user.is_administrator()
|
|
or current_user.is_following_corpus(corpus)):
|
|
abort(403)
|
|
return render_template(
|
|
'corpora/analyse_corpus.html.j2',
|
|
corpus=corpus,
|
|
title=f'Analyse Corpus {corpus.title}'
|
|
)
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/build', methods=['POST'])
|
|
@login_required
|
|
def build_corpus(corpus_id):
|
|
def _build_corpus(app, corpus_id):
|
|
with app.app_context():
|
|
corpus = Corpus.query.get(corpus_id)
|
|
corpus.build()
|
|
db.session.commit()
|
|
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
# Check if the corpus has corpus files
|
|
if not corpus.files.all():
|
|
response = {'errors': {'message': 'Corpus file(s) required'}}
|
|
return response, 409
|
|
thread = Thread(
|
|
target=_build_corpus,
|
|
args=(current_app._get_current_object(), corpus_id)
|
|
)
|
|
thread.start()
|
|
return {}, 202
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/files/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
def create_corpus_file(corpus_id):
|
|
corpus = Corpus.query.get_or_404(corpus_id)
|
|
if not (corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
form = CreateCorpusFileForm()
|
|
if form.is_submitted():
|
|
if not form.validate():
|
|
response = {'errors': form.errors}
|
|
return response, 400
|
|
try:
|
|
corpus_file = CorpusFile.create(
|
|
form.vrt.data,
|
|
address=form.address.data,
|
|
author=form.author.data,
|
|
booktitle=form.booktitle.data,
|
|
chapter=form.chapter.data,
|
|
editor=form.editor.data,
|
|
institution=form.institution.data,
|
|
journal=form.journal.data,
|
|
pages=form.pages.data,
|
|
publisher=form.publisher.data,
|
|
publishing_year=form.publishing_year.data,
|
|
school=form.school.data,
|
|
title=form.title.data,
|
|
mimetype='application/vrt+xml',
|
|
corpus=corpus
|
|
)
|
|
except (AttributeError, OSError):
|
|
abort(500)
|
|
corpus.status = CorpusStatus.UNPREPARED
|
|
db.session.commit()
|
|
message = Markup(
|
|
'Corpus file'
|
|
f'"<a href="{corpus_file.url}">{corpus_file.filename}</a>" added'
|
|
)
|
|
flash(message, category='corpus')
|
|
return {}, 201, {'Location': corpus.url}
|
|
return render_template(
|
|
'corpora/create_corpus_file.html.j2',
|
|
corpus=corpus,
|
|
form=form,
|
|
title='Add corpus file'
|
|
)
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/files/<hashid:corpus_file_id>', methods=['GET', 'POST'])
|
|
@login_required
|
|
def corpus_file(corpus_id, corpus_file_id):
|
|
corpus_file = CorpusFile.query.filter_by(corpus_id = corpus_id, id=corpus_file_id).first_or_404()
|
|
if not (corpus_file.corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
form = UpdateCorpusFileForm(data=corpus_file.to_json_serializeable())
|
|
if form.validate_on_submit():
|
|
form.populate_obj(corpus_file)
|
|
if db.session.is_modified(corpus_file):
|
|
corpus_file.corpus.status = CorpusStatus.UNPREPARED
|
|
db.session.commit()
|
|
message = Markup(f'Corpus file "<a href="{corpus_file.url}">{corpus_file.filename}</a>" updated')
|
|
flash(message, category='corpus')
|
|
return redirect(corpus_file.corpus.url)
|
|
return render_template(
|
|
'corpora/corpus_file.html.j2',
|
|
corpus=corpus_file.corpus,
|
|
corpus_file=corpus_file,
|
|
form=form,
|
|
title='Edit corpus file'
|
|
)
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/files/<hashid:corpus_file_id>', methods=['DELETE'])
|
|
@login_required
|
|
def delete_corpus_file(corpus_id, corpus_file_id):
|
|
def _delete_corpus_file(app, corpus_file_id):
|
|
with app.app_context():
|
|
corpus_file = CorpusFile.query.get(corpus_file_id)
|
|
corpus_file.delete()
|
|
db.session.commit()
|
|
|
|
corpus_file = CorpusFile.query.filter_by(corpus_id = corpus_id, id=corpus_file_id).first_or_404()
|
|
if not (corpus_file.corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
thread = Thread(
|
|
target=_delete_corpus_file,
|
|
args=(current_app._get_current_object(), corpus_file_id)
|
|
)
|
|
thread.start()
|
|
return {}, 202
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/files/<hashid:corpus_file_id>/download')
|
|
@login_required
|
|
def download_corpus_file(corpus_id, corpus_file_id):
|
|
corpus_file = CorpusFile.query.filter_by(corpus_id = corpus_id, id=corpus_file_id).first_or_404()
|
|
if not (corpus_file.corpus.user == current_user or current_user.is_administrator()):
|
|
abort(403)
|
|
return send_from_directory(
|
|
os.path.dirname(corpus_file.path),
|
|
os.path.basename(corpus_file.path),
|
|
as_attachment=True,
|
|
attachment_filename=corpus_file.filename,
|
|
mimetype=corpus_file.mimetype
|
|
)
|
|
|
|
|
|
@bp.route('/import', methods=['GET', 'POST'])
|
|
@login_required
|
|
def import_corpus():
|
|
abort(503)
|
|
|
|
|
|
@bp.route('/<hashid:corpus_id>/export')
|
|
@login_required
|
|
def export_corpus(corpus_id):
|
|
abort(503)
|