nopaque/app/blueprints/corpora/routes.py

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