nopaque/app/contributions/routes.py

from flask import (
    abort,
    current_app,
    flash,
    Markup,
    redirect,
    render_template,
    url_for
)
from flask_login import login_required, current_user
from threading import Thread
from app import db
from app.decorators import permission_required 
from app.models import (
    Permission,
    SpaCyNLPPipelineModel,
    TesseractOCRPipelineModel
)
from . import bp
from .forms import (
    CreateSpaCyNLPPipelineModelForm,
    CreateTesseractOCRPipelineModelForm,
    EditSpaCyNLPPipelineModelForm,
    EditTesseractOCRPipelineModelForm
)


@bp.before_request
@login_required
@permission_required(Permission.CONTRIBUTE)
def before_request():
    pass


@bp.route('/')
def contributions():
    return render_template(
        'contributions/contributions.html.j2',
        title='Contributions'
    )


@bp.route('/tesseract-ocr-pipeline-models')
def tesseract_ocr_pipeline_models():
    return render_template(
        'contributions/tesseract_ocr_pipeline_models.html.j2',
        title='Tesseract OCR Pipeline Models'
    )


@bp.route('/tesseract-ocr-pipeline-models/<hashid:tesseract_ocr_pipeline_model_id>', methods=['GET', 'POST'])
def tesseract_ocr_pipeline_model(tesseract_ocr_pipeline_model_id):
    tesseract_ocr_pipeline_model = TesseractOCRPipelineModel.query.get_or_404(tesseract_ocr_pipeline_model_id)
    form = EditTesseractOCRPipelineModelForm(
        obj=tesseract_ocr_pipeline_model,
        prefix='edit-tesseract-ocr-pipeline-model-form'
    )
    if form.validate_on_submit():
        form.populate_obj(tesseract_ocr_pipeline_model)
        if db.session.is_modified(tesseract_ocr_pipeline_model):
            message = Markup(f'Tesseract OCR Pipeline model "<a href="{tesseract_ocr_pipeline_model.url}">{tesseract_ocr_pipeline_model.title}</a>" updated')
            flash(message)
            db.session.commit()
        return redirect(url_for('.tesseract_ocr_pipeline_models'))
    return render_template(
        'contributions/tesseract_ocr_pipeline_model.html.j2',
        form=form,
        tesseract_ocr_pipeline_model=tesseract_ocr_pipeline_model,
        title='Edit Tesseract OCR Pipeline Model'
    )


@bp.route('/tesseract-ocr-pipeline-models/<hashid:tesseract_ocr_pipeline_model_id>', methods=['DELETE'])
def delete_tesseract_model(tesseract_ocr_pipeline_model_id):
    def _delete_tesseract_ocr_pipeline_model(app, tesseract_ocr_pipeline_model_id):
        with app.app_context():
            tesseract_ocr_pipeline_model = TesseractOCRPipelineModel.query.get(tesseract_ocr_pipeline_model_id)
            tesseract_ocr_pipeline_model.delete()
            db.session.commit()

    tesseract_ocr_pipeline_model = TesseractOCRPipelineModel.query.get_or_404(tesseract_ocr_pipeline_model_id)
    if not (tesseract_ocr_pipeline_model.user == current_user or current_user.is_administrator()):
        abort(403)
    thread = Thread(
        target=_delete_tesseract_ocr_pipeline_model,
        args=(current_app._get_current_object(), tesseract_ocr_pipeline_model_id)
    )
    thread.start()
    return {}, 202


@bp.route('/tesseract-ocr-pipeline-models/create', methods=['GET', 'POST'])
def create_tesseract_ocr_pipeline_model():
    form = CreateTesseractOCRPipelineModelForm(prefix='create-tesseract-ocr-pipeline-model-form')
    if form.is_submitted():
        if not form.validate():
            response = {'errors': form.errors}
            return response, 400
        try:
            tesseract_ocr_pipeline_model = TesseractOCRPipelineModel.create(
                form.tesseract_model_file.data,
                compatible_service_versions=form.compatible_service_versions.data,
                description=form.description.data,
                publisher=form.publisher.data,
                publisher_url=form.publisher_url.data,
                publishing_url=form.publishing_url.data,
                publishing_year=form.publishing_year.data,
                shared=form.shared.data,
                title=form.title.data,
                version=form.version.data,
                user=current_user
            )
        except OSError:
            abort(500)
        db.session.commit()
        tesseract_ocr_pipeline_model_url = url_for(
            '.tesseract_ocr_pipeline_model',
            tesseract_ocr_pipeline_model_id=tesseract_ocr_pipeline_model.id
        )
        message = Markup(f'Tesseract OCR Pipeline model "<a href="{tesseract_ocr_pipeline_model_url}">{tesseract_ocr_pipeline_model.title}</a>" created')
        flash(message)
        return {}, 201, {'Location': tesseract_ocr_pipeline_model_url}
    return render_template(
        'contributions/create_tesseract_ocr_pipeline_model.html.j2',
        form=form,
        title='Create Tesseract OCR Pipeline Model'
    )

@bp.route('/tesseract-ocr-pipeline-models/<hashid:tesseract_ocr_pipeline_model_id>/toggle-public-status', methods=['POST'])
def toggle_tesseract_ocr_pipeline_model_public_status(tesseract_ocr_pipeline_model_id):
    tesseract_ocr_pipeline_model = TesseractOCRPipelineModel.query.get_or_404(tesseract_ocr_pipeline_model_id)
    if not (tesseract_ocr_pipeline_model.user == current_user or current_user.is_administrator()):
        abort(403)
    tesseract_ocr_pipeline_model.shared = not tesseract_ocr_pipeline_model.shared
    db.session.commit()
    return {}, 201


@bp.route('/spacy-nlp-pipeline-models')
def spacy_nlp_pipeline_models():
    return render_template(
        'contributions/spacy_nlp_pipeline_models.html.j2',
        title='SpaCy NLP Pipeline Models'
    )


@bp.route('/spacy-nlp-pipeline-models/<hashid:spacy_nlp_pipeline_model_id>', methods=['GET', 'POST'])
def spacy_nlp_pipeline_model(spacy_nlp_pipeline_model_id):
    spacy_nlp_pipeline_model = SpaCyNLPPipelineModel.query.get_or_404(spacy_nlp_pipeline_model_id)
    form = EditSpaCyNLPPipelineModelForm(
        obj=spacy_nlp_pipeline_model,
        prefix='edit-spacy-nlp-pipeline-model-form'
    )
    if form.validate_on_submit():
        form.populate_obj(spacy_nlp_pipeline_model)
        if db.session.is_modified(spacy_nlp_pipeline_model):
            message = Markup(f'SpaCy NLP Pipeline model "<a href="{spacy_nlp_pipeline_model.url}">{spacy_nlp_pipeline_model.title}</a>" updated')
            flash(message)
            db.session.commit()
        return redirect(url_for('.spacy_nlp_pipeline_models'))
    return render_template(
        'contributions/spacy_nlp_pipeline_model.html.j2',
        form=form,
        spacy_nlp_pipeline_model=spacy_nlp_pipeline_model,
        title=f'{spacy_nlp_pipeline_model.title} [{spacy_nlp_pipeline_model.version}]'
    )

@bp.route('/spacy-nlp-pipeline-models/<hashid:spacy_nlp_pipeline_model_id>', methods=['DELETE'])
def delete_spacy_model(spacy_nlp_pipeline_model_id):
    def _delete_spacy_model(app, spacy_nlp_pipeline_model_id):
        with app.app_context():
            spacy_nlp_pipeline_model = SpaCyNLPPipelineModel.query.get(spacy_nlp_pipeline_model_id)
            spacy_nlp_pipeline_model.delete()
            db.session.commit()
    
    spacy_nlp_pipeline_model = SpaCyNLPPipelineModel.query.get_or_404(spacy_nlp_pipeline_model_id)
    if not (spacy_nlp_pipeline_model.user == current_user or current_user.is_administrator()):
        abort(403)
    thread = Thread(
        target=_delete_spacy_model,
        args=(current_app._get_current_object(), spacy_nlp_pipeline_model_id)
    )
    thread.start()
    return {}, 202


@bp.route('/spacy-nlp-pipeline-models/create', methods=['GET', 'POST'])
def create_spacy_nlp_pipeline_model():
    form = CreateSpaCyNLPPipelineModelForm(prefix='create-spacy-nlp-pipeline-model-form')
    if form.is_submitted():
        if not form.validate():
            response = {'errors': form.errors}
            return response, 400
        try:
            spacy_nlp_pipeline_model = SpaCyNLPPipelineModel.create(
                form.spacy_model_file.data,
                compatible_service_versions=form.compatible_service_versions.data,
                description=form.description.data,
                pipeline_name=form.pipeline_name.data,
                publisher=form.publisher.data,
                publisher_url=form.publisher_url.data,
                publishing_url=form.publishing_url.data,
                publishing_year=form.publishing_year.data,
                shared=form.shared.data,
                title=form.title.data,
                version=form.version.data,
                user=current_user
            )
        except OSError:
            abort(500)
        db.session.commit()
        spacy_nlp_pipeline_model_url = url_for(
            '.spacy_nlp_pipeline_model',
            spacy_nlp_pipeline_model_id=spacy_nlp_pipeline_model.id
        )
        message = Markup(f'SpaCy NLP Pipeline model "<a href="{spacy_nlp_pipeline_model_url}">{spacy_nlp_pipeline_model.title}</a>" created')
        flash(message)
        return {}, 201, {'Location': spacy_nlp_pipeline_model_url}
    return render_template(
        'contributions/create_spacy_nlp_pipeline_model.html.j2',
        form=form,
        title='Create SpaCy NLP Pipeline Model'
    )

@bp.route('/spacy-nlp-pipeline-models/<hashid:spacy_nlp_pipeline_model_id>/toggle-public-status', methods=['POST'])
def toggle_spacy_nlp_pipeline_model_public_status(spacy_nlp_pipeline_model_id):
    spacy_nlp_pipeline_model = SpaCyNLPPipelineModel.query.get_or_404(spacy_nlp_pipeline_model_id)
    if not (spacy_nlp_pipeline_model.user == current_user or current_user.is_administrator()):
        abort(403)
    spacy_nlp_pipeline_model.shared = not spacy_nlp_pipeline_model.shared
    db.session.commit()
    return {}, 201