diff --git a/app/commands/file.py b/app/commands/file.py index ee007762..acd53a0a 100644 --- a/app/commands/file.py +++ b/app/commands/file.py @@ -195,7 +195,7 @@ def file_put(**kwargs): # noqa: C901 'attribute': attribute, } if source_file: - upload_event['source_id'] = src_file_info.get('id') + upload_event['source_id'] = src_file_info.get('id', '') item_ids = simple_upload(upload_event, num_of_thread=thread, output_path=output_path) diff --git a/app/resources/custom_help.py b/app/resources/custom_help.py index 938f5844..f3dd9f2c 100644 --- a/app/resources/custom_help.py +++ b/app/resources/custom_help.py @@ -6,9 +6,11 @@ class HelpPage: page = { 'update': { - 'version': '2.2.0', - '1': 'CLI supports to perform multi-threading upload for file/folders', - '2': 'CLI supports to perform resumable upload for single file', + 'version': '2.3.0', + '1': 'The logic of normal upload and resumble are splited. ' + 'add new command for resumable upload as `pilotcli file resume -r manifest.json`', + '2': 'The manifest file will be output for both file/folder upload', + '3': 'Optimize logic, input and error message', }, 'dataset': { 'DATASET_DOWNLOAD': 'Download a dataset or a particular version of a dataset.', diff --git a/app/services/file_manager/file_upload/file_upload.py b/app/services/file_manager/file_upload/file_upload.py index f2a4cdd0..bbfc8682 100644 --- a/app/services/file_manager/file_upload/file_upload.py +++ b/app/services/file_manager/file_upload/file_upload.py @@ -123,7 +123,7 @@ def simple_upload( # noqa: C901 create_folder_flag = upload_event.get('create_folder_flag', False) compress_zip = upload_event.get('compress_zip', False) regular_file = upload_event.get('regular_file', True) - source_id = upload_event.get('source_id', None) + source_id = upload_event.get('source_id', '') attribute = upload_event.get('attribute') mhandler.SrvOutPutHandler.start_uploading(input_path) @@ -163,7 +163,7 @@ def simple_upload( # noqa: C901 input_path = os.path.dirname(input_path) for file in upload_file_path: # first remove the input path from the file path - file_path_sub = file.replace(input_path + '/', '') + file_path_sub = file.replace(input_path + '/', '') if input_path else file object_path = os.path.join(target_folder, file_path_sub) # generate a placeholder for each file @@ -256,28 +256,44 @@ def resume_upload( ) # check files in manifest if some of them are already uploaded - item_ids = [] + unfinished_items = [] all_files = manifest_json.get('file_objects') + item_ids = [] for item_id in all_files: item_ids.append(item_id) - items = get_file_info_by_geid(item_ids) - unfinished_items = [] - for x in items: - if x.get('result').get('status') == ItemStatus.REGISTERED: - file_info = all_files.get(x.get('result').get('id')) - unfinished_items.append( - FileObject( - file_info.get('object_path'), - file_info.get('local_path'), - file_info.get('resumable_id'), - file_info.get('job_id'), - file_info.get('item_id'), + # here add the batch of 500 per loop, the pre upload api cannot + # process very large amount of file at same time. otherwise it will timeout + num_of_batchs = math.ceil(len(all_files) / AppConfig.Env.upload_batch_size) + # here is list of pre upload result. We decided to call pre upload api by batch + for batch in range(0, num_of_batchs): + start_index = batch * AppConfig.Env.upload_batch_size + end_index = (batch + 1) * AppConfig.Env.upload_batch_size + file_batchs = item_ids[start_index:end_index] + items = get_file_info_by_geid(file_batchs) + + # get the detail of item to see if the file is already uploaded + unfinished_files = [] + for x in items: + if x.get('result').get('status') == ItemStatus.REGISTERED: + file_info = all_files.get(x.get('result').get('id')) + unfinished_files.append( + FileObject( + file_info.get('object_path'), + file_info.get('local_path'), + file_info.get('resumable_id'), + file_info.get('job_id'), + file_info.get('item_id'), + ) ) - ) - # then for the rest of the files, check if any chunks are already uploaded - unfinished_items = upload_client.resume_upload(unfinished_items) + # then for the rest of the files, check if any chunks are already uploaded + mhandler.SrvOutPutHandler.resume_check_in_progress() + if len(unfinished_files) > 0: + unfinished_items.extend(upload_client.resume_upload(unfinished_files)) + + mhandler.SrvOutPutHandler.resume_warning(len(unfinished_items)) + mhandler.SrvOutPutHandler.resume_check_success() # lastly, start resumable upload for the rest of the chunks # thread number +1 reserve one thread to refresh token diff --git a/app/services/file_manager/file_upload/upload_client.py b/app/services/file_manager/file_upload/upload_client.py index c77a9c36..e189dd89 100644 --- a/app/services/file_manager/file_upload/upload_client.py +++ b/app/services/file_manager/file_upload/upload_client.py @@ -114,7 +114,6 @@ def resume_upload(self, unfinished_file_objects: List[FileObject]) -> List[FileO - local_path(str): the local path of file. - chunk_info(dict): the mapping for chunks that already been uploaded. """ - mhandler.SrvOutPutHandler.resume_warning(len(unfinished_file_objects)) headers = {'Authorization': 'Bearer ' + self.user.access_token, 'Session-ID': self.user.session_id} url = AppConfig.Connections.url_bff + f'/v1/project/{self.project_code}/files/resumable' @@ -142,7 +141,6 @@ def resume_upload(self, unfinished_file_objects: List[FileObject]) -> List[FileO file_obj = rid_file_object_map.get(uploaded_info.get('resumable_id')) # update the chunk info file_obj.uploaded_chunks = uploaded_info.get('chunks_info') - mhandler.SrvOutPutHandler.resume_check_success() return unfinished_file_objects diff --git a/app/services/output_manager/message_handler.py b/app/services/output_manager/message_handler.py index 65217687..2ef1dd89 100644 --- a/app/services/output_manager/message_handler.py +++ b/app/services/output_manager/message_handler.py @@ -133,6 +133,11 @@ def resume_check_success(): """e.g. notify the resumable check succeed.""" return logger.info('Resumable upload check complete.') + @staticmethod + def resume_check_in_progress(): + """e.g. notify the resumable check succeed.""" + return logger.info('Resumable upload check in progress.') + @staticmethod def resume_warning(num_of_files: int): """e.g. notify the user if they comfirm the resumable upload.""" diff --git a/pyproject.toml b/pyproject.toml index 4a942f44..773ed5a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "app" -version = "2.2.4" +version = "2.4.0a0" description = "This service is designed to support pilot platform" authors = ["Indoc Research"]