diff --git a/Dockerfile b/Dockerfile index 1c1903f..f038eb8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.9 RUN mkdir -p /usr/src/app WORKDIR /usr/src/app -RUN apt-get update && apt-get install python3-pil tesseract-ocr libtesseract-dev tesseract-ocr-eng tesseract-ocr-slv tesseract-ocr-script-latn ffmpeg libsm6 libxext6 libgl1 -y +RUN apt-get update && apt-get install python3-pil tesseract-ocr libtesseract-dev tesseract-ocr-eng tesseract-ocr-slv tesseract-ocr-script-latn ffmpeg libsm6 libxext6 libgl1 libmagic1 -y COPY requirements.txt /usr/src/app/ diff --git a/requirements.txt b/requirements.txt index ea9e7a2..6a42c12 100644 --- a/requirements.txt +++ b/requirements.txt @@ -15,3 +15,4 @@ requests pytesseract==0.3.10 opencv-python==4.5.2.54 numpy==1.20.3 +python-magic==0.4.27 diff --git a/swagger_server/controllers/jobs_controller.py b/swagger_server/controllers/jobs_controller.py index 1afc869..7edd67a 100644 --- a/swagger_server/controllers/jobs_controller.py +++ b/swagger_server/controllers/jobs_controller.py @@ -1,15 +1,23 @@ import datetime +import os.path import peewee import asyncio - +import concurrent.futures as cf from swagger_server.models.job_response import JobResponse # noqa: E501 from swagger_server.requets_db.models.vrsta import (Job) from threading import Thread from swagger_server.classla import cl_utils +from swagger_server.utils import txt_utils +from werkzeug.datastructures import FileStorage +import threading +import time -CLASSLA_CONCURANCE_LIMIT = 4 -classla_sem = asyncio.Semaphore(CLASSLA_CONCURANCE_LIMIT) +CLASSLA_CONCURANCE_LIMIT = 3 +DOC2TEXT_CONCURANCE_LIMIT = 4 + +classla_sem = threading.Semaphore(CLASSLA_CONCURANCE_LIMIT) +doc2text_sem = asyncio.Semaphore(DOC2TEXT_CONCURANCE_LIMIT) def delete_job(job_id): # noqa: E501 @@ -53,51 +61,123 @@ def clear_up_unfinished_jobs(): """ In case server crashed while jobs were in queue... """ - # tu more but !=, ne deluje ce je "is not" - Job.update(started_on=None).where(Job.started_on != None, Job.finished_on == None).execute() + Job.update(started_on=None).where(Job.started_on.is_null(False), Job.finished_on.is_null()).execute() + if os.path.exists('tmp'): + for tmp_file in os.listdir('tmp'): + if not Job.select().where(Job.input_file == tmp_file).exists(): + os.remove(f'tmp/{tmp_file}') + + +async def try_do_jobs(): + with cf.ThreadPoolExecutor(max_workers=2) as ex: + ex.submit(try_do_jobs_classla) + ex.submit(try_do_jobs_doc2text) ### Job looping -async def try_do_jobs(): - await asyncio.sleep(15) # wait for tokenizers to load for classla ... +def try_do_jobs_classla(): + time.sleep(15) # wait for tokenizers to load for classla ... while True: try: if cl_utils.nlp_loaded: - # print(cl_utils.raw_text_to_conllu("Danes je lep soncen dan. Res je!")) if classla_sem._value > 0: - # classla - unfinished_jobs = Job.select() \ - .where(Job.finished_on == None, Job.started_on == None, Job.job_type == 2) \ + unfinished_jobs_txt = Job.select() \ + .where(Job.finished_on.is_null(), Job.job_type == 2, + Job.input_file.is_null(False)) \ .limit(classla_sem._value) - tasks = [ - asyncio.ensure_future(execute_classla_job(job)) - for job in unfinished_jobs - ] - asyncio.get_event_loop().create_task(prep_classla_jobs(tasks)) - # await asyncio.gather(*tasks) + unfinished_jobs_no_txt = Job.select() \ + .where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 2, + Job.input_file.is_null()) \ + .limit(classla_sem._value) + unfinished_jobs = [j for j in unfinished_jobs_txt] + [j for j in unfinished_jobs_no_txt] + unfinished_jobs = unfinished_jobs[:classla_sem._value] + with cf.ThreadPoolExecutor(max_workers=CLASSLA_CONCURANCE_LIMIT) as ex: + [ex.submit(execute_classla_job, job) for job in unfinished_jobs] - else: - pass except Exception as e: - print(f"Exception in ... {e}") + print(f"Exception in {__name__}: {e}") finally: - await asyncio.sleep(3) + time.sleep(3) -async def prep_classla_jobs(tasks): +### Job looping +def try_do_jobs_doc2text(): + while True: + try: + if doc2text_sem._value > 0: + unfinished_jobs = Job.select() \ + .where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type << [1, 12, 3, 32]) \ + .limit(doc2text_sem._value) + with cf.ThreadPoolExecutor(max_workers=DOC2TEXT_CONCURANCE_LIMIT) as ex: + [ex.submit(execute_doc2text_job, job) for job in unfinished_jobs] + except Exception as e: + print(f"Exception in {__name__}: {e}") + finally: + time.sleep(3) + + +async def prep_jobs(tasks): await asyncio.gather(*tasks) -async def execute_classla_job(job: Job): - async with classla_sem: +def execute_doc2text_job(job: Job): + try: + doc2text_sem.acquire() + del_file = False + job.started_on = datetime.datetime.utcnow() + job.save() + + tmp_file_path = job.input_file + if not os.path.exists(tmp_file_path): + job.finished_on = datetime.datetime.utcnow() + job.job_output = "ERROR - Temporary file went missing, couldn't properly finish job" + job.save() + return + + with open(tmp_file_path, 'rb+') as f: + file = FileStorage(f) + jtype = job.job_type + text = "" + if jtype in [1, 12]: + text = txt_utils.extract_text_prepResp(file) + elif jtype in [3, 32]: + text = txt_utils.ocr_text_prepResp(file) + + if jtype in [1, 3]: + job.job_output = text + job.finished_on = datetime.datetime.utcnow() + elif jtype in [12, 32]: + job.job_input = text + job.job_type = 2 + job.input_size = len(text) + del_file = True + job.save() + if del_file: + try: + os.remove(tmp_file_path) + except: + pass + + except: + job.started_on = None + job.save() + finally: + doc2text_sem.release() + + +def execute_classla_job(job: Job): + try: + classla_sem.acquire() job.started_on = datetime.datetime.utcnow() job.save() conllu, _ = cl_utils.raw_text_to_conllu(job.job_input) job.job_output = conllu job.finished_on = datetime.datetime.utcnow() job.save() + finally: + classla_sem.release() clear_up_unfinished_jobs() @@ -111,7 +191,3 @@ def loop_in_thread(loop): t = Thread(target=loop_in_thread, args=(loop,)) t.start() - -# clear_up_unfinished_jobs() -# loop = asyncio.get_event_loop() -# loop.run_until_complete(try_do_jobs()) this version seems more at home, but it blocks the thread, fix that? diff --git a/swagger_server/controllers/marktext_async_controller.py b/swagger_server/controllers/marktext_async_controller.py index 1cd4efb..36d8657 100644 --- a/swagger_server/controllers/marktext_async_controller.py +++ b/swagger_server/controllers/marktext_async_controller.py @@ -2,7 +2,6 @@ import connexion from swagger_server.models.oznaci_besedilo_async_body import OznaciBesediloAsyncBody # noqa: E501 from swagger_server.requets_db.models.vrsta import (JobManager) -from swagger_server.utils import txt_utils def get_text(body): # noqa: E501 @@ -17,6 +16,8 @@ def get_text(body): # noqa: E501 """ if connexion.request.is_json: body = OznaciBesediloAsyncBody.from_dict(connexion.request.get_json()) # noqa: E501 + else: + return "Request in wrong format", 400 # conllu = cl_utils.raw_text_to_conllu(body.besedilo) # return conllu job, is_old_job = JobManager.create_job(2, body.besedilo) @@ -37,7 +38,11 @@ def get_conllu_from_file_async(file=None): # noqa: E501 :rtype: str """ - return 'do some magic!' + job, is_old_job = JobManager.create_job(12, file) + if job is None: + return "Something went wrong", 500 + ret = {'check_job_url': f'{connexion.request.url_root}/job/{job.id}'} + return ret, 200 def get_conllu_from_file_ocr_async(file=None): # noqa: E501 @@ -50,7 +55,11 @@ def get_conllu_from_file_ocr_async(file=None): # noqa: E501 :rtype: str """ - return 'do some magic!' + job, is_old_job = JobManager.create_job(32, file) + if job is None: + return "Something went wrong", 500 + ret = {'check_job_url': f'{connexion.request.url_root}/job/{job.id}'} + return ret, 200 def get_text_from_doc_async(file=None): # noqa: E501 @@ -63,7 +72,11 @@ def get_text_from_doc_async(file=None): # noqa: E501 :rtype: str """ - return 'do some magic!' + job, is_old_job = JobManager.create_job(1, file) + if job is None: + return "Something went wrong", 500 + ret = {'check_job_url': f'{connexion.request.url_root}/job/{job.id}'} + return ret, 200 def get_text_from_file_ocr_async(file=None): # noqa: E501 @@ -76,4 +89,8 @@ def get_text_from_file_ocr_async(file=None): # noqa: E501 :rtype: str """ - return 'do some magic!' + job, is_old_job = JobManager.create_job(3, file) + if job is None: + return "Something went wrong", 500 + ret = {'check_job_url': f'{connexion.request.url_root}/job/{job.id}'} + return ret, 200 diff --git a/swagger_server/requets_db/models/vrsta.py b/swagger_server/requets_db/models/vrsta.py index 73f8a96..3c9fcb7 100644 --- a/swagger_server/requets_db/models/vrsta.py +++ b/swagger_server/requets_db/models/vrsta.py @@ -1,6 +1,11 @@ +import pathlib + +import werkzeug.datastructures from peewee import * from datetime import datetime import os +from swagger_server.util import get_random_filename +from werkzeug.utils import secure_filename LOCAL_PATH = "requets_db/dbs" SERVER_PATH = "swagger_server/requets_db/dbs" @@ -27,17 +32,18 @@ class BaseModel(Model): class Job(BaseModel): - id = AutoField() - job_type = IntegerField() # 1 = pretvori datoteko v besedilo, 2 = oznaci besedilo, 21 = oboje - job_input = TextField(index=True) + id = AutoField(index=True) + job_type = IntegerField() + job_input = TextField(index=True, null=True) job_output = TextField(null=True) created_on = DateTimeField(default=datetime.utcnow) finished_on = DateTimeField(null=True) started_on = DateTimeField(null=True) input_size = IntegerField() + input_file = TextField(index=True, null=True) -# db.drop_tables([Job]) +db.drop_tables([Job]) # TODO: After pushing this, comment it and push again db.create_tables([Job]) @@ -47,15 +53,27 @@ class JobManager: """ :param: job_type :possibilities: - 1 = txt to classla - 2 = file to txt and then mark with classla - 3 = file with ocr then to classla + # 1 = pretvori datoteko v besedilo, 2 = oznaci besedilo, 12 = oboje + # 3 = pretvori dat v besedilo OCR, 2 = oznaci besedilo, 32 = oboje :return: Job object, Did already exist boolean """ try: - job, is_new = Job.get_or_create(job_type=job_type, job_input=job_input, input_size=len(job_input)) - return job, not is_new + if job_type == 2: + job, is_new = Job.get_or_create(job_type=job_type, job_input=job_input, input_size=len(job_input)) + elif job_type in [1, 3, 12, 32]: + tmp_file = "" + while True: + # just in case a VERY rare chance of a same generate name happens + tmp_file = "tmp/" + secure_filename(get_random_filename() + "_" + job_input.filename) + if not os.path.exists(tmp_file): + break + pathlib.Path('tmp').mkdir(exist_ok=True) + job_input: werkzeug.datastructures.FileStorage + job_input.save(tmp_file) + job, is_new = Job.get_or_create(job_type=job_type, input_file=tmp_file, input_size=-1) + return job, is_new + except Exception as e: print(f'Exception at creating a job: {e}') return None, False diff --git a/swagger_server/util.py b/swagger_server/util.py index 8130360..d0cad9f 100644 --- a/swagger_server/util.py +++ b/swagger_server/util.py @@ -5,6 +5,9 @@ import six import typing from swagger_server import type_util import pandas as pd +from datetime import datetime +import string +import random def _deserialize(data, klass): @@ -186,3 +189,10 @@ def get_files_by_keywords(kljucnebesede): if amount >= 1: ret.append(_file[9:][:-12]) return ret + + +def get_random_filename(): + ts = str(int(datetime.now().timestamp())) + extra = ''.join(random.choices(string.ascii_letters + string.digits, k=8)) + return f'{ts}_{extra}' + diff --git a/swagger_server/utils/txt_utils.py b/swagger_server/utils/txt_utils.py index d28ee67..f3b83fe 100644 --- a/swagger_server/utils/txt_utils.py +++ b/swagger_server/utils/txt_utils.py @@ -8,6 +8,7 @@ from PyPDF2 import PdfReader from swagger_server.classla import cl_utils import cv2 import numpy as np +import magic tika_server = "http://tika2:9999/tika" @@ -15,17 +16,21 @@ tika_server = "http://tika2:9999/tika" # endpoint below to be used only for development purposes (don't need to run docker) # tika_server = "http://rsdo.lhrs.feri.um.si:9998/tika" -def extract_text_prepResp(file): +def extract_text_prepResp(file, content_type=""): + content_type = file.content_type + if content_type is None: + content_type = magic.from_file(file.stream.name, mime=True) + if tika_responding(): response = requests.put(tika_server, data=file) return response.text, 200 - if "openxmlformats-officedocument.wordprocessingml.document" in file.content_type: + if "openxmlformats-officedocument.wordprocessingml.document" in content_type: content = '\n'.join([p.text for p in docx.Document(file).paragraphs]) - elif "application/pdf" in file.content_type: + elif "application/pdf" in content_type: reader = PdfReader(file) content = '\n'.join([p.extract_text() for p in reader.pages]) content = content - elif "text/xml" in file.content_type: + elif "text/xml" in content_type: root = ET.parse(file).getroot() plainText = root.findall('PlainText') if len(plainText) == 0: