rest_api.py

from flask import send_file, send_from_directory
from apiflask import APIFlask, Schema, abort
from apiflask.fields import Integer, String, File
from apiflask.validators import Length, OneOf
# from marshmallow import Schema, fields
from werkzeug.utils import secure_filename
import os
from open_api.prep_file_api import preparation
from open_api.conversion_api import conversion_stage
from open_api.validation_api import validation_stage
from open_api.fileconverter import FileConverter
import random, string
from open_api.clean_folders import delete_old_folders

# from api_scripts.loading_animation import Loader

main_path = os.getcwd()
temp_file_folder = os.path.join(main_path, "input_files")

app = APIFlask(__name__, title='Mass Spectrometry (MS) File Converter', version='1.0', docs_path='/api-docs')

app.config['DESCRIPTION'] = "The objective is to convert MS file into mzML"
app.config['EXTERNAL_DOCS'] = {
    'description': "NFDI4Chem",
    'url': 'https://www.nfdi4chem.de/'
}

app.config['TERMS_OF_SERVICE'] = 'http://localhost:5000'
app.config['LICENSE'] = {
    'name': 'MIT',
    'url': 'https://opensource.org/licenses/MIT'
}

app.config['TAGS'] = [
    {'name': 'msconvert:Conversion',
     'description': 'The aim of conversion is to convert a format to a standardized/agreed format'},
    {'name': 'FileInfo:Validation',
     'description': 'The goal of validation is to Validate the converted format converted by converter service.'},
    {'name': 'FileConvert:Conversion',
     'description': 'The aim of conversion is to test the mzML file'},
]

app.config['IMAGE_FOLDER'] = 'input_files'


class Files_convert(Schema):
    main_file = File(
        required=True,
        metadata={'title': 'The path of the input file (in this case RAW file)',
                  'description': 'The objective is to convert the RAW file to mzML file using Proteowizard-->'
                                 'msconvert software'}
    )
    test = String(
        required=False,
        metadata={'title': 'This is for additional parameter that could be added for conversion',
                  'description': 'Additional parameter that can modify the outputs'}
    )


class Files_validate(Schema):
    validation_file = File(
        required=False,
        metadata={'title': 'The path of the intermediate file or "validate.yml" file generated by the converter service',
                  'description': 'The path of the intermediate yml file is given'}
    )
    test = String(
        required=False,
        metadata={'title': 'For conversion of convert_file',
                  'description': 'The conversion takes any convert_raw files and then converts to mzML'}
    )

class Filesconverter(Schema):
    main_file = File(
        required=True,
        metadata={'title': 'The path of the input file (in this case mzML file)',
                  'description': 'The objective is to test the mzML file to FileConvert'}
    )
    test = String(
        required=False,
        metadata={'title': 'This is for additional parameter that could be added for conversion',
                  'description': 'Additional parameter that can modify the outputs'}
    )

@app.post('/msconvert_convert')
@app.input(Files_convert, location='form_and_files')
@app.doc(tags=['msconvert:Conversion'])
def raw_conversion(form_and_files_data):
    #print(f'The folder path beforehand is {main_path}')
    file_t = form_and_files_data['main_file']
    filename = secure_filename(file_t.filename)
    folder_name = random_string_folder_generator()
    test = preparation(folder_name)
    delete_old_folders()
    #print(f'{temp_file_folder} and {filename}')
    temp_full_path = os.path.join(temp_file_folder, folder_name)
    file_t.save(os.path.join(temp_full_path, filename))
    #test = preparation()
    #test.current_path()
    test.create_intermediate_files()
    os.chdir(main_path)
    test2 = conversion_stage(folder_name)
    #test2.current_path()
    test2.create_bash()
    os.chdir(main_path)
    new_temp_file_folder = os.path.join(temp_file_folder, folder_name)
    for file in os.listdir(new_temp_file_folder):
        if ".mzML" in file:
            mzML_file = os.path.join(new_temp_file_folder,file)
            return send_from_directory(new_temp_file_folder, file, as_attachment=True)
            #return send_file(mzML_file, as_attachment=True)
    print("Download Complete")
    os.chdir(main_path)
    delete_old_folders()
    return {'message': 'Upload Success',
            'name of file': f'{filename}',
            'another_message': f'Have a nice day :)'
            }


@app.post('/fileinfo_validate')
@app.input(Files_validate, location='form_and_files')
@app.doc(tags=['FileInfo:Validation'])
def validation_result(form_and_files_data):
    #print(f'The folder path beforehand is {main_path}')
    file_t = form_and_files_data['validation_file']
    filename = secure_filename(file_t.filename)
    validation_folder_name = random_string_folder_generator()
    test = validation_stage(validation_folder_name)
    delete_old_folders()
    temp_full_path = os.path.join(temp_file_folder, validation_folder_name)
    file_t.save(os.path.join(temp_full_path, filename))
    #print(f'The folder path ATM is {main_path}')
    #test = validation_stage()
    test.validation_process()
    os.chdir(main_path)
    new_temp_file_folder = os.path.join(temp_file_folder, validation_folder_name)
    for file in os.listdir(new_temp_file_folder):
        if "_validation_result.txt" in file:
            print(f'validation Report Generated for file: {filename}')
            with open(os.path.join(new_temp_file_folder, file), 'r') as f_in:
                data = f_in.read()
            print(data)
    for file in os.listdir(new_temp_file_folder):
        if "_validation_result.txt" in file:
            return send_from_directory(new_temp_file_folder, file)
            # download_validation_result()
    #print(f'path of the folder is : {main_path}')
    # os.chdir(main_path)
    # test2.cleanup()
    delete_old_folders()
    return {'message': f'{data}'
            }

@app.post('/fileconvert_convert')
@app.input(Filesconverter, location='form_and_files')
@app.doc(tags=['FileConvert:Conversion'])
def Fileconvert(form_and_files_data):
    #print(f'The folder path beforehand is {main_path}')
    file_t = form_and_files_data['main_file']
    filename = secure_filename(file_t.filename)
    fileconversion_folder_name = random_string_folder_generator()
    test = FileConverter(fileconversion_folder_name)
    delete_old_folders()
    temp_full_path = os.path.join(temp_file_folder, fileconversion_folder_name)
    file_t.save(os.path.join(temp_full_path, filename))
    #print(f'The folder path ATM is {main_path}')
    #test = validation_stage()
    test.fileconverter_validation_process()
    os.chdir(main_path)
    new_temp_file_folder = os.path.join(temp_file_folder, fileconversion_folder_name)
    for file in os.listdir(new_temp_file_folder):
        if "_FileConverter_output_validation_result.txt" in file:
            print(f'validation Report Generated for file: {filename}')
            with open(os.path.join(new_temp_file_folder, file), 'r') as f_in:
                data = f_in.read()
            print(data)
            # download_validation_result()
    for file in os.listdir(new_temp_file_folder):
        if "_FileConverter_output.mzML" in file:
            return send_from_directory(new_temp_file_folder, file)
    #print(f'path of the folder is : {main_path}')
    # os.chdir(main_path)
    # test2.cleanup()
    delete_old_folders()
    return {'message': f'str({data})'
            }


def random_string_folder_generator():
    random_string= str(''.join(random.choices(string.ascii_lowercase+string.digits,k=8)))
    os.makedirs(os.path.join(temp_file_folder,random_string))
    return random_string


if __name__ == "__main__":
    os.chdir(main_path)
    app.run(host='0.0.0.0', debug=True)