nopaque/app/services/routes.py

from flask import abort, current_app, flash, make_response, Markup, render_template, request
from flask_login import current_user, login_required
import requests
from app import db, hashids
from app.models import (
    Job,
    JobInput,
    JobStatus,
    TesseractOCRPipelineModel,
    SpaCyNLPPipelineModel
)
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),
                    'ocropus_nlbin_threshold': float(form.ocropus_nlbin_threshold.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_pipeline_models = [
        x for x in TesseractOCRPipelineModel.query.all()
        if version in x.compatible_service_versions and (x.is_public == True or x.user == current_user)
    ]
    return render_template(
        'services/tesseract_ocr_pipeline.html.j2',
        form=form,
        tesseract_ocr_pipeline_models=tesseract_ocr_pipeline_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)
    r = requests.get(
        'https://transkribus.eu/TrpServer/rest/models/text',
        headers={'Accept': 'application/json'}
    )
    if r.status_code != 200:
        abort(500)
    transkribus_htr_pipeline_models = r.json()['trpModelMetadata']
    form = CreateTranskribusHTRPipelineJobForm(
        transkribus_htr_pipeline_models=transkribus_htr_pipeline_models,
        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': 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}
    return render_template(
        'services/transkribus_htr_pipeline.html.j2',
        form=form,
        title=service_manifest['name'],
        transkribus_htr_pipeline_models=transkribus_htr_pipeline_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)
    spacy_nlp_pipeline_models = SpaCyNLPPipelineModel.query.all()
    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,
        spacy_nlp_pipeline_models=spacy_nlp_pipeline_models,
        title=service_manifest['name']
    )


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