import traceback import time import uuid 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, process_type=ProcessType.SBT_INVOICE.value)[0] # 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}") redis_client.remove_cache(rq_id) rq.save() else: rq.status = 404 # stop waiting rq.predict_result = result redis_client.remove_cache(rq_id) 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"