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('/') 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('/', 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('//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('//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('//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('//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('//follow/') 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('//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