mirror of
https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git
synced 2024-12-26 03:14:19 +00:00
300 lines
7.3 KiB
Python
300 lines
7.3 KiB
Python
from datetime import datetime
|
||
from flask import (
|
||
abort,
|
||
current_app,
|
||
flash,
|
||
Flask,
|
||
jsonify,
|
||
redirect,
|
||
request,
|
||
render_template,
|
||
url_for
|
||
)
|
||
from flask_login import current_user
|
||
from string import punctuation
|
||
from threading import Thread
|
||
import nltk
|
||
from app import db
|
||
from app.models import (
|
||
Corpus,
|
||
CorpusFollowerAssociation,
|
||
CorpusFollowerRole,
|
||
User
|
||
)
|
||
from . import bp
|
||
from .decorators import corpus_follower_permission_required
|
||
from .forms import CreateCorpusForm
|
||
|
||
|
||
|
||
def _delete_corpus(app: Flask, corpus_id: int):
|
||
with app.app_context():
|
||
corpus: Corpus = Corpus.query.get(corpus_id)
|
||
corpus.delete()
|
||
db.session.commit()
|
||
|
||
|
||
def _build_corpus(app: Flask, corpus_id: int):
|
||
with app.app_context():
|
||
corpus = Corpus.query.get(corpus_id)
|
||
corpus.build()
|
||
db.session.commit()
|
||
|
||
|
||
@bp.route('')
|
||
def corpora():
|
||
return redirect(url_for('main.dashboard', _anchor='corpora'))
|
||
|
||
|
||
@bp.route('/create', methods=['GET', 'POST'])
|
||
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()
|
||
|
||
flash(f'Corpus "{corpus.title}" created', 'corpus')
|
||
return redirect(corpus.url)
|
||
|
||
return render_template(
|
||
'corpora/create.html.j2',
|
||
title='Create corpus',
|
||
form=form
|
||
)
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>')
|
||
def corpus(corpus_id: int):
|
||
corpus = Corpus.query.get_or_404(corpus_id)
|
||
|
||
cfa = CorpusFollowerAssociation.query.filter_by(
|
||
corpus_id=corpus_id,
|
||
follower_id=current_user.id
|
||
).first()
|
||
|
||
if cfa is None:
|
||
if corpus.user == current_user or current_user.is_administrator:
|
||
cfr = CorpusFollowerRole.query.filter_by(name='Administrator').first()
|
||
else:
|
||
cfr = CorpusFollowerRole.query.filter_by(name='Anonymous').first()
|
||
else:
|
||
cfr = cfa.role
|
||
|
||
cfrs = CorpusFollowerRole.query.all()
|
||
|
||
# TODO: Better solution for filtering admin
|
||
users = User.query.filter(
|
||
User.is_public == True,
|
||
User.id != current_user.id,
|
||
User.id != corpus.user.id,
|
||
User.role_id < 4
|
||
).all()
|
||
|
||
if (
|
||
corpus.user == current_user
|
||
or current_user.is_administrator
|
||
):
|
||
return render_template(
|
||
'corpora/corpus.html.j2',
|
||
title=corpus.title,
|
||
corpus=corpus,
|
||
cfr=cfr,
|
||
cfrs=cfrs,
|
||
users=users
|
||
)
|
||
|
||
if (
|
||
current_user.is_following_corpus(corpus)
|
||
or corpus.is_public
|
||
):
|
||
cfas = CorpusFollowerAssociation.query.filter(
|
||
Corpus.id == corpus_id,
|
||
CorpusFollowerAssociation.follower_id != corpus.user.id
|
||
).all()
|
||
return render_template(
|
||
'corpora/public_corpus.html.j2',
|
||
title=corpus.title,
|
||
corpus=corpus,
|
||
cfrs=cfrs,
|
||
cfr=cfr,
|
||
cfas=cfas,
|
||
users=users
|
||
)
|
||
|
||
abort(403)
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>', methods=['DELETE'])
|
||
def delete_corpus(corpus_id: int):
|
||
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 jsonify(f'Corpus "{corpus.title}" marked for deletion.'), 202
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>/build', methods=['POST'])
|
||
def build_corpus(corpus_id: int):
|
||
corpus = Corpus.query.get_or_404(corpus_id)
|
||
|
||
cfa = CorpusFollowerAssociation.query.filter_by(
|
||
corpus_id=corpus_id,
|
||
follower_id=current_user.id
|
||
).first()
|
||
|
||
if not (
|
||
cfa is not None and cfa.role.has_permission('MANAGE_FILES')
|
||
or corpus.user == current_user
|
||
or current_user.is_administrator
|
||
):
|
||
abort(403)
|
||
|
||
if len(corpus.files.all()) == 0:
|
||
abort(409)
|
||
|
||
thread = Thread(
|
||
target=_build_corpus,
|
||
args=(current_app._get_current_object(), corpus.id)
|
||
)
|
||
thread.start()
|
||
|
||
return jsonify(f'Corpus "{corpus.title}" marked for building.'), 202
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>/create-share-link', methods=['POST'])
|
||
def create_share_link(corpus_id: int):
|
||
data = request.json
|
||
|
||
expiration_date = data['expiration_date']
|
||
if not isinstance(expiration_date, str):
|
||
abort(400)
|
||
|
||
role_name = data['role_name']
|
||
if not isinstance(role_name, str):
|
||
abort(400)
|
||
|
||
corpus = Corpus.query.get_or_404(corpus_id)
|
||
|
||
cfa = CorpusFollowerAssociation.query.filter_by(
|
||
corpus_id=corpus_id,
|
||
follower_id=current_user.id
|
||
).first()
|
||
|
||
if not (
|
||
cfa is not None and cfa.role.has_permission('MANAGE_FOLLOWERS')
|
||
or corpus.user == current_user
|
||
or current_user.is_administrator
|
||
):
|
||
abort(403)
|
||
|
||
_expiration_date = datetime.strptime(expiration_date, '%b %d, %Y')
|
||
|
||
cfr = CorpusFollowerRole.query.filter_by(name=role_name).first()
|
||
if cfr is None:
|
||
abort(400)
|
||
|
||
token = current_user.generate_follow_corpus_token(
|
||
corpus.hashid,
|
||
role_name,
|
||
_expiration_date
|
||
)
|
||
|
||
corpus_share_link = url_for(
|
||
'corpora.follow_corpus',
|
||
corpus_id=corpus_id,
|
||
token=token,
|
||
_external=True
|
||
)
|
||
|
||
return jsonify(corpus_share_link)
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>/analysis')
|
||
@corpus_follower_permission_required('VIEW')
|
||
def analysis(corpus_id: int):
|
||
corpus = Corpus.query.get_or_404(corpus_id)
|
||
|
||
return render_template(
|
||
'corpora/analysis.html.j2',
|
||
corpus=corpus,
|
||
title=f'Analyse Corpus {corpus.title}'
|
||
)
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>/analysis/stopwords')
|
||
def get_stopwords(corpus_id: int):
|
||
languages = [
|
||
'german',
|
||
'english',
|
||
'catalan',
|
||
'greek',
|
||
'spanish',
|
||
'french',
|
||
'italian',
|
||
'russian',
|
||
'chinese'
|
||
]
|
||
|
||
nltk.download('stopwords', quiet=True)
|
||
stopwords = {
|
||
language: nltk.corpus.stopwords.words(language)
|
||
for language in languages
|
||
}
|
||
stopwords['punctuation'] = list(punctuation)
|
||
stopwords['punctuation'] += ['—', '|', '–', '“', '„', '--']
|
||
stopwords['user_stopwords'] = []
|
||
|
||
return jsonify(stopwords)
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>/follow/<token>')
|
||
def follow_corpus(corpus_id: int, token: str):
|
||
corpus = Corpus.query.get_or_404(corpus_id)
|
||
|
||
if not current_user.follow_corpus_by_token(token):
|
||
abort(403)
|
||
|
||
db.session.commit()
|
||
|
||
flash(f'You are following "{corpus.title}" now', category='corpus')
|
||
return redirect(corpus.url)
|
||
|
||
|
||
@bp.route('/<hashid:corpus_id>/is-public', methods=['PUT'])
|
||
def update_is_public(corpus_id):
|
||
new_value = request.json
|
||
if not isinstance(new_value, bool):
|
||
abort(400)
|
||
|
||
corpus = Corpus.query.get_or_404(corpus_id)
|
||
|
||
if not (
|
||
corpus.user == current_user
|
||
or current_user.is_administrator
|
||
):
|
||
abort(403)
|
||
|
||
|
||
corpus.is_public = new_value
|
||
db.session.commit()
|
||
|
||
return jsonify(f'Corpus "{corpus.title}" is now {"public" if new_value else "private"}'), 200
|