sbt-idp/cope2n-api/fwd_api/celery_worker/process_result_tasks.py

import traceback
import time
import uuid
import logging

from copy import deepcopy

from fwd_api.celery_worker.worker import app
from fwd_api.models import SubscriptionRequest
from fwd_api.exception.exceptions import InvalidException
from fwd_api.models import SubscriptionRequest
from fwd_api.constant.common import ProcessType
from fwd_api.utils.redis import RedisUtils
from fwd_api.utils import process as ProcessUtil

redis_client = RedisUtils()

def aggregate_result(results, doc_types):
    doc_types = doc_types.split(',')

    des_result = deepcopy(list(results.values()))[0]
    des_result["content"]["total_pages"] = 0
    des_result["content"]["ocr_num_pages"] = 0 
    des_result["content"]["document"][0]["end_page"] = 0
    des_result["content"]["document"][0]["content"][3]["value"] = [None for _ in range(doc_types.count("imei"))]
    des_result["content"]["document"][0]["content"][2]["value"] = []

    sorted_results = [None] * len(doc_types)
    for index, result in results.items():
        index = int(index)
        doc_type = doc_types[index]
        sorted_results[index] = ((doc_type, result))


    imei_count = 0
    for doc_type, result in sorted_results:
        des_result["content"]["total_pages"] += 1
        des_result["content"]["ocr_num_pages"] += 1
        des_result["content"]["document"][0]["end_page"] += 1
        if doc_type == "imei":
            des_result["content"]["document"][0]["content"][3]["value"][imei_count] = result["content"]["document"][0]["content"][3]["value"][0]
            imei_count += 1
        elif doc_type == "invoice":
            des_result["content"]["document"][0]["content"][0]["value"] = result["content"]["document"][0]["content"][0]["value"]
            des_result["content"]["document"][0]["content"][1]["value"] = result["content"]["document"][0]["content"][1]["value"]
            des_result["content"]["document"][0]["content"][2]["value"] += result["content"]["document"][0]["content"][2]["value"]
        elif doc_type == "all":
            des_result.update(result)
        else:
            raise InvalidException(f"doc_type: {doc_type}")

    return des_result

def print_id(rq_id):
    print(" [x] Received {rq}".format(rq=rq_id))


def to_status(result):
    print('X')
    if 'status' in result and result['status'] not in [200, 201, 202]:
        return 4
    return 3


def update_user(rq: SubscriptionRequest):
    sub = rq.subscription
    predict_status = rq.status
    if predict_status == 3:
        sub.current_token += ProcessUtil.token_value(int(rq.process_type))
        sub.save()

@app.task(name='process_sap_invoice_result')
def process_invoice_sap_result(rq_id, result):
    try:
        rq: SubscriptionRequest = \
            SubscriptionRequest.objects.filter(request_id=rq_id, process_type=ProcessType.INVOICE.value)[0]
        status = to_status(result)

        rq.predict_result = result
        rq.status = status
        rq.save()

        update_user(rq)
    except IndexError as e:
        print(e)
        print("NotFound request by requestId, %d", rq_id)
    except Exception as e:
        print(e)
        print("Fail Invoice %d", rq_id)
        traceback.print_exc()
        return "FailInvoice"


@app.task(name='process_fi_invoice_result')
def process_invoice_fi_result(rq_id, result):
    try:
        rq: SubscriptionRequest = \
            SubscriptionRequest.objects.filter(request_id=rq_id, process_type=ProcessType.FI_INVOICE.value)[0]
        status = to_status(result)

        rq.predict_result = result
        rq.status = status
        rq.save()

        update_user(rq)
    except IndexError as e:
        print(e)
        print("NotFound request by requestId, %d", rq_id)
    except Exception as e:
        print(e)
        print("Fail Invoice %d", rq_id)
        traceback.print_exc()
        return "FailInvoice"
    
@app.task(name='process_manulife_invoice_result')
def process_invoice_manulife_result(rq_id, result):
    try:
        rq: SubscriptionRequest = \
            SubscriptionRequest.objects.filter(request_id=rq_id, process_type=ProcessType.MANULIFE_INVOICE.value)[0]
        status = to_status(result)

        rq.predict_result = result
        rq.status = status
        rq.save()

        update_user(rq)
    except IndexError as e:
        print(e)
        print("NotFound request by requestId, %d", rq_id)
    except Exception as e:
        print(e)
        print("Fail Invoice %d", rq_id)
        traceback.print_exc()
        return "FailInvoice"
    
random_processor_name = None

@app.task(name='process_sbt_invoice_result')
def process_invoice_sbt_result(rq_id, result):
    global random_processor_name
    if random_processor_name is None:
        random_processor_name = uuid.uuid4()
    print(rq_id, random_processor_name)
    print_id(f"[DEBUG]: Received SBT request with id {rq_id}")
    try:
        page_index = int(rq_id.split("_sub_")[1])
        rq_id = rq_id.split("_sub_")[0]
        rq: SubscriptionRequest = SubscriptionRequest.objects.filter(request_id=rq_id).first()
        for i in range(10):
            if rq.ai_inference_start_time == 0:
                logging.warn(f"ai_inference_start_time = 0, looks like database is lagging, attemp {i} in 0.2 second ...")
                rq.refresh_from_db()
                time.sleep(0.2)
                if i == 9: # return an error
                    logging.warn("Unable to retrieve rq, exiting")
                    rq.status = 404 # stop waiting
                    rq.predict_result = result
                    rq.save()
                    update_user(rq)
                    return "FailInvoice"
            else:
                break

        # status = to_status(result)
        status = result.get("status", 200)
        redis_client.set_cache(rq_id, page_index, result)
        done = rq.pages == redis_client.get_size(rq_id)
        if status == 200:
            if done:
                rq.status = 200 # stop waiting
                results = redis_client.get_all_cache(rq_id)
                rq.predict_result = aggregate_result(results, rq.doc_type)
                # print(f"[DEBUG]: rq.predict_result: {rq.predict_result}")
                rq.save()
                
        else:
            rq.status = 404 # stop waiting
            rq.predict_result = result
            rq.save()

        rq.ai_inference_time = time.time() - rq.ai_inference_start_time
        rq.save()
        update_user(rq)
    except IndexError as e:
        print(e)
        print("NotFound request by requestId, %d", rq_id)
        rq.ai_inference_time = 0
        rq.save()
    except Exception as e:
        print(e)
        print("Fail Invoice %d", rq_id)
        traceback.print_exc()
        rq.ai_inference_time = 0
        rq.save()
        return "FailInvoice"