修复ai识别卡住的问题

BaseDataMaintenance/maintenance/dataflow.py

@@ -3018,7 +3018,7 @@ class Dataflow_dumplicate(Dataflow):
     def flow_dumpcate_comsumer(self):
         from multiprocessing import Process
         process_count = 6
-        thread_count = 12
+        thread_count = 3
         list_process = []
         def start_thread():
             mt = MultiThreadHandler(self.queue_dumplicate,self.dumplicate_comsumer_handle,None,thread_count,1,need_stop=False,restart=True,timeout=600,ots_client=self.ots_client)
@@ -3376,7 +3376,7 @@ class Dataflow_dumplicate(Dataflow):
 
         set_docid = set()
         list_delete_projects = []
-        list_projects = self.search_projects_with_document([docid])
+        list_projects = self.search_projects_with_document([docid],"project2","project2_index")
         for _proj in list_projects:
             _p = {}
             _docids = _proj.get(project_docids,"")
@@ -5514,7 +5514,7 @@ if __name__ == '__main__':
     # test_attachment_interface()
     df_dump = Dataflow_dumplicate(start_delete_listener=False)
     # df_dump.start_flow_dumplicate()
-    df_dump.test_dumplicate(638830991
+    df_dump.test_dumplicate(656253771
                             )
     # df_dump.dumplicate_comsumer_handle_interface(603504420,document_table="document_0000",document_table_index="document_0000_index",project_table="project_0000",project_table_index="project_0000_index_formerge")
     # compare_dumplicate_check()

BaseDataMaintenance/maintenance/dataflow_mq.py

@@ -1375,12 +1375,13 @@ class Dataflow_ActivteMQ_extract(Dataflow_extract):
         subscription = frame.headers.setdefault('subscription', None)
         item = json.loads(frame.body)
 
-        _extract_json = None
-        if document_extract2_extract_json in item:
-            _extract_json = item.get(document_extract2_extract_json)
-            item.pop(document_extract2_extract_json)
-
         try:
+            _extract_json = None
+            if document_extract2_extract_json in item:
+                _extract_json = item.get(document_extract2_extract_json)
+                item.pop(document_extract2_extract_json)
+
+
             message_acknowledged = False
             dtmp = Document_tmp(item)
 
@@ -1494,7 +1495,9 @@ class Dataflow_ActivteMQ_extract(Dataflow_extract):
             traceback.print_exc()
             if not message_acknowledged:
                 ackMsg(conn,message_id,subscription)
-
+            else:
+                send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract_ai)
+                ackMsg(conn,message_id,subscription)
 
     def merge_json(self,extract_json,extract_ai_json):
 

BaseDataMaintenance/model/ots/document_tmp.py

@@ -268,9 +268,9 @@ def turn_document_tmp_status():
             must_queries=[
                 # TermQuery("fingerprint","md5=2cc044b81ec13acddcc970b71b780365")
                 # TermQuery("save",0),
-                # RangeQuery("crtime","2025-03-05 09:30:00")
-                MatchPhraseQuery("doctitle","质量竣工验收监督")
-                # RangeQuery("status",1,51),
+                RangeQuery("crtime","2025-05-26 09:30:00","2025-07-26 12:30:00"),
+                # MatchPhraseQuery("doctitle","质量竣工验收监督")
+                RangeQuery("status",66,71),
                 # BoolQuery(should_queries=[
                 #                           # TermQuery("tenderee","山西利民工业有限责任公司"),
                 #                           # MatchPhraseQuery("doctitle","中国电信"),
@@ -372,8 +372,8 @@ def turn_document_tmp_status():
         # _extract_json = _extract_json.replace("\x06", "").replace("\x05", "").replace("\x07", "").replace('\\', '')
         # item.setValue(document_tmp_extract_json,_extract_json,True)
         # json.loads(_extract_json)
-        # item.setValue(document_tmp_status,0,True)
-        item.setValue(document_tmp_save,1,True)
+        item.setValue(document_tmp_status,1,True)
+        item.setValue(document_tmp_save,71,True)
         # if item.exists_row(ots_client):
         #     item.update_row(ots_client)
         # print(item.getProperties())