<commit_before>from ddsc.core.util import ProjectWalker, KindType
from ddsc.core.ddsapi import DataServiceAuth, DataServiceApi
from ddsc.core.fileuploader import FileUploader, FileUploadOperations, ParentData
from ddsc.core.parallel import TaskExecutor, TaskRunner


class UploadSettings(object):
    """
    Settings used to upload a project
    """
    def __init__(self, config, data_service, watcher, project_name_or_id, file_upload_post_processor):
        """
        :param config: ddsc.config.Config user configuration settings from YAML file/environment
        :param data_service: DataServiceApi: where we will upload to
        :param watcher: ProgressPrinter we notify of our progress
        :param project_name_or_id: ProjectNameOrId: name or id of the project so we can create it if necessary
        :param file_upload_post_processor: object: has run(data_service, file_response) method to run after download
        """
        self.config = config
        self.data_service = data_service
        self.watcher = watcher
        self.project_name_or_id = project_name_or_id
        self.project_id = None
        self.file_upload_post_processor = file_upload_post_processor

    def get_data_service_auth_data(self):
        """
        Serialize data_service setup into something that can be passed to another process.
        :return: tuple of data service settings
        """
        return self.data_service.auth.get_auth_data()

    @staticmethod
    def rebuild_data_service(config, data_service_auth_data):
        """
        Deserialize value into DataServiceApi object.
        :param config:
        :param data_service_auth_data:
        :return:
        """
        auth = DataServiceAuth(config)
        auth.set_auth_data(data_service_auth_data)
        return DataServiceApi(auth, config.url)


class UploadContext(object):
    """
    Values passed to a background worker.
    Contains UploadSettings and parameters specific to the function to be run.
    """
    def __init__(self, settings, params, message_queue, task_id):
        """
        Setup context so it can be passed.
        :param settings: UploadSettings: project level info
        :param params: tuple: values specific to the function being run
        :param message_queue: Queue: queue background process can send messages to us on
        :param task_id: int: id of this command's task so message will be routed correctly
        """
        self.data_service_auth_data = settings.get_data_service_auth_data()
        self.config = settings.config
        self.project_name_or_id = settings.project_name_or_id
        self.project_id = settings.project_id
        self.params = params
        self.message_queue = message_queue
        self.task_id = task_id

    def make_data_service(self):
        """
        Recreate data service from within background worker.
        :return: DataServiceApi
        """
        return UploadSettings.rebuild_data_service(self.config, self.data_service_auth_data)

    def send_message(self, data):
        """
        Sends a message to the command's on_message(data) method.
        :param data: object: data sent to on_message
        """
        self.message_queue.put((self.task_id, data))

    def start_waiting(self):
        """
        Called when we start waiting for project to be ready for file uploads.
        """
        self.send_message(True)

    def done_waiting(self):
        """
        Called when project is ready for file uploads (after waiting).
        """
        self.send_message(False)


class ProjectUploader(object):
    """
    Uploads a project based on UploadSettings.
    """
    def __init__(self, settings):
        """
        Setup to talk to the data service based on settings.
        :param settings: UploadSettings: settings to use for uploading.
        """
        self.runner = TaskRunner(TaskExecutor(settings.config.upload_workers))
        self.settings = settings
        self.small_item_task_builder = SmallItemUploadTaskBuilder(self.settings, self.runner)
        self.small_items = []
        self.large_items = []

    def run(self, local_project):
        """
        Upload a project by uploading project, folders, and small files then uploading the large files.
        :param local_project: LocalProject: project to upload
        """
        # Walk project adding small items to runner saving large items to large_items
        ProjectWalker.walk_project(local_project, self)
        # Run small items in parallel
        self.runner.run()
        # Run parts of each large item in parallel
        self.upload_large_items()

    # Methods called by ProjectWalker.walk_project
    def visit_project(self, item):
        """
        Add create project to small task list.
        """
        self.small_item_task_builder.visit_project(item)

    def visit_folder(self, item, parent):
        """
        Add create folder to small task list.
        """
        self.small_item_task_builder.visit_folder(item, parent)

    def visit_file(self, item, parent):
        """
        If file is large add it to the large items to be processed after small task list.
        else file is small add it to the small task list.
        """
        if self.is_large_file(item):
            self.large_items.append((item, parent))
        else:
            self.small_item_task_builder.visit_file(item, parent)

    def is_large_file(self, item):
        return item.size > self.settings.config.upload_bytes_per_chunk

    def upload_large_items(self):
        """
        Upload files that were too large.
        """
        for local_file, parent in self.large_items:
            hash_data = local_file.calculate_local_hash()
            if local_file.hash_matches_remote(hash_data):
                self.file_already_uploaded(local_file)
            else:
                self.process_large_file(local_file, parent, hash_data)

    def process_large_file(self, local_file, parent, hash_data):
        """
        Upload a single file using multiple processes to upload multiple chunks at the same time.
        Updates local_file with it's remote_id when done.
        :param local_file: LocalFile: file we are uploading
        :param parent: LocalFolder/LocalProject: parent of the file
        """
        file_content_sender = FileUploader(self.settings.config, self.settings.data_service, local_file, hash_data,
                                           self.settings.watcher, self.settings.file_upload_post_processor)
        remote_id = file_content_sender.upload(self.settings.project_id, parent.kind, parent.remote_id)
        local_file.set_remote_id_after_send(remote_id)

    def file_already_uploaded(self, local_file):
        """
        Updates progress bar for a file that was already uploaded
        :param local_file: LocalFile
        """
        self.settings.watcher.transferring_item(local_file, increment_amt=local_file.size)


class SmallItemUploadTaskBuilder(object):<commit_msg>fix large file already uploaded progress bar<commit_after>from ddsc.core.util import ProjectWalker, KindType
from ddsc.core.ddsapi import DataServiceAuth, DataServiceApi
from ddsc.core.fileuploader import FileUploader, FileUploadOperations, ParentData, ParallelChunkProcessor
from ddsc.core.parallel import TaskExecutor, TaskRunner


class UploadSettings(object):
    """
    Settings used to upload a project
    """
    def __init__(self, config, data_service, watcher, project_name_or_id, file_upload_post_processor):
        """
        :param config: ddsc.config.Config user configuration settings from YAML file/environment
        :param data_service: DataServiceApi: where we will upload to
        :param watcher: ProgressPrinter we notify of our progress
        :param project_name_or_id: ProjectNameOrId: name or id of the project so we can create it if necessary
        :param file_upload_post_processor: object: has run(data_service, file_response) method to run after download
        """
        self.config = config
        self.data_service = data_service
        self.watcher = watcher
        self.project_name_or_id = project_name_or_id
        self.project_id = None
        self.file_upload_post_processor = file_upload_post_processor

    def get_data_service_auth_data(self):
        """
        Serialize data_service setup into something that can be passed to another process.
        :return: tuple of data service settings
        """
        return self.data_service.auth.get_auth_data()

    @staticmethod
    def rebuild_data_service(config, data_service_auth_data):
        """
        Deserialize value into DataServiceApi object.
        :param config:
        :param data_service_auth_data:
        :return:
        """
        auth = DataServiceAuth(config)
        auth.set_auth_data(data_service_auth_data)
        return DataServiceApi(auth, config.url)


class UploadContext(object):
    """
    Values passed to a background worker.
    Contains UploadSettings and parameters specific to the function to be run.
    """
    def __init__(self, settings, params, message_queue, task_id):
        """
        Setup context so it can be passed.
        :param settings: UploadSettings: project level info
        :param params: tuple: values specific to the function being run
        :param message_queue: Queue: queue background process can send messages to us on
        :param task_id: int: id of this command's task so message will be routed correctly
        """
        self.data_service_auth_data = settings.get_data_service_auth_data()
        self.config = settings.config
        self.project_name_or_id = settings.project_name_or_id
        self.project_id = settings.project_id
        self.params = params
        self.message_queue = message_queue
        self.task_id = task_id

    def make_data_service(self):
        """
        Recreate data service from within background worker.
        :return: DataServiceApi
        """
        return UploadSettings.rebuild_data_service(self.config, self.data_service_auth_data)

    def send_message(self, data):
        """
        Sends a message to the command's on_message(data) method.
        :param data: object: data sent to on_message
        """
        self.message_queue.put((self.task_id, data))

    def start_waiting(self):
        """
        Called when we start waiting for project to be ready for file uploads.
        """
        self.send_message(True)

    def done_waiting(self):
        """
        Called when project is ready for file uploads (after waiting).
        """
        self.send_message(False)


class ProjectUploader(object):
    """
    Uploads a project based on UploadSettings.
    """
    def __init__(self, settings):
        """
        Setup to talk to the data service based on settings.
        :param settings: UploadSettings: settings to use for uploading.
        """
        self.runner = TaskRunner(TaskExecutor(settings.config.upload_workers))
        self.settings = settings
        self.small_item_task_builder = SmallItemUploadTaskBuilder(self.settings, self.runner)
        self.small_items = []
        self.large_items = []

    def run(self, local_project):
        """
        Upload a project by uploading project, folders, and small files then uploading the large files.
        :param local_project: LocalProject: project to upload
        """
        # Walk project adding small items to runner saving large items to large_items
        ProjectWalker.walk_project(local_project, self)
        # Run small items in parallel
        self.runner.run()
        # Run parts of each large item in parallel
        self.upload_large_items()

    # Methods called by ProjectWalker.walk_project
    def visit_project(self, item):
        """
        Add create project to small task list.
        """
        self.small_item_task_builder.visit_project(item)

    def visit_folder(self, item, parent):
        """
        Add create folder to small task list.
        """
        self.small_item_task_builder.visit_folder(item, parent)

    def visit_file(self, item, parent):
        """
        If file is large add it to the large items to be processed after small task list.
        else file is small add it to the small task list.
        """
        if self.is_large_file(item):
            self.large_items.append((item, parent))
        else:
            self.small_item_task_builder.visit_file(item, parent)

    def is_large_file(self, item):
        return item.size > self.settings.config.upload_bytes_per_chunk

    def upload_large_items(self):
        """
        Upload files that were too large.
        """
        for local_file, parent in self.large_items:
            hash_data = local_file.calculate_local_hash()
            if local_file.hash_matches_remote(hash_data):
                self.file_already_uploaded(local_file)
            else:
                self.process_large_file(local_file, parent, hash_data)

    def process_large_file(self, local_file, parent, hash_data):
        """
        Upload a single file using multiple processes to upload multiple chunks at the same time.
        Updates local_file with it's remote_id when done.
        :param local_file: LocalFile: file we are uploading
        :param parent: LocalFolder/LocalProject: parent of the file
        """
        file_content_sender = FileUploader(self.settings.config, self.settings.data_service, local_file, hash_data,
                                           self.settings.watcher, self.settings.file_upload_post_processor)
        remote_id = file_content_sender.upload(self.settings.project_id, parent.kind, parent.remote_id)
        local_file.set_remote_id_after_send(remote_id)

    def file_already_uploaded(self, local_file):
        """
        Updates progress bar for a file that was already uploaded
        :param local_file: LocalFile
        """
        num_chunks = ParallelChunkProcessor.determine_num_chunks(self.settings.config.upload_bytes_per_chunk,
                                                                 local_file.size)
        self.settings.watcher.transferring_item(local_file, increment_amt=num_chunks)


class SmallItemUploadTaskBuilder(object):
{'fraction_non_alphanumeric': 0.07092745895888873, 'fraction_numerical': 0.0, 'mean_word_length': 2.6825109790751744, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4711134', 'n_tokens_mistral': 3861, 'n_tokens_neox': 3765, 'n_words': 1113}