imporve task_id (#430)

Author: likesclever

dtable_events/dtable_io/task_big_data_manager.py

@@ -1,4 +1,5 @@
 import time
+import uuid
 import queue
 import threading
 
@@ -54,6 +55,9 @@ def handle_task(self):
 
             try:
                 task = self.tasks_map[task_id]
+                if type(task[0]).__name__ != 'function':
+                    continue
+
                 self.current_task_info = task_id + ' ' + str(task[0])
                 dtable_io_logger.info('Run task: %s' % self.current_task_info)
                 start_time = time.time()
@@ -73,7 +77,7 @@ def handle_task(self):
 
     def add_import_big_excel_task(self, username, dtable_uuid, table_name, file_path):
         from dtable_events.dtable_io import import_big_excel
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_big_excel,
                 (username, dtable_uuid, table_name, file_path, task_id, self.tasks_status_map))
         self.tasks_queue.put(task_id)
@@ -82,7 +86,7 @@ def add_import_big_excel_task(self, username, dtable_uuid, table_name, file_path
 
     def add_update_big_excel_task(self, username, dtable_uuid, table_name, file_path, ref_columns, is_insert_new_data=False):
         from dtable_events.dtable_io import update_big_excel
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (update_big_excel,
                 (username, dtable_uuid, table_name, file_path, ref_columns, is_insert_new_data, task_id, self.tasks_status_map))
         self.tasks_queue.put(task_id)
@@ -92,7 +96,7 @@ def add_update_big_excel_task(self, username, dtable_uuid, table_name, file_path
     def add_convert_big_data_view_to_execl_task(self, dtable_uuid, table_id, view_id, username, name, repo_id, is_support_image):
         from dtable_events.dtable_io import convert_big_data_view_to_execl
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (convert_big_data_view_to_execl,
                 (dtable_uuid, table_id, view_id, username, name, task_id, self.tasks_status_map, repo_id, is_support_image))
         self.tasks_queue.put(task_id)

dtable_events/dtable_io/task_data_sync_manager.py

@@ -1,6 +1,7 @@
 import queue
 import threading
 import time
+import uuid
 
 
 class TaskDataSyncManager(object):
@@ -25,7 +26,7 @@ def is_valid_task_id(self, task_id):
     def add_sync_email_task(self, context):
         from dtable_events.dtable_io import email_sync
 
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (email_sync, (context, self.config))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -53,6 +54,8 @@ def handle_task(self):
 
             try:
                 task = self.tasks_map[task_id]
+                if type(task[0]).__name__ != 'function':
+                    continue
                 task_info = task_id + ' ' + str(task[0])
                 self.current_task_info[task_id] = task_info
                 dtable_data_sync_logger.info('Run task: %s' % task_info)

dtable_events/dtable_io/task_manager.py

@@ -2,6 +2,7 @@
 import queue
 import threading
 import time
+import uuid
 from threading import Lock
 
 
@@ -38,7 +39,7 @@ def add_export_task(self, username, repo_id, workspace_id, dtable_uuid, dtable_n
             asset_dir_path = os.path.join('/asset', dtable_uuid)
             asset_dir_id = seafile_api.get_dir_id_by_path(repo_id, asset_dir_path)
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (get_dtable_export_content,
                 (username, repo_id, workspace_id, dtable_uuid, asset_dir_id, self.config))
         self.tasks_queue.put(task_id)
@@ -50,7 +51,7 @@ def add_import_task(self, username, repo_id, workspace_id, dtable_uuid, dtable_f
                         can_use_automation_rules, can_use_workflows, can_use_external_apps, owner, org_id):
         from dtable_events.dtable_io import post_dtable_import_files
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (post_dtable_import_files,
                 (username, repo_id, workspace_id, dtable_uuid, dtable_file_name, in_storage,
                  can_use_automation_rules, can_use_workflows, can_use_external_apps, owner, org_id, self.config))
@@ -61,7 +62,7 @@ def add_import_task(self, username, repo_id, workspace_id, dtable_uuid, dtable_f
     def add_export_dtable_asset_files_task(self, username, repo_id, dtable_uuid, files, files_map=None):
         from dtable_events.dtable_io import get_dtable_export_asset_files
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (get_dtable_export_asset_files,
                 (username, repo_id, dtable_uuid, files, task_id, files_map))
         self.tasks_queue.put(task_id)
@@ -70,7 +71,7 @@ def add_export_dtable_asset_files_task(self, username, repo_id, dtable_uuid, fil
 
     def add_transfer_dtable_asset_files_task(self, username, repo_id, dtable_uuid, files, files_map, parent_dir, relative_path, replace, repo_api_token, seafile_server_url):
         from dtable_events.dtable_io import get_dtable_transfer_asset_files
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (get_dtable_transfer_asset_files,
                 (username,
                  repo_id,
@@ -90,7 +91,7 @@ def add_transfer_dtable_asset_files_task(self, username, repo_id, dtable_uuid, f
     def add_parse_excel_csv_task(self, username, repo_id, workspace_id, dtable_name, file_type, custom):
         from dtable_events.dtable_io import parse_excel_csv
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (parse_excel_csv,
                 (username, repo_id, workspace_id, dtable_name, file_type, custom, self.config))
         self.tasks_queue.put(task_id)
@@ -100,7 +101,7 @@ def add_parse_excel_csv_task(self, username, repo_id, workspace_id, dtable_name,
     def add_import_excel_csv_task(self, username, repo_id, workspace_id, dtable_uuid, dtable_name, lang):
         from dtable_events.dtable_io import import_excel_csv
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_excel_csv,
                 (username, repo_id, workspace_id, dtable_uuid, dtable_name, lang, self.config))
         self.tasks_queue.put(task_id)
@@ -110,7 +111,7 @@ def add_import_excel_csv_task(self, username, repo_id, workspace_id, dtable_uuid
     def add_import_excel_csv_add_table_task(self, username, repo_id, workspace_id, dtable_uuid, dtable_name, lang):
         from dtable_events.dtable_io import import_excel_csv_add_table
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_excel_csv_add_table,
                 (username, repo_id, workspace_id, dtable_uuid, dtable_name, lang, self.config))
         self.tasks_queue.put(task_id)
@@ -120,7 +121,7 @@ def add_import_excel_csv_add_table_task(self, username, repo_id, workspace_id, d
     def add_append_excel_csv_append_parsed_file_task(self, username, repo_id, dtable_uuid, file_name, table_name):
         from dtable_events.dtable_io import append_excel_csv_append_parsed_file
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (append_excel_csv_append_parsed_file,
                 (username, repo_id, dtable_uuid, file_name, table_name))
         self.tasks_queue.put(task_id)
@@ -130,7 +131,7 @@ def add_append_excel_csv_append_parsed_file_task(self, username, repo_id, dtable
     def add_append_excel_csv_upload_file_task(self, username, repo_id, file_name, dtable_uuid, table_name, file_type):
         from dtable_events.dtable_io import append_excel_csv_upload_file
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (append_excel_csv_upload_file,
                 (username, repo_id, file_name, dtable_uuid, table_name, file_type))
         self.tasks_queue.put(task_id)
@@ -139,7 +140,7 @@ def add_append_excel_csv_upload_file_task(self, username, repo_id, file_name, dt
 
     def add_run_auto_rule_task(self, automation_rule_id, username, org_id, dtable_uuid, run_condition, trigger, actions):
         from dtable_events.automations.auto_rules_utils import run_auto_rule_task
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         options = {
             'run_condition': run_condition,
             'dtable_uuid': dtable_uuid,
@@ -157,7 +158,7 @@ def add_update_excel_csv_update_parsed_file_task(self, username, repo_id, dtable
                                                      selected_columns):
         from dtable_events.dtable_io import update_excel_csv_update_parsed_file
 
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (update_excel_csv_update_parsed_file,
                 (username, repo_id, dtable_uuid, file_name, table_name, selected_columns))
         self.tasks_queue.put(task_id)
@@ -167,7 +168,7 @@ def add_update_excel_csv_update_parsed_file_task(self, username, repo_id, dtable
     def add_update_excel_upload_excel_task(self, username, repo_id, file_name, dtable_uuid, table_name):
         from dtable_events.dtable_io import update_excel_upload_excel
 
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (update_excel_upload_excel,
                 (username, repo_id, file_name, dtable_uuid, table_name))
         self.tasks_queue.put(task_id)
@@ -177,7 +178,7 @@ def add_update_excel_upload_excel_task(self, username, repo_id, file_name, dtabl
     def add_update_csv_upload_csv_task(self, username, repo_id, file_name, dtable_uuid, table_name):
         from dtable_events.dtable_io import update_csv_upload_csv
 
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (update_csv_upload_csv,
                 (username, repo_id, file_name, dtable_uuid, table_name))
         self.tasks_queue.put(task_id)
@@ -187,7 +188,7 @@ def add_update_csv_upload_csv_task(self, username, repo_id, file_name, dtable_uu
     def add_import_excel_csv_to_dtable_task(self, username, repo_id, workspace_id, dtable_name, dtable_uuid, file_type, lang):
         from dtable_events.dtable_io import import_excel_csv_to_dtable
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_excel_csv_to_dtable, (username, repo_id, workspace_id, dtable_name, dtable_uuid, file_type, lang))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -196,7 +197,7 @@ def add_import_excel_csv_to_dtable_task(self, username, repo_id, workspace_id, d
     def add_import_excel_csv_to_table_task(self, username, repo_id, workspace_id, file_name, dtable_uuid, file_type, lang):
         from dtable_events.dtable_io import import_excel_csv_to_table
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_excel_csv_to_table, (username, repo_id, workspace_id, file_name, dtable_uuid, file_type, lang))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -205,7 +206,7 @@ def add_import_excel_csv_to_table_task(self, username, repo_id, workspace_id, fi
     def add_update_table_via_excel_csv_task(self, username, repo_id, file_name, dtable_uuid, table_name, selected_columns, file_type):
         from dtable_events.dtable_io import update_table_via_excel_csv
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (update_table_via_excel_csv, (username, repo_id, file_name, dtable_uuid, table_name, selected_columns, file_type))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -214,7 +215,7 @@ def add_update_table_via_excel_csv_task(self, username, repo_id, file_name, dtab
     def add_append_excel_csv_to_table_task(self, username, repo_id, file_name, dtable_uuid, table_name, file_type):
         from dtable_events.dtable_io import append_excel_csv_to_table
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (append_excel_csv_to_table, (username, repo_id, file_name, dtable_uuid, table_name, file_type))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -233,7 +234,7 @@ def query_status(self, task_id):
     def convert_page_to_pdf(self, dtable_uuid, page_id, row_id, access_token, session_id):
         from dtable_events.dtable_io import convert_page_to_pdf
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (convert_page_to_pdf,
                 (dtable_uuid, page_id, row_id, access_token, session_id))
         self.tasks_queue.put(task_id)
@@ -244,7 +245,7 @@ def convert_page_to_pdf(self, dtable_uuid, page_id, row_id, access_token, sessio
     def add_import_table_from_base_task(self, context):
         from dtable_events.dtable_io.import_table_from_base import import_table_from_base
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_table_from_base, (context,))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -254,7 +255,7 @@ def add_import_table_from_base_task(self, context):
     def add_import_common_dataset_task(self, context):
         from dtable_events.dtable_io.import_sync_common_dataset import import_common_dataset
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (import_common_dataset, (context, self.config))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -273,7 +274,7 @@ def add_sync_common_dataset_task(self, context):
                 return None, 'syncing'
             self.dataset_sync_ids.add(dataset_sync_id)
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (sync_common_dataset, (context, self.config))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -283,7 +284,7 @@ def add_sync_common_dataset_task(self, context):
     def add_convert_view_to_execl_task(self, dtable_uuid, table_id, view_id, username, id_in_org, permission, name, repo_id, is_support_image):
         from dtable_events.dtable_io import convert_view_to_execl
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (convert_view_to_execl, (dtable_uuid, table_id, view_id, username, id_in_org, permission, name, repo_id, is_support_image))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -293,7 +294,7 @@ def add_convert_view_to_execl_task(self, dtable_uuid, table_id, view_id, usernam
     def add_convert_table_to_execl_task(self, dtable_uuid, table_id, username, permission, name, repo_id, is_support_image):
         from dtable_events.dtable_io import convert_table_to_execl
 
-        task_id = str(int(time.time()*1000))
+        task_id = str(uuid.uuid4())
         task = (convert_table_to_execl, (dtable_uuid, table_id, username, permission, name, repo_id, is_support_image))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -302,7 +303,7 @@ def add_convert_table_to_execl_task(self, dtable_uuid, table_id, username, permi
 
     def add_app_users_sync_task(self, dtable_uuid, app_name, app_id, table_name, table_id, username):
         from dtable_events.dtable_io import app_user_sync
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (app_user_sync, (dtable_uuid, app_name, app_id, table_name, table_id, username, self.config))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -329,6 +330,9 @@ def handle_task(self):
 
             task = self.tasks_map.get(task_id)
             try:
+                if type(task[0]).__name__ != 'function':
+                    continue
+
                 task_info = task_id + ' ' + str(task[0])
                 self.current_task_info[task_id] = task_info
                 dtable_io_logger.info('Run task: %s' % task_info)

dtable_events/dtable_io/task_message_manager.py

@@ -4,6 +4,7 @@
 import sys
 import threading
 import time
+import uuid
 
 from dtable_events.app.config import DTABLE_WEB_SERVICE_URL, DTABLE_PRIVATE_KEY, DTABLE_SERVER_URL
 
@@ -31,23 +32,23 @@ def is_valid_task_id(self, task_id):
 
     def add_email_sending_task(self, auth_info, send_info, username):
         from dtable_events.dtable_io import send_email_msg
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (send_email_msg,(auth_info, send_info, username, self.config))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
         return task_id
 
     def add_wechat_sending_task(self, webhook_url, msg, msg_type):
         from dtable_events.dtable_io import send_wechat_msg
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (send_wechat_msg, (webhook_url, msg, msg_type))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
         return task_id
 
     def add_dingtalk_sending_task(self, webhook_url, msg ):
         from dtable_events.dtable_io import send_dingtalk_msg
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (send_dingtalk_msg, (webhook_url, msg))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -76,6 +77,9 @@ def handle_task(self):
 
             try:
                 task = self.tasks_map[task_id]
+                if type(task[0]).__name__ != 'function':
+                    continue
+
                 self.current_task_info = task_id + ' ' + str(task[0])
                 dtable_message_logger.info('Run task: %s' % self.current_task_info)
                 start_time = time.time()

dtable_events/dtable_io/task_plugin_email_manager.py

@@ -1,3 +1,4 @@
+import uuid
 import queue
 import threading
 import time
@@ -25,7 +26,7 @@ def is_valid_task_id(self, task_id):
     def add_send_email_task(self, context):
         from dtable_events.dtable_io import plugin_email_send_email
 
-        task_id = str(int(time.time() * 1000))
+        task_id = str(uuid.uuid4())
         task = (plugin_email_send_email, (context, self.config))
         self.tasks_queue.put(task_id)
         self.tasks_map[task_id] = task
@@ -53,6 +54,9 @@ def handle_task(self):
 
             try:
                 task = self.tasks_map[task_id]
+                if type(task[0]).__name__ != 'function':
+                    continue
+
                 task_info = task_id + ' ' + str(task[0])
                 self.current_task_info[task_id] = task_info
                 dtable_plugin_email_logger.info('Run task: %s' % task_info)