Merge branch 'feature/CHINARPA-5155'

@@ -8,4 +8,6 @@ broker = conf.CELERY_BROKER_URL
 
 
 app = Celery('celery_compare', broker=broker, include=['celery_compare.tasks'])
 app = Celery('celery_compare', broker=broker, include=['celery_compare.tasks'])
 
 
-app.conf.update(worker_max_tasks_per_child=5, timezone='Asia/Shanghai')
+# worker_max_tasks_per_child ,worker执行了几次任务就会死
+#app.conf.update(worker_max_tasks_per_child=10, timezone='Asia/Shanghai')
+app.conf.update(timezone='Asia/Shanghai')

@@ -39,6 +39,7 @@ from apps.doc.models import (
     HILCompareReportNew,
     HILCompareReportNew,
     AFCCompareReportNew,
     AFCCompareReportNew,
     AFCDoc, 
     AFCDoc, 
+    HILDoc,
     DealerMapping,
     DealerMapping,
 )
 )
 from apps.doc import consts
 from apps.doc import consts
@@ -49,6 +50,12 @@ from apps.doc.named_enum import RequestTeam, RequestTrigger, ProcessName, ErrorT
 from common.tools.comparison import cp
 from common.tools.comparison import cp
 from common.tools.des import decode_des
 from common.tools.des import decode_des
 
 
+import threading
+import concurrent.futures
+from concurrent.futures import ThreadPoolExecutor
+
+pool = ThreadPoolExecutor(max_workers=50, thread_name_prefix="compare_thread_")
+
 compare_log = logging.getLogger('compare')
 compare_log = logging.getLogger('compare')
 log_base = '[Compare]'
 log_base = '[Compare]'
 # e_log_base = '[e-contract]'
 # e_log_base = '[e-contract]'
@@ -3663,6 +3670,24 @@ def se_compare(application_id, application_entity, ocr_res_id, last_obj, ocr_res
             }
             }
         }
         }
         try:
         try:
+            compare_log.info('{0} [SE] [cms sleep start] [entity={1}] [id={2}] '.format(log_base, application_entity, application_id))
+            # 实时查询延迟时间
+            try:
+                cms_delay_time_config = Configs.objects.filter(id=5).first()
+                if cms_delay_time_config is not None and cms_delay_time_config.value is not None and cms_delay_time_config.value.isdigit():
+                    cms_delay_time = cms_delay_time_config.value
+                else:
+                    cms_delay_time = 0
+            except Exception as e:
+                cms_delay_time = 0
+                compare_log.info('[get cms_delay_time_config fail] [error={0}]'.format(traceback.format_exc()))
+
+            compare_log.info('cms_delay_time:{0}'.format(cms_delay_time))
+            try:
+                time.sleep(int(cms_delay_time))
+            except Exception as e:
+                compare_log.info('[sleep error] [error={0}]'.format(traceback.format_exc()))
+            compare_log.info('{0} [SE] [cms sleep end] [entity={1}] [id={2}] '.format(log_base, application_entity, application_id))
             response = cms.send(data)  # interface_report ocr to cms
             response = cms.send(data)  # interface_report ocr to cms
         except Exception as e:
         except Exception as e:
             is_success = False
             is_success = False
@@ -3694,9 +3719,52 @@ def se_compare(application_id, application_entity, ocr_res_id, last_obj, ocr_res
 
 
 @app.task
 @app.task
 def fsm_compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True, is_cms=False):
 def fsm_compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True, is_cms=False):
+    # try:
+        # producer_thread_fsm = threading.Thread(target=fsm_compare_thread, args=(application_id, application_entity, uniq_seq, ocr_res_id, is_ca, is_cms))
+    #     producer_thread_fsm.start()
+    # except Exception as e:
+    #     compare_log.info('[fsm thread error] [error={0}]'.format(traceback.format_exc()))
+    
+    # with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
+    #     # 使用map函数提交多个任务
+    #     results = list(executor.map(fsm_compare_thread, application_id, application_entity, uniq_seq, ocr_res_id, is_ca, is_cms))
+    
+    compare_log.info('[fsm thread]')
+    #pool = ThreadPoolExecutor(max_workers=6, thread_name_prefix="fsm_thread_")
+    pool.submit(fsm_compare_thread, application_id, application_entity, uniq_seq, ocr_res_id, is_ca, is_cms)
+    #pool.shutdown(wait=True)
+    
+
+def fsm_compare_thread(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True, is_cms=False):
     compare_log.info('{0} [receive fsm task] [entity={1}] [id={2}] [uniq_seq={3}] [ocr_res_id={4}] [is_ca={5}] '
     compare_log.info('{0} [receive fsm task] [entity={1}] [id={2}] [uniq_seq={3}] [ocr_res_id={4}] [is_ca={5}] '
                      '[is_cms={6}]'.format(log_base, application_entity, application_id, uniq_seq, ocr_res_id,
                      '[is_cms={6}]'.format(log_base, application_entity, application_id, uniq_seq, ocr_res_id,
                                            is_ca, is_cms))
                                            is_ca, is_cms))
+
+    # 查看此订单号下是否有未完成的文件，如果有，等1分钟
+    doc_wait_file_class = HILDoc if application_entity == consts.HIL_PREFIX else AFCDoc
+    doc_wait_file_result = doc_wait_file_class.objects.filter(application_id=application_id, status=1).first()
+    compare_log.info('doc_wait_file_result:{0}'.format(doc_wait_file_result))
+    compare_log.info('{0} [comparison unfinished file check] [entity={1}] [id={2}] [doc_wait_file_result={3}]'.format(log_base, application_entity, application_id, doc_wait_file_result))
+    if doc_wait_file_result is not None:
+        # 实时查询延迟时间
+        try:
+            delay_time_config = Configs.objects.filter(id=4).first()
+            if delay_time_config is not None and delay_time_config.value is not None and delay_time_config.value.isdigit():
+                delay_time = delay_time_config.value
+            else:
+                delay_time = 0
+        except Exception as e:
+            delay_time = 0
+            compare_log.info('[get delay_time_config fail] [error={0}]'.format(traceback.format_exc()))
+        compare_log.info('delay_time:{0}'.format(delay_time))
+        compare_log.info('{0} [comparison unfinished file wait delay_time start] [entity={1}] [id={2}] [doc_id={3}]'.format(log_base, application_entity, application_id, doc_wait_file_result.id))
+        try:
+            time.sleep(int(delay_time))
+        except Exception as e:
+            compare_log.info('[sleep error] [error={0}]'.format(traceback.format_exc()))
+        compare_log.info('{0} [comparison unfinished file wait delay_time end] [entity={1}] [id={2}] [doc_id={3}]'.format(log_base, application_entity, application_id, doc_wait_file_result.id))
+
+
     # 调用java fsm 比对流程接口（http）
     # 调用java fsm 比对流程接口（http）
     # 调用Java fsm 比对流程接口, fsm 是se流程, ca可以暂时忽略
     # 调用Java fsm 比对流程接口, fsm 是se流程, ca可以暂时忽略
     auto_class = HILAutoSettlement if application_entity == consts.HIL_PREFIX else AFCAutoSettlement
     auto_class = HILAutoSettlement if application_entity == consts.HIL_PREFIX else AFCAutoSettlement
@@ -3729,6 +3797,22 @@ def fsm_compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=
 
 
 @app.task
 @app.task
 def compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True, is_cms=False):
 def compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True, is_cms=False):
+    # try:
+    #     producer_thread = threading.Thread(target=compare_thread, args=(application_id, application_entity, uniq_seq, ocr_res_id, is_ca, is_cms))
+    #     producer_thread.start()
+    # except Exception as e:
+    #     compare_log.info('[thread error] [error={0}]'.format(traceback.format_exc()))
+    
+    # with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
+    #     # 使用map函数提交多个任务
+    #     results = list(executor.map(compare_thread, application_id, application_entity, uniq_seq, ocr_res_id, is_ca, is_cms))
+
+    compare_log.info('[non fsm thread]')
+    #pool = ThreadPoolExecutor(max_workers=6, thread_name_prefix="non_fsm_thread_")
+    pool.submit(compare_thread, application_id, application_entity, uniq_seq, ocr_res_id, is_ca, is_cms)
+    #pool.shutdown(wait=True)
+
+def compare_thread(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True, is_cms=False):
     # POS: application_id, application_entity, uniq_seq, None
     # POS: application_id, application_entity, uniq_seq, None
     # OCR: application_id, business_type(application_entity), None, ocr_res_id
     # OCR: application_id, business_type(application_entity), None, ocr_res_id
 
 
@@ -3736,6 +3820,31 @@ def compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True
                      '[is_cms={6}]'.format(log_base, application_entity, application_id, uniq_seq, ocr_res_id,
                      '[is_cms={6}]'.format(log_base, application_entity, application_id, uniq_seq, ocr_res_id,
                                            is_ca, is_cms))
                                            is_ca, is_cms))
 
 
+    # 查看此订单号下是否有未完成的文件，如果有，等?分钟
+    doc_wait_file_class = HILDoc if application_entity == consts.HIL_PREFIX else AFCDoc
+    doc_wait_file_result = doc_wait_file_class.objects.filter(application_id=application_id, status=1).first()
+    compare_log.info('doc_wait_file_result:{0}'.format(doc_wait_file_result))
+    compare_log.info('{0} [comparison unfinished file check] [entity={1}] [id={2}] [doc_wait_file_result={3}]'.format(log_base, application_entity, application_id, doc_wait_file_result))
+    if doc_wait_file_result is not None:
+        # 实时查询延迟时间
+        try:
+            delay_time_config = Configs.objects.filter(id=4).first()
+            if delay_time_config is not None and delay_time_config.value is not None and delay_time_config.value.isdigit():
+                delay_time = delay_time_config.value
+            else:
+                delay_time = 0
+        except Exception as e:
+            delay_time = 0
+            compare_log.info('[get delay_time_config fail] [error={0}]'.format(traceback.format_exc()))
+        compare_log.info('delay_time:{0}'.format(delay_time))
+        compare_log.info('{0} [comparison unfinished file wait delay_time start] [entity={1}] [id={2}] [doc_id={3}]'.format(log_base, application_entity, application_id, doc_wait_file_result.id))
+        try:
+            time.sleep(int(delay_time))
+        except Exception as e:
+            compare_log.info('[sleep error] [error={0}]'.format(traceback.format_exc()))
+        compare_log.info('{0} [comparison unfinished file wait delay_time end] [entity={1}] [id={2}] [doc_id={3}]'.format(log_base, application_entity, application_id, doc_wait_file_result.id))
+
+
     # 根据application_id查找最新的比对信息，如果没有，结束
     # 根据application_id查找最新的比对信息，如果没有，结束
     if is_ca:
     if is_ca:
         comparison_class = HILComparisonInfo if application_entity == consts.HIL_PREFIX else AFCComparisonInfo
         comparison_class = HILComparisonInfo if application_entity == consts.HIL_PREFIX else AFCComparisonInfo