# Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import functools
import logging
import math
import os
import random
import socket
import stat
import string
import threading
from collections import defaultdict

from botocore.exceptions import IncompleteReadError, ReadTimeoutError

from s3transfer.compat import SOCKET_ERROR, fallocate, rename_file

MAX_PARTS = 10000
# The maximum file size you can upload via S3 per request.
# See: http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
# and: http://docs.aws.amazon.com/AmazonS3/latest/dev/qfacts.html
MAX_SINGLE_UPLOAD_SIZE = 5 * (1024 ** 3)
MIN_UPLOAD_CHUNKSIZE = 5 * (1024 ** 2)
logger = logging.getLogger(__name__)


S3_RETRYABLE_DOWNLOAD_ERRORS = (
    socket.timeout,
    SOCKET_ERROR,
    ReadTimeoutError,
    IncompleteReadError,
)


def random_file_extension(num_digits=8):
    return ''.join(random.choice(string.hexdigits) for _ in range(num_digits))


def signal_not_transferring(request, operation_name, **kwargs):
    if operation_name in ['PutObject', 'UploadPart'] and hasattr(
        request.body, 'signal_not_transferring'
    ):
        request.body.signal_not_transferring()


def signal_transferring(request, operation_name, **kwargs):
    if operation_name in ['PutObject', 'UploadPart'] and hasattr(
        request.body, 'signal_transferring'
    ):
        request.body.signal_transferring()


def calculate_num_parts(size, part_size):
    return int(math.ceil(size / float(part_size)))


def calculate_range_parameter(
    part_size, part_index, num_parts, total_size=None
):
    """Calculate the range parameter for multipart downloads/copies

    :type part_size: int
    :param part_size: The size of the part

    :type part_index: int
    :param part_index: The index for which this parts starts. This index starts
        at zero

    :type num_parts: int
    :param num_parts: The total number of parts in the transfer

    :returns: The value to use for Range parameter on downloads or
        the CopySourceRange parameter for copies
    """
    # Used to calculate the Range parameter
    start_range = part_index * part_size
    if part_index == num_parts - 1:
        end_range = ''
        if total_size is not None:
            end_range = str(total_size - 1)
    else:
        end_range = start_range + part_size - 1
    range_param = f'bytes={start_range}-{end_range}'
    return range_param


def get_callbacks(transfer_future, callback_type):
    """Retrieves callbacks from a subscriber

    :type transfer_future: s3transfer.futures.TransferFuture
    :param transfer_future: The transfer future the subscriber is associated
        to.

    :type callback_type: str
    :param callback_type: The type of callback to retrieve from the subscriber.
        Valid types include:
            * 'queued'
            * 'progress'
            * 'done'

    :returns: A list of callbacks for the type specified. All callbacks are
        preinjected with the transfer future.
    """
    callbacks = []
    for subscriber in transfer_future.meta.call_args.subscribers:
        callback_name = 'on_' + callback_type
        if hasattr(subscriber, callback_name):
            callbacks.append(
                functools.partial(
                    getattr(subscriber, callback_name), future=transfer_future
                )
            )
    return callbacks


def invoke_progress_callbacks(callbacks, bytes_transferred):
    """Calls all progress callbacks

    :param callbacks: A list of progress callbacks to invoke
    :param bytes_transferred: The number of bytes transferred. This is passed
        to the callbacks. If no bytes were transferred the callbacks will not
        be invoked because no progress was achieved. It is also possible
        to receive a negative amount which comes from retrying a transfer
        request.
    """
    # Only invoke the callbacks if bytes were actually transferred.
    if bytes_transferred:
        for callback in callbacks:
            callback(bytes_transferred=bytes_transferred)


def get_filtered_dict(original_dict, whitelisted_keys):
    """Gets a dictionary filtered by whitelisted keys

    :param original_dict: The original dictionary of arguments to source keys
        and values.
    :param whitelisted_key: A list of keys to include in the filtered
        dictionary.

    :returns: A dictionary containing key/values from the original dictionary
        whose key was included in the whitelist
    """
    filtered_dict = {}
    for key, value in original_dict.items():
        if key in whitelisted_keys:
            filtered_dict[key] = value
    return filtered_dict


class CallArgs:
    def __init__(self, **kwargs):
        """A class that records call arguments

        The call arguments must be passed as keyword arguments. It will set
        each keyword argument as an attribute of the object along with its
        associated value.
        """
        for arg, value in kwargs.items():
            setattr(self, arg, value)


class FunctionContainer:
    """An object that contains a function and any args or kwargs to call it

    When called the provided function will be called with provided args
    and kwargs.
    """

    def __init__(self, func, *args, **kwargs):
        self._func = func
        self._args = args
        self._kwargs = kwargs

    def __repr__(self):
        return 'Function: {} with args {} and kwargs {}'.format(
            self._func, self._args, self._kwargs
        )

    def __call__(self):
        return self._func(*self._args, **self._kwargs)


class CountCallbackInvoker:
    """An abstraction to invoke a callback when a shared count reaches zero

    :param callback: Callback invoke when finalized count reaches zero
    """

    def __init__(self, callback):
        self._lock = threading.Lock()
        self._callback = callback
        self._count = 0