<commit_before>"""
Objects to upload a number of chunks from a file to a remote store as part of an upload.
"""

import math
from multiprocessing import Process, Queue
from ddsc.core.localstore import HashUtil
from ddsc.core.ddsapi import DataServiceAuth, DataServiceApi
from ddsc.core.util import ProgressQueue, wait_for_processes

class FileUploader(object):
    """
    Handles sending the contents of a file to a a remote data_service.
    Process:
    1) It creates an 'upload' with the remote service
    2) Uses a chunk_processor to send the parts
    3) Sends the complete message to finalize the 'upload'
    4) Sends create_file message to remote store with the 'upload' id
    """
    def __init__(self, config, data_service, local_file, watcher):
        """
        Setup for sending to remote store.
        :param config: ddsc.config.Config user configuration settings from YAML file/environment
        :param data_service: DataServiceApi data service we are sending the content to.
        :param local_file: LocalFile file we are sending to remote store
        :param watcher: ProgressPrinter we notify of our progress
        """
        self.config = config
        self.data_service = data_service
        self.local_file = local_file
        self.upload_id = None
        self.watcher = watcher

    def _make_chunk_processor(self):
        """
        Based on the passed in config determine if we should use parallel chunk processor or serial one.
        """
        if self.config.upload_workers:
            if self.config.upload_workers > 1:
                return ParallelChunkProcessor(self)
        return SerialChunkProcessor(self)

    def upload(self, project_id, parent_kind, parent_id):
        """
        Upload file contents to project within a specified parent.
        :param project_id: str project uuid
        :param parent_kind: str type of parent ('dds-project' or 'dds-folder')
        :param parent_id: str uuid of parent
        :return: str uuid of the newly uploaded file
        """
        size = self.local_file.size
        (hash_alg, hash_value) = self.local_file.get_hashpair()
        name = self.local_file.name
        resp = self.data_service.create_upload(project_id, name, self.local_file.mimetype, size, hash_value, hash_alg)
        self.upload_id = resp.json()['id']
        chunk_processor = self._make_chunk_processor()
        chunk_processor.run()
        self.data_service.complete_upload(self.upload_id)
        if self.local_file.remote_id:
            file_id = self.local_file.remote_id
            self.data_service.update_file(file_id, self.upload_id)
            return file_id
        else:
            result = self.data_service.create_file(parent_kind, parent_id, self.upload_id)
            return result.json()['id']

    @staticmethod
    def send_file_external(data_service, url_json, chunk):
        """
        Send chunk to external store specified in url_json.
        Raises ValueError on upload failure.
        :param data_service: data service to use for sending chunk
        :param url_json: dict contains where/how to upload chunk
        :param chunk: data to be uploaded
        """
        http_verb = url_json['http_verb']
        host = url_json['host']
        url = url_json['url']
        http_headers = url_json['http_headers']
        resp = data_service.send_external(http_verb, host, url, http_headers, chunk)
        if resp.status_code != 200 and resp.status_code != 201:
            raise ValueError("Failed to send file to external store. Error:" + str(resp.status_code))

    @staticmethod
    def hash_chunk(chunk):
        """Creates a hash from the bytes in chunk."""
        hash = HashUtil()
        hash.add_chunk(chunk)
        return hash.hexdigest()


class SerialChunkProcessor(object):
    """
    Uploads chunks one at a time without any additional processes.
    """
    def __init__(self, file_uploader):
        """
        Send chunks in the file specified in file_uploader to the remote data service.
        :param file_uploader: FileUploader contains all data we need to upload chunks of a file.
        """
        self.local_file = file_uploader.local_file
        self.bytes_per_chunk = file_uploader.config.upload_bytes_per_chunk
        self.data_service = file_uploader.data_service
        self.watcher = file_uploader.watcher
        self.upload_id = file_uploader.upload_id
        self.chunk_num = 0

    def run(self):
        """
        Sends contents of a local file to a remote data service.
        """
        processor = self.process_chunk
        if self.local_file.size == 0:
            chunk = ''
            (chunk_hash_alg, chunk_hash_value) = FileUploader.hash_chunk(chunk)
            self.process_chunk(chunk, chunk_hash_alg, chunk_hash_value)
        else:
            with open(self.local_file.path, 'rb') as infile:
                number = 0
                for chunk in SerialChunkProcessor.read_in_chunks(infile, self.bytes_per_chunk):
                    (chunk_hash_alg, chunk_hash_value) = FileUploader.hash_chunk(chunk)
                    self.process_chunk(chunk, chunk_hash_alg, chunk_hash_value)

    def process_chunk(self, chunk, chunk_hash_alg, chunk_hash_value):
        """
        Creates 'upload' url and send bytes to that url.
        Raises ValueError on upload failure.
        :param chunk: bytes part of the file to send
        :param chunk_hash_alg: str the algorithm used to hash chunk
        :param chunk_hash_value: str the hash value of chunk
        """
        self.watcher.transferring_item(self.local_file)
        resp = self.data_service.create_upload_url(self.upload_id, self.chunk_num, len(chunk),
                                                   chunk_hash_value, chunk_hash_alg)
        if resp.status_code == 200:
            FileUploader.send_file_external(self.data_service, resp.json(), chunk)
            self.chunk_num += 1
        else:
            raise ValueError("Failed to retrieve upload url status:" + str(resp.status_code))

    @staticmethod
    def read_in_chunks(infile, blocksize):
        """
        Generator to read chunks lazily.
        :param infile: filehandle file to read from
        :param blocksize: int size of blocks to read
        """
        """"""
        while True:
            data = infile.read(blocksize)
            if not data:
                break
            yield data


class ParallelChunkProcessor(object):
    """
    Uploads grouped chunks in separate processes.
    """
    def __init__(self, file_uploader):
        """
        Send chunks in the file specified in file_uploader to the remote data service using multiple processes.
        :param file_uploader: FileUploader contains all data we need to upload chunks of a file.
        """
        self.config = file_uploader.config
        self.data_service = file_uploader.data_service
        self.upload_id = file_uploader.upload_id
        self.watcher = file_uploader.watcher
        self.local_file = file_uploader.local_file

    def run(self):
        """
        Sends contents of a local file to a remote data service.
        """
        processes = []
        progress_queue = ProgressQueue(Queue())
        num_chunks = ParallelChunkProcessor.determine_num_chunks(self.config.upload_bytes_per_chunk,
                                                                 self.local_file.size)
        work_parcels = ParallelChunkProcessor.make_work_parcels(self.config.upload_workers, num_chunks)
        for (index, num_items) in work_parcels:
            processes.append(self.make_and_start_process(index, num_items, progress_queue))
        wait_for_processes(processes, num_chunks, progress_queue, self.watcher, self.local_file)

    @staticmethod
    def determine_num_chunks(chunk_size, file_size):
        """
        Figure out how many pieces we are sending the file in.
        NOTE: duke-data-service requires an emtpy chunk to be uploaded for empty files.
        """<commit_msg>fix comment spelling<commit_after>"""
Objects to upload a number of chunks from a file to a remote store as part of an upload.
"""

import math
from multiprocessing import Process, Queue
from ddsc.core.localstore import HashUtil
from ddsc.core.ddsapi import DataServiceAuth, DataServiceApi
from ddsc.core.util import ProgressQueue, wait_for_processes

class FileUploader(object):
    """
    Handles sending the contents of a file to a a remote data_service.
    Process:
    1) It creates an 'upload' with the remote service
    2) Uses a chunk_processor to send the parts
    3) Sends the complete message to finalize the 'upload'
    4) Sends create_file message to remote store with the 'upload' id
    """
    def __init__(self, config, data_service, local_file, watcher):
        """
        Setup for sending to remote store.
        :param config: ddsc.config.Config user configuration settings from YAML file/environment
        :param data_service: DataServiceApi data service we are sending the content to.
        :param local_file: LocalFile file we are sending to remote store
        :param watcher: ProgressPrinter we notify of our progress
        """
        self.config = config
        self.data_service = data_service
        self.local_file = local_file
        self.upload_id = None
        self.watcher = watcher

    def _make_chunk_processor(self):
        """
        Based on the passed in config determine if we should use parallel chunk processor or serial one.
        """
        if self.config.upload_workers:
            if self.config.upload_workers > 1:
                return ParallelChunkProcessor(self)
        return SerialChunkProcessor(self)

    def upload(self, project_id, parent_kind, parent_id):
        """
        Upload file contents to project within a specified parent.
        :param project_id: str project uuid
        :param parent_kind: str type of parent ('dds-project' or 'dds-folder')
        :param parent_id: str uuid of parent
        :return: str uuid of the newly uploaded file
        """
        size = self.local_file.size
        (hash_alg, hash_value) = self.local_file.get_hashpair()
        name = self.local_file.name
        resp = self.data_service.create_upload(project_id, name, self.local_file.mimetype, size, hash_value, hash_alg)
        self.upload_id = resp.json()['id']
        chunk_processor = self._make_chunk_processor()
        chunk_processor.run()
        self.data_service.complete_upload(self.upload_id)
        if self.local_file.remote_id:
            file_id = self.local_file.remote_id
            self.data_service.update_file(file_id, self.upload_id)
            return file_id
        else:
            result = self.data_service.create_file(parent_kind, parent_id, self.upload_id)
            return result.json()['id']

    @staticmethod
    def send_file_external(data_service, url_json, chunk):
        """
        Send chunk to external store specified in url_json.
        Raises ValueError on upload failure.
        :param data_service: data service to use for sending chunk
        :param url_json: dict contains where/how to upload chunk
        :param chunk: data to be uploaded
        """
        http_verb = url_json['http_verb']
        host = url_json['host']
        url = url_json['url']
        http_headers = url_json['http_headers']
        resp = data_service.send_external(http_verb, host, url, http_headers, chunk)
        if resp.status_code != 200 and resp.status_code != 201:
            raise ValueError("Failed to send file to external store. Error:" + str(resp.status_code))

    @staticmethod
    def hash_chunk(chunk):
        """Creates a hash from the bytes in chunk."""
        hash = HashUtil()
        hash.add_chunk(chunk)
        return hash.hexdigest()


class SerialChunkProcessor(object):
    """
    Uploads chunks one at a time without any additional processes.
    """
    def __init__(self, file_uploader):
        """
        Send chunks in the file specified in file_uploader to the remote data service.
        :param file_uploader: FileUploader contains all data we need to upload chunks of a file.
        """
        self.local_file = file_uploader.local_file
        self.bytes_per_chunk = file_uploader.config.upload_bytes_per_chunk
        self.data_service = file_uploader.data_service
        self.watcher = file_uploader.watcher
        self.upload_id = file_uploader.upload_id
        self.chunk_num = 0

    def run(self):
        """
        Sends contents of a local file to a remote data service.
        """
        processor = self.process_chunk
        if self.local_file.size == 0:
            chunk = ''
            (chunk_hash_alg, chunk_hash_value) = FileUploader.hash_chunk(chunk)
            self.process_chunk(chunk, chunk_hash_alg, chunk_hash_value)
        else:
            with open(self.local_file.path, 'rb') as infile:
                number = 0
                for chunk in SerialChunkProcessor.read_in_chunks(infile, self.bytes_per_chunk):
                    (chunk_hash_alg, chunk_hash_value) = FileUploader.hash_chunk(chunk)
                    self.process_chunk(chunk, chunk_hash_alg, chunk_hash_value)

    def process_chunk(self, chunk, chunk_hash_alg, chunk_hash_value):
        """
        Creates 'upload' url and send bytes to that url.
        Raises ValueError on upload failure.
        :param chunk: bytes part of the file to send
        :param chunk_hash_alg: str the algorithm used to hash chunk
        :param chunk_hash_value: str the hash value of chunk
        """
        self.watcher.transferring_item(self.local_file)
        resp = self.data_service.create_upload_url(self.upload_id, self.chunk_num, len(chunk),
                                                   chunk_hash_value, chunk_hash_alg)
        if resp.status_code == 200:
            FileUploader.send_file_external(self.data_service, resp.json(), chunk)
            self.chunk_num += 1
        else:
            raise ValueError("Failed to retrieve upload url status:" + str(resp.status_code))

    @staticmethod
    def read_in_chunks(infile, blocksize):
        """
        Generator to read chunks lazily.
        :param infile: filehandle file to read from
        :param blocksize: int size of blocks to read
        """
        """"""
        while True:
            data = infile.read(blocksize)
            if not data:
                break
            yield data


class ParallelChunkProcessor(object):
    """
    Uploads grouped chunks in separate processes.
    """
    def __init__(self, file_uploader):
        """
        Send chunks in the file specified in file_uploader to the remote data service using multiple processes.
        :param file_uploader: FileUploader contains all data we need to upload chunks of a file.
        """
        self.config = file_uploader.config
        self.data_service = file_uploader.data_service
        self.upload_id = file_uploader.upload_id
        self.watcher = file_uploader.watcher
        self.local_file = file_uploader.local_file

    def run(self):
        """
        Sends contents of a local file to a remote data service.
        """
        processes = []
        progress_queue = ProgressQueue(Queue())
        num_chunks = ParallelChunkProcessor.determine_num_chunks(self.config.upload_bytes_per_chunk,
                                                                 self.local_file.size)
        work_parcels = ParallelChunkProcessor.make_work_parcels(self.config.upload_workers, num_chunks)
        for (index, num_items) in work_parcels:
            processes.append(self.make_and_start_process(index, num_items, progress_queue))
        wait_for_processes(processes, num_chunks, progress_queue, self.watcher, self.local_file)

    @staticmethod
    def determine_num_chunks(chunk_size, file_size):
        """
        Figure out how many pieces we are sending the file in.
        NOTE: duke-data-service requires an empty chunk to be uploaded for empty files.
        """
{'fraction_non_alphanumeric': 0.0738661176617381, 'fraction_numerical': 0.0022459292532285235, 'mean_word_length': 2.474962063732929, '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': '2727416', 'n_tokens_mistral': 4440, 'n_tokens_neox': 4362, 'n_words': 1320}