nopaque/app/corpora/routes.py

from datetime import datetime, timedelta
from flask import (
    abort,
    current_app,
    flash,
    make_response,
    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,
    CorpusFollowerPermission,
    CorpusStatus,
    User
)
from . import bp
from .forms import (
    CreateCorpusFileForm,
    CreateCorpusForm,
    UpdateCorpusFileForm
)


@bp.route('/<hashid:corpus_id>/is_public/enable', 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>/is_public/disable', 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>/followers/<hashid:follower_id>/unfollow', methods=['POST'])
@login_required
def unfollow_corpus(corpus_id, follower_id):
    corpus = Corpus.query.get_or_404(corpus_id)
    follower = User.query.get_or_404(follower_id)
    if not (corpus.user == current_user or follower == current_user or current_user.is_administrator()):
        abort(403)
    if not follower.is_following_corpus(corpus):
        abort(409)  # 'User is not following the corpus'
    follower.unfollow_corpus(corpus)
    db.session.commit()
    flash(f'{follower.username} is not following {corpus.title} anymore', category='corpus')
    return '', 204


@bp.route('/<hashid:corpus_id>/unfollow', methods=['POST'])
@login_required
def current_user_unfollow_corpus(corpus_id):
    corpus = Corpus.query.get_or_404(corpus_id)
    if not current_user.is_following_corpus(corpus):
        abort(409)  # 'You are not following the corpus'
    current_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:follower_id>/permissions/<permission_name>/add', methods=['POST'])
def add_permission(corpus_id, follower_id, permission_name):
    try:
        permission = CorpusFollowerPermission[permission_name]
    except KeyError:
        abort(409)  # f'Permission "{permission_name}" does not exist'
    corpus_follower_association = CorpusFollowerAssociation.query.filter_by(corpus_id=corpus_id, follower_id=follower_id).first_or_404()
    if not (corpus_follower_association.corpus.user == current_user or current_user.is_administrator()):
        abort(403)
    corpus_follower_association.add_permission(permission)
    db.session.commit()
    return '', 204


@bp.route('/<hashid:corpus_id>/followers/<hashid:follower_id>/permissions/<permission_name>/remove', methods=['POST'])
def remove_permission(corpus_id, follower_id, permission_name):
    try:
        permission = CorpusFollowerPermission[permission_name]
    except KeyError:
        return make_response(f'Permission "{permission_name}" does not exist', 409)
    corpus_follower_association = CorpusFollowerAssociation.query.filter_by(corpus_id=corpus_id, follower_id=follower_id).first_or_404()
    if not (corpus_follower_association.corpus.user == current_user or current_user.is_administrator()):
        abort(403)
    corpus_follower_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)