nopaque/app/services/routes.py

from flask import abort, current_app, flash, Markup, render_template, request
from flask_login import current_user, login_required
from app import db, hashids
from app.models import (
    Job,
    JobInput,
    JobStatus,
    TesseractOCRModel,
    TRANSKRIBUS_HTR_MODELS,
    TranskribusHTRModel
)
from . import bp, SERVICES
from .forms import (
    CreateFileSetupPipelineJobForm,
    CreateTesseractOCRPipelineJobForm,
    CreateTranskribusHTRPipelineJobForm,
    CreateSpacyNLPPipelineJobForm
)


@bp.route('/file-setup-pipeline', methods=['GET', 'POST'])
@login_required
def file_setup_pipeline():
    service = 'file-setup-pipeline'
    service_manifest = SERVICES[service]
    version = request.args.get('version', service_manifest['latest_version'])
    if version not in service_manifest['versions']:
        abort(404)
    form = CreateFileSetupPipelineJobForm(prefix='create-job-form', version=version)
    if form.is_submitted():
        if not form.validate():
            response = {'errors': form.errors}
            return response, 400
        try:
            job = Job.create(
                title=form.title.data,
                description=form.description.data,
                service=service,
                service_args={},
                service_version=form.version.data,
                user=current_user
            )
        except OSError:
            abort(500)
        for input_file in form.images.data:
            try:
                JobInput.create(input_file, job=job)
            except (AttributeError, OSError):
                abort(500)
        job.status = JobStatus.SUBMITTED
        db.session.commit()
        message = Markup(f'Job "<a href="{job.url}">{job.title}</a>" created')
        flash(message, 'job')
        return {}, 201, {'Location': job.url}
    return render_template(
        'services/file_setup_pipeline.html.j2',
        form=form,
        title=service_manifest['name']
    )


@bp.route('/tesseract-ocr-pipeline', methods=['GET', 'POST'])
@login_required
def tesseract_ocr_pipeline():
    service_name = 'tesseract-ocr-pipeline'
    service_manifest = SERVICES[service_name]
    version = request.args.get('version', service_manifest['latest_version'])
    if version not in service_manifest['versions']:
        abort(404)
    form = CreateTesseractOCRPipelineJobForm(prefix='create-job-form', version=version)
    if form.is_submitted():
        if not form.validate():
            response = {'errors': form.errors}
            return response, 400
        try:
            job = Job.create(
                title=form.title.data,
                description=form.description.data,
                service=service_name,
                service_args={
                    'binarization': form.binarization.data,
                    'model': hashids.decode(form.model.data)
                },
                service_version=form.version.data,
                user=current_user
            )
        except OSError:
            abort(500)
        try:
            JobInput.create(form.pdf.data, job=job)
        except (AttributeError, OSError):
            abort(500)
        job.status = JobStatus.SUBMITTED
        db.session.commit()
        message = Markup(f'Job "<a href="{job.url}">{job.title}</a>" created')
        flash(message, 'job')
        return {}, 201, {'Location': job.url}
    tesseract_ocr_models = [
        x for x in TesseractOCRModel.query.all()
        if version in x.compatible_service_versions and (x.shared == True or x.user == current_user)
    ]
    return render_template(
        'services/tesseract_ocr_pipeline.html.j2',
        form=form,
        tesseract_ocr_models=tesseract_ocr_models,
        title=service_manifest['name']
    )


@bp.route('/transkribus-htr-pipeline', methods=['GET', 'POST'])
@login_required
def transkribus_htr_pipeline():
    if not current_app.config.get('NOPAQUE_TRANSKRIBUS_ENABLED'):
        abort(404)
    service = 'transkribus-htr-pipeline'
    service_manifest = SERVICES[service]
    version = request.args.get('version', service_manifest['latest_version'])
    if version not in service_manifest['versions']:
        abort(404)
    form = CreateTranskribusHTRPipelineJobForm(prefix='create-job-form', version=version)
    if form.is_submitted():
        if not form.validate():
            response = {'errors': form.errors}
            return response, 400
        try:
            job = Job.create(
                title=form.title.data,
                description=form.description.data,
                service=service,
                service_args={
                    'binarization': form.binarization.data,
                    'model': hashids.decode(form.model.data)
                },
                service_version=form.version.data,
                user=current_user
            )
        except OSError:
            abort(500)
        try:
            JobInput.create(form.pdf.data, job=job)
        except (AttributeError, OSError):
            abort(500)
        job.status = JobStatus.SUBMITTED
        db.session.commit()
        message = Markup(f'Job "<a href="{job.url}">{job.title}</a>" created')
        flash(message, 'job')
        return {}, 201, {'Location': job.url}
    transkribus_htr_models = [
        x for x in TranskribusHTRModel.query.all()
        if x.shared == True or x.user == current_user
    ]
    return render_template(
        'services/transkribus_htr_pipeline.html.j2',
        form=form,
        title=service_manifest['name'],
        TRANSKRIBUS_HTR_MODELS=TRANSKRIBUS_HTR_MODELS,
        transkribus_htr_models=transkribus_htr_models
    )


@bp.route('/spacy-nlp-pipeline', methods=['GET', 'POST'])
@login_required
def spacy_nlp_pipeline():
    service = 'spacy-nlp-pipeline'
    service_manifest = SERVICES[service]
    version = request.args.get('version', SERVICES[service]['latest_version'])
    if version not in service_manifest['versions']:
        abort(404)
    form = CreateSpacyNLPPipelineJobForm(prefix='create-job-form', version=version)
    if form.is_submitted():
        if not form.validate():
            response = {'errors': form.errors}
            return response, 400
        try:
            job = Job.create(
                title=form.title.data,
                description=form.description.data,
                service=service,
                service_args={
                    'encoding_detection': form.encoding_detection.data,
                    'model': form.model.data
                },
                service_version=form.version.data,
                user=current_user
            )
        except OSError:
            abort(500)
        try:
            JobInput.create(form.txt.data, job=job)
        except (AttributeError, OSError):
            abort(500)
        job.status = JobStatus.SUBMITTED
        db.session.commit()
        message = Markup(f'Job "<a href="{job.url}">{job.title}</a>" created')
        flash(message, 'job')
        return {}, 201, {'Location': job.url}
    return render_template(
        'services/spacy_nlp_pipeline.html.j2',
        form=form,
        title=service_manifest['name']
    )


@bp.route('/corpus-analysis')
@login_required
def corpus_analysis():
    return render_template(
        'services/corpus_analysis.html.j2',
        title='Corpus analysis'
    )