from datetime import datetime 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, CorpusStatus, User from . import bp from .forms import ( CreateCorpusFileForm, CreateCorpusForm, UpdateCorpusFileForm ) # @bp.route('/share/', methods=['GET', 'POST']) # def share_corpus(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 # corpus_hashid = payload.get('sub') # corpus_id = hashids.decode(corpus_hashid) # return redirect(url_for('.corpus', corpus_id=corpus_id)) @bp.route('//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('//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('//follow', methods=['GET', 'POST']) # @login_required # def follow_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 # else: # if not current_user.is_administrator(): # abort(403) # else: # user_id = hashids.decode(user_hashid) # user = User.query.get_or_404(user_id) # if not user.is_following_corpus(corpus): # user.follow_corpus(corpus) # db.session.commit() # flash(f'You are following {corpus.title} now', category='corpus') # return {}, 202 @bp.route('//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 {}, 202 # @bp.route('/add_permission///') # def add_permission(corpus_id, user_id, permission): # a = CorpusFollowerAssociation.query.filter_by(followed_corpus_id=corpus_id, following_user_id=user_id).first_or_404() # a.add_permission(permission) # db.session.commit() # return 'ok' # @bp.route('/remove_permission///') # def remove_permission(corpus_id, user_id, permission): # a = CorpusFollowerAssociation.query.filter_by(followed_corpus_id=corpus_id, following_user_id=user_id).first_or_404() # a.remove_permission(permission) # db.session.commit() # return 'ok' @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 "{corpus.title}" created' ) flash(message, 'corpus') return redirect(corpus.url) return render_template( 'corpora/create_corpus.html.j2', form=form, title='Create corpus' ) @bp.route('/', methods=['GET', 'POST']) @login_required def corpus(corpus_id): corpus = Corpus.query.get_or_404(corpus_id) if corpus.user == current_user or current_user.is_administrator(): # now = datetime.utcnow() # payload = { # 'exp': now + timedelta(weeks=1), # 'iat': now, # 'iss': current_app.config['SERVER_NAME'], # 'sub': corpus.hashid # } # token = jwt.encode( # payload, # current_app.config['SECRET_KEY'], # algorithm='HS256' # ) return render_template( 'corpora/corpus.html.j2', corpus=corpus, # token=token, 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('/', 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('//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('//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('//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'"{corpus_file.filename}" 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('//files/', 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 "{corpus_file.filename}" 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('//files/', 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('//files//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('//export') @login_required def export_corpus(corpus_id): abort(503)