# Copyright Cloudinary
import base64
import copy
import hashlib
import json
import os
import random
import re
import string
import struct
import time
import urllib
import zlib
from collections import OrderedDict
from datetime import datetime, date
from fractions import Fraction
from numbers import Number

import six.moves.urllib.parse
from six import iteritems
from urllib3 import ProxyManager, PoolManager

import cloudinary
from cloudinary import auth_token
from cloudinary.api_client.tcp_keep_alive_manager import TCPKeepAlivePoolManager, TCPKeepAliveProxyManager
from cloudinary.compat import PY3, to_bytes, to_bytearray, to_string, string_types, urlparse

try:  # Python 3.4+
    from pathlib import Path as PathLibPathType
except ImportError:
    PathLibPathType = None

VAR_NAME_RE = r'(\$\([a-zA-Z]\w+\))'

urlencode = six.moves.urllib.parse.urlencode
unquote = six.moves.urllib.parse.unquote

""" @deprecated: use cloudinary.SHARED_CDN """
SHARED_CDN = "res.cloudinary.com"

DEFAULT_RESPONSIVE_WIDTH_TRANSFORMATION = {"width": "auto", "crop": "limit"}

RANGE_VALUE_RE = r'^(?P<value>(\d+\.)?\d+)(?P<modifier>[%pP])?$'
RANGE_RE = r'^(\d+\.)?\d+[%pP]?\.\.(\d+\.)?\d+[%pP]?$'
FLOAT_RE = r'^(\d+)\.(\d+)?$'
REMOTE_URL_RE = r'ftp:|https?:|s3:|gs:|data:([\w-]+\/[\w-]+(\+[\w-]+)?)?(;[\w-]+=[\w-]+)*;base64,([a-zA-Z0-9\/+\n=]+)$'
__LAYER_KEYWORD_PARAMS = [("font_weight", "normal"),
                          ("font_style", "normal"),
                          ("text_decoration", "none"),
                          ("text_align", None),
                          ("stroke", "none")]

# a list of keys used by the cloudinary_url function
__URL_KEYS = [
    'api_secret',
    'auth_token',
    'cdn_subdomain',
    'cloud_name',
    'cname',
    'format',
    'private_cdn',
    'resource_type',
    'secure',
    'secure_cdn_subdomain',
    'secure_distribution',
    'shorten',
    'sign_url',
    'ssl_detected',
    'type',
    'url_suffix',
    'use_root_path',
    'version',
    'long_url_signature',
    'signature_algorithm',
]

__SIMPLE_UPLOAD_PARAMS = [
    "public_id",
    "public_id_prefix",
    "callback",
    "format",
    "type",
    "backup",
    "faces",
    "image_metadata",
    "media_metadata",
    "exif",
    "colors",
    "use_filename",
    "unique_filename",
    "display_name",
    "use_filename_as_display_name",
    "discard_original_filename",
    "filename_override",
    "invalidate",
    "notification_url",
    "eager_notification_url",
    "eager_async",
    "eval",
    "on_success",
    "proxy",
    "folder",
    "asset_folder",
    "use_asset_folder_as_public_id_prefix",
    "unique_display_name",
    "overwrite",
    "moderation",
    "raw_convert",
    "quality_override",
    "quality_analysis",
    "ocr",
    "categorization",
    "detection",
    "similarity_search",
    "visual_search",
    "background_removal",
    "upload_preset",
    "phash",
    "return_delete_token",
    "auto_tagging",
    "async",
    "cinemagraph_analysis",
    "accessibility_analysis",
]

__SERIALIZED_UPLOAD_PARAMS = [
    "timestamp",
    "transformation",
    "headers",
    "eager",
    "tags",
    "allowed_formats",
    "face_coordinates",
    "custom_coordinates",
    "regions",
    "context",
    "auto_tagging",
    "responsive_breakpoints",
    "access_control",
    "metadata",
]

upload_params = __SIMPLE_UPLOAD_PARAMS + __SERIALIZED_UPLOAD_PARAMS

_SIMPLE_TRANSFORMATION_PARAMS = {
        "ac": "audio_codec",
        "af": "audio_frequency",
        "br": "bit_rate",
        "cs": "color_space",
        "d": "default_image",
        "dl": "delay",
        "dn": "density",
        "f": "fetch_format",
        "g": "gravity",
        "p": "prefix",
        "pg": "page",
        "sp": "streaming_profile",
        "vs": "video_sampling",
    }

SHORT_URL_SIGNATURE_LENGTH = 8
LONG_URL_SIGNATURE_LENGTH = 32

SIGNATURE_SHA1 = "sha1"
SIGNATURE_SHA256 = "sha256"

signature_algorithms = {
    SIGNATURE_SHA1: hashlib.sha1,
    SIGNATURE_SHA256: hashlib.sha256,
}


def compute_hex_hash(s, algorithm=SIGNATURE_SHA1):
    """
    Computes string hash using specified algorithm and return HEX string representation of hash.

    :param s:         String to compute hash for
    :param algorithm: The name of algorithm to use for computing hash

    :return: HEX string of computed hash value
    """
    try:
        hash_fn = signature_algorithms[algorithm]
    except KeyError:
        raise ValueError('Unsupported hash algorithm: {}'.format(algorithm))
    return hash_fn(to_bytes(s)).hexdigest()


def build_array(arg):
    if isinstance(arg, (list, tuple)):
        return arg
    elif arg is None:
        return []
    return [arg]


def build_list_of_dicts(val):
    """
    Converts a value that can be presented as a list of dict.

    In case top level item is not a list, it is wrapped with a list

    Valid values examples:
        - Valid dict: {"k": "v", "k2","v2"}
        - List of dict: [{"k": "v"}, {"k2","v2"}]
        - JSON decodable string: '{"k": "v"}', or '[{"k": "v"}]'
        - List of JSON decodable strings: ['{"k": "v"}', '{"k2","v2"}']

    Invalid values examples:
        - ["not", "a", "dict"]
        - [123, None],
        - [["another", "list"]]

    :param val: Input value
    :type val: Union[list, dict, str]

    :return: Converted(or original) list of dict
    :raises: ValueError in case value cannot be converted to a list of dict
    """
    if val is None:
        return []

    if isinstance(val, str):
        # use OrderedDict to preserve order
        val = json.loads(val, object_pairs_hook=OrderedDict)

    if isinstance(val, dict):
        val = [val]

    for index, item in enumerate(val):
        if isinstance(item, str):
            # use OrderedDict to preserve order
            val[index] = json.loads(item, object_pairs_hook=OrderedDict)
        if not isinstance(val[index], dict):
            raise ValueError("Expected a list of dicts")
    return val


def encode_double_array(array):
    array = build_array(array)
    if len(array) > 0 and isinstance(array[0], list):
        return "|".join([",".join([str(i) for i in build_array(inner)]) for inner in array])
    return encode_list([str(i) for i in array])


def encode_dict(arg):
    if isinstance(arg, dict):
        if PY3:
            items = arg.items()
        else:
            items = arg.iteritems()
        return "|".join((k + "=" + v) for k, v in items)
    return arg


def normalize_context_value(value):
    """
    Escape "=" and "|" delimiter characters and json encode lists

    :param value: Value to escape
    :type value: int or str or list or tuple

    :return: The normalized value
    :rtype: str
    """

    if isinstance(value, (list, tuple)):
        value = json_encode(value)

    return str(value).replace("=", "\\=").replace("|", "\\|")


def encode_context(context):
    """
    Encode metadata fields based on incoming value.

    List and tuple values are encoded to json strings.

    :param context: dict of context to be encoded

    :return: a joined string of all keys and values properly escaped and separated by a pipe character
    """
    if not isinstance(context, dict):
        return context

    return "|".join(("{}={}".format(k, normalize_context_value(v))) for k, v in iteritems(context))


def json_encode(value, sort_keys=False):
    """
    Converts value to a json encoded string

    :param value: value to be encoded
    :param sort_keys: whether to sort keys

    :return: JSON encoded string
    """

    if isinstance(value, str) or value is None:
        return value

    return json.dumps(value, default=__json_serializer, separators=(',', ':'), sort_keys=sort_keys)


def encode_date_to_usage_api_format(date_obj):
    """
    Encodes date object to `dd-mm-yyyy` format string

    :param date_obj: datetime.date object to encode

    :return: Encoded date as a string
    """
    return date_obj.strftime('%d-%m-%Y')


def patch_fetch_format(options):
    """
    When upload type is fetch, remove the format options.
    In addition, set the fetch_format options to the format value unless it was already set.
    Mutates the "options" parameter!

    :param options: URL and transformation options
    """
    use_fetch_format = options.pop("use_fetch_format", cloudinary.config().use_fetch_format)

    if options.get("type", "upload") != "fetch" and not use_fetch_format:
        return

    resource_format = options.pop("format", None)
    if "fetch_format" not in options:
        options["fetch_format"] = resource_format


def generate_transformation_string(**options):
    responsive_width = options.pop("responsive_width", cloudinary.config().responsive_width)
    size = options.pop("size", None)
    if size:
        options["width"], options["height"] = size.split("x")
    width = options.get("width")
    height = options.get("height")
    has_layer = ("underlay" in options) or ("overlay" in options)

    crop = options.pop("crop", None)
    angle = ".".join([str(value) for value in build_array(options.pop("angle", None))])
    no_html_sizes = has_layer or angle or crop == "fit" or crop == "limit" or responsive_width

    if width and (str(width).startswith("auto") or str(width) == "ow" or is_fraction(width) or no_html_sizes):
        del options["width"]
    if height and (str(height) == "oh" or is_fraction(height) or no_html_sizes):
        del options["height"]

    background = options.pop("background", None)
    if background:
        background = background.replace("#", "rgb:")
    color = options.pop("color", None)
    if color:
        color = color.replace("#", "rgb:")

    base_transformations = build_array(options.pop("transformation", None))
    if any(isinstance(bs, dict) for bs in base_transformations):
        def recurse(bs):
            if isinstance(bs, dict):
                return generate_transformation_string(**bs)[0]
            return generate_transformation_string(transformation=bs)[0]

        base_transformations = list(map(recurse, base_transformations))
        named_transformation = None
    else:
        named_transformation = ".".join(base_transformations)
        base_transformations = []

    effect = options.pop("effect", None)
    if isinstance(effect, list):
        effect = ":".join([str(x) for x in effect])
    elif isinstance(effect, dict):
        effect = ":".join([str(x) for x in list(effect.items())[0]])

    border = options.pop("border", None)
    if isinstance(border, dict):
        border_color = border.get("color", "black").replace("#", "rgb:")
        border = "%(width)spx_solid_%(color)s" % {"color": border_color,
                                                  "width": str(border.get("width", 2))}

    flags = ".".join(build_array(options.pop("flags", None)))
    dpr = options.pop("dpr", cloudinary.config().dpr)
    duration = norm_range_value(options.pop("duration", None))

    so_raw = options.pop("start_offset", None)
    start_offset = norm_auto_range_value(so_raw)
    if start_offset == None:
        start_offset = so_raw

    eo_raw = options.pop("end_offset", None)
    end_offset = norm_range_value(eo_raw)
    if end_offset == None:
        end_offset = eo_raw

    offset = split_range(options.pop("offset", None))
    if offset:
        start_offset = norm_auto_range_value(offset[0])
        end_offset = norm_range_value(offset[1])

    video_codec = process_video_codec_param(options.pop("video_codec", None))

    aspect_ratio = options.pop("aspect_ratio", None)
    if isinstance(aspect_ratio, Fraction):
        aspect_ratio = str(aspect_ratio.numerator) + ":" + str(aspect_ratio.denominator)

    overlay = process_layer(options.pop("overlay", None), "overlay")
    underlay = process_layer(options.pop("underlay", None), "underlay")
    if_value = process_conditional(options.pop("if", None))
    custom_function = process_custom_function(options.pop("custom_function", None))
    custom_pre_function = process_custom_pre_function(options.pop("custom_pre_function", None))
    fps = process_fps(options.pop("fps", None))

    params = {
        "a": normalize_expression(angle),
        "ar": normalize_expression(aspect_ratio),
        "b": background,
        "bo": border,
        "c": crop,
        "co": color,
        "dpr": normalize_expression(dpr),
        "du": normalize_expression(duration),
        "e": normalize_expression(effect),
        "eo": normalize_expression(end_offset),
        "fl": flags,
        "fn": custom_function or custom_pre_function,
        "fps": fps,
        "h": normalize_expression(height),
        "ki": process_ki(options.pop("keyframe_interval", None)),
        "l": overlay,
        "o": normalize_expression(options.pop('opacity', None)),
        "q": normalize_expression(options.pop('quality', None)),
        "r": process_radius(options.pop('radius', None)),
        "so": normalize_expression(start_offset),
        "t": named_transformation,
        "u": underlay,
        "w": normalize_expression(width),
        "x": normalize_expression(options.pop('x', None)),
        "y": normalize_expression(options.pop('y', None)),
        "vc": video_codec,
        "z": normalize_expression(options.pop('zoom', None))
    }

    for param, option in _SIMPLE_TRANSFORMATION_PARAMS.items():
        params[param] = options.pop(option, None)

    variables = options.pop('variables', {})
    var_params = []
    for key, value in options.items():
        if re.match(r'^\$', key):
            var_params.append(u"{0}_{1}".format(key, normalize_expression(str(value))))

    var_params.sort()

    if variables:
        for var in variables:
            var_params.append(u"{0}_{1}".format(var[0], normalize_expression(str(var[1]))))

    variables = ','.join(var_params)

    sorted_params = sorted([param + "_" + str(value) for param, value in params.items() if (value or value == 0)])
    if variables:
        sorted_params.insert(0, str(variables))

    if if_value is not None:
        sorted_params.insert(0, "if_" + str(if_value))

    if "raw_transformation" in options and (options["raw_transformation"] or options["raw_transformation"] == 0):
        sorted_params.append(options.pop("raw_transformation"))

    transformation = ",".join(sorted_params)

    transformations = base_transformations + [transformation]

    if responsive_width:
        responsive_width_transformation = cloudinary.config().responsive_width_transformation \
                                          or DEFAULT_RESPONSIVE_WIDTH_TRANSFORMATION
        transformations += [generate_transformation_string(**responsive_width_transformation)[0]]
    url = "/".join([trans for trans in transformations if trans])

    if str(width).startswith("auto") or responsive_width:
        options["responsive"] = True
    if dpr == "auto":
        options["hidpi"] = True
    return url, options


def chain_transformations(options, transformations):
    """
    Helper function, allows chaining transformations to the end of transformations list

    The result of this function is an updated options parameter

    :param options:         Original options
    :param transformations: Transformations to chain at the end

    :return: Resulting options
    """

    transformations = copy.deepcopy(transformations)

    transformations = build_array(transformations)
    # preserve url options
    url_options = dict((o, options[o]) for o in __URL_KEYS if o in options)

    transformations.insert(0, options)

    url_options["transformation"] = transformations

    return url_options


def is_fraction(width):
    width = str(width)
    return re.match(FLOAT_RE, width) and float(width) < 1


def split_range(range):
    if (isinstance(range, list) or isinstance(range, tuple)) and len(range) >= 2:
        return [range[0], range[-1]]
    elif isinstance(range, string_types) and re.match(RANGE_RE, range):
        return range.split("..", 1)
    return None


def norm_range_value(value):
    if value is None:
        return None

    match = re.match(RANGE_VALUE_RE, str(value))

    if match is None:
        return None

    modifier = ''
    if match.group('modifier') is not None:
        modifier = 'p'
    return match.group('value') + modifier


def norm_auto_range_value(value):
    if value == "auto":
        return value
    return norm_range_value(value)


def process_video_codec_param(param):
    out_param = param
    if isinstance(out_param, dict):
        out_param = param['codec']
        if 'profile' in param:
            out_param = out_param + ':' + param['profile']
            if 'level' in param:
                out_param = out_param + ':' + param['level']
                if param.get('b_frames') is False:
                    out_param = out_param + ':' + 'bframes_no'

    return out_param


def process_radius(param):
    if param is None:
        return

    if isinstance(param, (list, tuple)):
        if not 1 <= len(param) <= 4:
            raise ValueError("Invalid radius param")
        return ':'.join(normalize_expression(t) for t in param)

    return normalize_expression(str(param))


def process_params(params):
    processed_params = None
    if isinstance(params, dict):
        processed_params = {}
        for key, value in params.items():
            if isinstance(value, list) or isinstance(value, tuple):
                if len(value) == 2 and value[0] == "file":  # keep file parameter as is.
                    processed_params[key] = value
                    continue
                value_list = {"{}[{}]".format(key, i): i_value for i, i_value in enumerate(value)}
                processed_params.update(value_list)
            elif value is not None:
                processed_params[key] = value
    return processed_params


def cleanup_params(params):
    """
    Cleans and normalizes parameters when calculating signature in Upload API.

    :param params:
    :return:
    """
    return dict([(k, __safe_value(v)) for (k, v) in params.items() if v is not None and not v == ""])


def normalize_params(params):
    """
    Normalizes Admin API parameters.

    :param params:
    :return:
    """
    if not params or not isinstance(params, dict):
        return params

    return dict([(k, __bool_string(v)) for (k, v) in params.items() if v is not None and not v == ""])


def sign_request(params, options):
    api_key = options.get("api_key", cloudinary.config().api_key)
    if not api_key:
        raise ValueError("Must supply api_key")
    api_secret = options.get("api_secret", cloudinary.config().api_secret)
    if not api_secret:
        raise ValueError("Must supply api_secret")
    signature_algorithm = options.get("signature_algorithm", cloudinary.config().signature_algorithm)

    params = cleanup_params(params)
    params["signature"] = api_sign_request(params, api_secret, signature_algorithm)
    params["api_key"] = api_key

    return params


def api_sign_request(params_to_sign, api_secret, algorithm=SIGNATURE_SHA1):
    params = [(k + "=" + (",".join(v) if isinstance(v, list) else str(v))) for k, v in params_to_sign.items() if v]
    to_sign = "&".join(sorted(params))
    return compute_hex_hash(to_sign + api_secret, algorithm)


def breakpoint_settings_mapper(breakpoint_settings):
    breakpoint_settings = copy.deepcopy(breakpoint_settings)
    transformation = breakpoint_settings.get("transformation")
    if transformation is not None:
        breakpoint_settings["transformation"], _ = generate_transformation_string(**transformation)
    return breakpoint_settings


def generate_responsive_breakpoints_string(breakpoints):
    if breakpoints is None:
        return None
    breakpoints = build_array(breakpoints)
    return json.dumps(list(map(breakpoint_settings_mapper, breakpoints)))


def finalize_source(source, format, url_suffix):
    source = re.sub(r'([^:])/+', r'\1/', source)
    if re.match(r'^https?:/', source):
        source = smart_escape(source)
        source_to_sign = source
    else:
        source = unquote(source)
        if not PY3:
            source = source.encode('utf8')
        source = smart_escape(source)
        source_to_sign = source
        if url_suffix is not None:
            if re.search(r'[\./]', url_suffix):
                raise ValueError("url_suffix should not include . or /")
            source = source + "/" + url_suffix
        if format is not None:
            source = source + "." + format
            source_to_sign = source_to_sign + "." + format

    return source, source_to_sign


def finalize_resource_type(resource_type, type, url_suffix, use_root_path, shorten):
    upload_type = type or "upload"
    if url_suffix is not None:
        if resource_type == "image" and upload_type == "upload":
            resource_type = "images"
            upload_type = None
        elif resource_type == "raw" and upload_type == "upload":
            resource_type = "files"
            upload_type = None
        else:
            raise ValueError("URL Suffix only supported for image/upload and raw/upload")

    if use_root_path:
        if (resource_type == "image" and upload_type == "upload") or (
                resource_type == "images" and upload_type is None):
            resource_type = None
            upload_type = None
        else:
            raise ValueError("Root path only supported for image/upload")

    if shorten and resource_type == "image" and upload_type == "upload":
        resource_type = "iu"
        upload_type = None

    return resource_type, upload_type


def unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain,
                                 secure_cdn_subdomain, cname, secure, secure_distribution):
    """cdn_subdomain and secure_cdn_subdomain
    1) Customers in shared distribution (e.g. res.cloudinary.com)
      if cdn_domain is true uses res-[1-5].cloudinary.com for both http and https.
      Setting secure_cdn_subdomain to false disables this for https.
    2) Customers with private cdn
      if cdn_domain is true uses cloudname-res-[1-5].cloudinary.com for http
      if secure_cdn_domain is true uses cloudname-res-[1-5].cloudinary.com for https
      (please contact support if you require this)
    3) Customers with cname
      if cdn_domain is true uses a[1-5].cname for http. For https, uses the same naming scheme
      as 1 for shared distribution and as 2 for private distribution."""
    shared_domain = not private_cdn
    shard = __crc(source)
    if secure:
        if secure_distribution is None or secure_distribution == cloudinary.OLD_AKAMAI_SHARED_CDN:
            secure_distribution = cloud_name + "-res.cloudinary.com" \
                if private_cdn else cloudinary.SHARED_CDN

        shared_domain = shared_domain or secure_distribution == cloudinary.SHARED_CDN
        if secure_cdn_subdomain is None and shared_domain:
            secure_cdn_subdomain = cdn_subdomain

        if secure_cdn_subdomain:
            secure_distribution = re.sub('res.cloudinary.com', "res-" + shard + ".cloudinary.com",
                                         secure_distribution)

        prefix = "https://" + secure_distribution
    elif cname:
        subdomain = "a" + shard + "." if cdn_subdomain else ""
        prefix = "http://" + subdomain + cname
    else:
        subdomain = cloud_name + "-res" if private_cdn else "res"
        if cdn_subdomain:
            subdomain = subdomain + "-" + shard
        prefix = "http://" + subdomain + ".cloudinary.com"

    if shared_domain:
        prefix += "/" + cloud_name

    return prefix


def build_distribution_domain(options):
    source = options.pop('source', '')
    cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name or None)
    if cloud_name is None:
        raise ValueError("Must supply cloud_name in tag or in configuration")
    secure = options.pop("secure", cloudinary.config().secure)
    private_cdn = options.pop("private_cdn", cloudinary.config().private_cdn)
    cname = options.pop("cname", cloudinary.config().cname)
    secure_distribution = options.pop("secure_distribution",
                                      cloudinary.config().secure_distribution)
    cdn_subdomain = options.pop("cdn_subdomain", cloudinary.config().cdn_subdomain)
    secure_cdn_subdomain = options.pop("secure_cdn_subdomain",
                                       cloudinary.config().secure_cdn_subdomain)

    return unsigned_download_url_prefix(
        source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain,
        cname, secure, secure_distribution)


def merge(*dict_args):
    result = None
    for dictionary in dict_args:
        if dictionary is not None:
            if result is None:
                result = dictionary.copy()
            else:
                result.update(dictionary)
    return result


def cloudinary_url(source, **options):
    original_source = source

    patch_fetch_format(options)
    type = options.pop("type", "upload")

    transformation, options = generate_transformation_string(**options)

    resource_type = options.pop("resource_type", "image")

    force_version = options.pop("force_version", cloudinary.config().force_version)
    if force_version is None:
        force_version = True

    version = options.pop("version", None)

    format = options.pop("format", None)
    shorten = options.pop("shorten", cloudinary.config().shorten)

    sign_url = options.pop("sign_url", cloudinary.config().sign_url)
    api_secret = options.pop("api_secret", cloudinary.config().api_secret)
    url_suffix = options.pop("url_suffix", None)
    use_root_path = options.pop("use_root_path", cloudinary.config().use_root_path)
    auth_token = options.pop("auth_token", None)
    long_url_signature = options.pop("long_url_signature", cloudinary.config().long_url_signature)
    signature_algorithm = options.pop("signature_algorithm", cloudinary.config().signature_algorithm)
    if auth_token is not False:
        auth_token = merge(cloudinary.config().auth_token, auth_token)

    if (not source) or type == "upload" and re.match(r'^https?:', source):
        return original_source, options

    resource_type, type = finalize_resource_type(
        resource_type, type, url_suffix, use_root_path, shorten)
    source, source_to_sign = finalize_source(source, format, url_suffix)

    if not version and force_version \
            and source_to_sign.find("/") >= 0 \
            and not re.match(r'^https?:/', source_to_sign) \
            and not re.match(r'^v[0-9]+', source_to_sign):
        version = "1"
    if version:
        version = "v" + str(version)
    else:
        version = None

    transformation = re.sub(r'([^:])/+', r'\1/', transformation)

    signature = None
    if sign_url and not auth_token:
        to_sign = "/".join(__compact([transformation, source_to_sign]))
        if long_url_signature:
            # Long signature forces SHA256
            signature_algorithm = SIGNATURE_SHA256
            chars_length = LONG_URL_SIGNATURE_LENGTH
        else:
            chars_length = SHORT_URL_SIGNATURE_LENGTH
        if signature_algorithm not in signature_algorithms:
            raise ValueError("Unsupported signature algorithm '{}'".format(signature_algorithm))
        hash_fn = signature_algorithms[signature_algorithm]
        signature = "s--" + to_string(
            base64.urlsafe_b64encode(
                hash_fn(to_bytes(to_sign + api_secret)).digest())[0:chars_length]) + "--"

    options["source"] = source
    prefix = build_distribution_domain(options)

    source = "/".join(__compact(
        [prefix, resource_type, type, signature, transformation, version, source]))
    if sign_url and auth_token:
        path = urlparse(source).path
        token = cloudinary.auth_token.generate(**merge(auth_token, {"url": path}))
        source = "%s?%s" % (source, token)
    return source, options


def base_api_url(path, **options):
    cloudinary_prefix = options.get("upload_prefix", cloudinary.config().upload_prefix) \
                        or "https://api.cloudinary.com"
    cloud_name = options.get("cloud_name", cloudinary.config().cloud_name)

    if not cloud_name:
        raise ValueError("Must supply cloud_name")

    path = build_array(path)

    return encode_unicode_url("/".join([cloudinary_prefix, cloudinary.API_VERSION, cloud_name] + path))


def cloudinary_api_url(action='upload', **options):
    resource_type = options.get("resource_type", "image")

    return base_api_url([resource_type, action], **options)


def cloudinary_api_download_url(action, params, **options):
    params = params.copy()
    params["mode"] = "download"
    cloudinary_params = sign_request(params, options)
    return cloudinary_api_url(action, **options) + "?" + urlencode(bracketize_seq(cloudinary_params), True)


def cloudinary_scaled_url(source, width, transformation, options):
    """
    Generates a cloudinary url scaled to specified width.

    :param source:          The resource
    :param width:           Width in pixels of the srcset item
    :param transformation:  Custom transformation that overrides transformations provided in options
    :param options:         A dict with additional options

    :return: Resulting URL of the item
    """

    # preserve options from being destructed
    options = copy.deepcopy(options)

    if transformation:
        if isinstance(transformation, string_types):
            transformation = {"raw_transformation": transformation}

        # Remove all transformation related options
        options = dict((o, options[o]) for o in __URL_KEYS if o in options)
        options.update(transformation)

    scale_transformation = {"crop": "scale", "width": width}

    url_options = options
    patch_fetch_format(url_options)
    url_options = chain_transformations(url_options, scale_transformation)

    return cloudinary_url(source, **url_options)[0]


def smart_escape(source, unsafe=r"([^a-zA-Z0-9_.\-\/:]+)"):
    """
    Based on ruby's CGI::unescape. In addition does not escape / :

    :param source: Source string to escape
    :param unsafe: Unsafe characters

    :return: Escaped string
    """
    def pack(m):
        return to_bytes('%' + "%".join(
            ["%02X" % x for x in struct.unpack('B' * len(m.group(1)), m.group(1))]
        ).upper())

    return to_string(re.sub(to_bytes(unsafe), pack, to_bytes(source)))


def random_public_id():
    return ''.join(random.SystemRandom().choice(string.ascii_lowercase + string.digits)
                   for _ in range(16))


def signed_preloaded_image(result):
    filename = ".".join([x for x in [result["public_id"], result["format"]] if x])
    path = "/".join([result["resource_type"], "upload", "v" + str(result["version"]), filename])
    return path + "#" + result["signature"]


def now():
    return str(int(time.time()))


def private_download_url(public_id, format, **options):
    cloudinary_params = sign_request({
        "timestamp": now(),
        "public_id": public_id,
        "format": format,
        "type": options.get("type"),
        "attachment": options.get("attachment"),
        "expires_at": options.get("expires_at")
    }, options)

    return cloudinary_api_url("download", **options) + "?" + urlencode(cloudinary_params)


def zip_download_url(tag, **options):
    cloudinary_params = sign_request({
        "timestamp": now(),
        "tag": tag,
        "transformation": generate_transformation_string(**options)[0]
    }, options)

    return cloudinary_api_url("download_tag.zip", **options) + "?" + urlencode(cloudinary_params)


def bracketize_seq(params):
    url_params = dict()
    for param_name in params:
        param_value = params[param_name]
        if isinstance(param_value, list):
            param_name += "[]"
        url_params[param_name] = param_value
    return url_params


def download_archive_url(**options):
    return cloudinary_api_download_url(action="generate_archive", params=archive_params(**options), **options)


def download_zip_url(**options):
    new_options = options.copy()
    new_options.update(target_format="zip")
    return download_archive_url(**new_options)


def download_folder(folder_path, **options):
    """
    Creates and returns a URL that when invoked creates an archive of a folder.
    :param folder_path: The full path from the root that is used to generate download url.
    :type folder_path:  str
    :param options:     Additional options.
    :type options:      dict, optional
    :return:            Signed URL to download the folder.
    :rtype:             str
    """
    options["prefixes"] = folder_path
    options.setdefault("resource_type", "all")

    return download_archive_url(**options)


def download_backedup_asset(asset_id, version_id, **options):
    """
    The returned url allows downloading the backedup asset based on the the asset ID and the version ID.

    Parameters asset_id and version_id are returned with api.resource(<PUBLIC_ID1>, versions=True) API call.

    :param  asset_id:   The asset ID of the asset.
    :type   asset_id:   str
    :param  version_id: The version ID of the asset.
    :type   version_id: str
    :param  options:    Additional options.
    :type   options:    dict, optional
    :return:The signed URL for downloading backup version of the asset.
    :rtype: str
    """
    params = {
        "timestamp": options.get("timestamp", now()),
        "asset_id": asset_id,
        "version_id": version_id
    }
    cloudinary_params = sign_request(params, options)

    return base_api_url("download_backup", **options) + "?" + urlencode(bracketize_seq(cloudinary_params), True)


def generate_auth_token(**options):
    token_options = merge(cloudinary.config().auth_token, options)
    return auth_token.generate(**token_options)


def archive_params(**options):
    if options.get("timestamp") is None:
        timestamp = now()
    else:
        timestamp = options.get("timestamp")
    params = {
        "allow_missing": options.get("allow_missing"),
        "async": options.get("async"),
        "expires_at": options.get("expires_at"),
        "flatten_folders": options.get("flatten_folders"),
        "flatten_transformations": options.get("flatten_transformations"),
        "keep_derived": options.get("keep_derived"),
        "mode": options.get("mode"),
        "notification_url": options.get("notification_url"),
        "phash": options.get("phash"),
        "prefixes": options.get("prefixes") and build_array(options.get("prefixes")),
        "public_ids": options.get("public_ids") and build_array(options.get("public_ids")),
        "fully_qualified_public_ids": options.get("fully_qualified_public_ids") and build_array(
            options.get("fully_qualified_public_ids")),
        "skip_transformation_name": options.get("skip_transformation_name"),
        "tags": options.get("tags") and build_array(options.get("tags")),
        "target_format": options.get("target_format"),
        "target_asset_folder": options.get("target_asset_folder"),
        "target_public_id": options.get("target_public_id"),
        "target_tags": options.get("target_tags") and build_array(options.get("target_tags")),
        "timestamp": timestamp,
        "transformations": build_eager(options.get("transformations")),
        "type": options.get("type"),
        "use_original_filename": options.get("use_original_filename"),
    }
    return params


def build_eager(transformations):
    if transformations is None:
        return None

    return "|".join([build_single_eager(et) for et in build_array(transformations)])


def build_single_eager(options):
    """
    Builds a single eager transformation which consists of transformation and (optionally) format joined by "/"

    :param options: Options containing transformation parameters and (optionally) a "format" key
        format can be a string value (jpg, gif, etc) or can be set to "" (empty string).
        The latter leads to transformation ending with "/", which means "No extension, use original format"
        If format is not provided or set to None, only transformation is used (without the trailing "/")

    :return: Resulting eager transformation string
    """
    if isinstance(options, string_types):
        return options

    trans_str = generate_transformation_string(**options)[0]

    if not trans_str:
        return ""

    file_format = options.get("format")

    return trans_str + ("/" + file_format if file_format is not None else "")


def build_custom_headers(headers):
    if headers is None:
        return None
    elif isinstance(headers, list):
        pass
    elif isinstance(headers, dict):
        headers = [k + ": " + v for k, v in headers.items()]
    else:
        return headers
    return "\n".join(headers)


def build_upload_params(**options):
    params = {param_name: options.get(param_name) for param_name in __SIMPLE_UPLOAD_PARAMS if param_name in options}
    params["upload_preset"] = params.pop("upload_preset", cloudinary.config().upload_preset)

    serialized_params = {
        "timestamp": now(),
        "metadata": encode_context(options.get("metadata")),
        "transformation": generate_transformation_string(**options)[0],
        "headers": build_custom_headers(options.get("headers")),
        "eager": build_eager(options.get("eager")),
        "tags": options.get("tags") and encode_list(build_array(options["tags"])),
        "allowed_formats": options.get("allowed_formats") and encode_list(build_array(options["allowed_formats"])),
        "face_coordinates": encode_double_array(options.get("face_coordinates")),
        "custom_coordinates": encode_double_array(options.get("custom_coordinates")),
        "regions": json_encode(options.get("regions")),
        "context": encode_context(options.get("context")),
        "auto_tagging": options.get("auto_tagging") and str(options.get("auto_tagging")),
        "responsive_breakpoints": generate_responsive_breakpoints_string(options.get("responsive_breakpoints")),
        "access_control": options.get("access_control") and json_encode(
            build_list_of_dicts(options.get("access_control")))
    }

    # make sure that we are in-sync with __SERIALIZED_UPLOAD_PARAMS which are in use by other methods
    serialized_params = {param_name: serialized_params[param_name] for param_name in __SERIALIZED_UPLOAD_PARAMS}

    params.update(serialized_params)

    return params


def handle_file_parameter(file, filename):
    if not file:
        return None

    if PathLibPathType and isinstance(file, PathLibPathType):
        name = filename or file.name
        data = file.read_bytes()
    elif isinstance(file, string_types):
        if is_remote_url(file):
            # URL
            name = None
            data = file
        else:
            # file path
            name = filename or file
            with open(file, "rb") as opened:
                data = opened.read()
    elif hasattr(file, 'read') and callable(file.read):
        # stream
        data = file.read()
        name = filename or (file.name if hasattr(file, 'name') and isinstance(file.name, str) else "stream")
    elif isinstance(file, tuple):
        name, data = file
    else:
        # Not a string, not a stream
        name = filename or "file"
        data = file

    return (name, data) if name else data


def build_multi_and_sprite_params(**options):
    """
    Build params for multi, download_multi, generate_sprite, and download_generated_sprite methods
    """
    tag = options.get("tag")
    urls = options.get("urls")
    if bool(tag) == bool(urls):
        raise ValueError("Either 'tag' or 'urls' parameter has to be set but not both")
    params = {
        "mode": options.get("mode"),
        "timestamp": now(),
        "async": options.get("async"),
        "notification_url": options.get("notification_url"),
        "tag": tag,
        "urls": urls,
        "transformation": generate_transformation_string(fetch_format=options.get("format"), **options)[0]
    }
    return params


def __process_text_options(layer, layer_parameter):
    text_style = str(layer.get("text_style", ""))
    if text_style and not text_style.isspace():
        return text_style

    font_family = layer.get("font_family")
    font_size = layer.get("font_size")
    keywords = []
    for attr, default_value in __LAYER_KEYWORD_PARAMS:
        attr_value = layer.get(attr)
        if attr_value != default_value and attr_value is not None:
            keywords.append(attr_value)

    letter_spacing = layer.get("letter_spacing")
    if letter_spacing is not None:
        keywords.append("letter_spacing_" + str(letter_spacing))

    line_spacing = layer.get("line_spacing")
    if line_spacing is not None:
        keywords.append("line_spacing_" + str(line_spacing))

    font_antialiasing = layer.get("font_antialiasing")
    if font_antialiasing is not None:
        keywords.append("antialias_" + str(font_antialiasing))

    font_hinting = layer.get("font_hinting")
    if font_hinting is not None:
        keywords.append("hinting_" + str(font_hinting))

    if font_size is None and font_family is None and len(keywords) == 0:
        return None

    if font_family is None:
        raise ValueError("Must supply font_family for text in " + layer_parameter)

    if font_size is None:
        raise ValueError("Must supply font_size for text in " + layer_parameter)

    keywords.insert(0, font_size)
    keywords.insert(0, font_family)

    return '_'.join([str(k) for k in keywords])


def process_layer(layer, layer_parameter):
    if isinstance(layer, string_types):
        resource_type = None
        if layer.startswith("fetch:"):
            url = layer[len('fetch:'):]
        elif layer.find(":fetch:", 0, 12) != -1:
            resource_type, _, url = layer.split(":", 2)
        else:
            # nothing to process, a raw string, keep as is.
            return layer

        # handle remote fetch URL
        layer = {"url": url, "type": "fetch"}
        if resource_type:
            layer["resource_type"] = resource_type

    if not isinstance(layer, dict):
        return layer

    resource_type = layer.get("resource_type")
    text = layer.get("text")
    type = layer.get("type")
    public_id = layer.get("public_id")
    format = layer.get("format")
    fetch_url = layer.get("url")
    components = list()

    if text is not None and resource_type is None:
        resource_type = "text"

    if fetch_url and type is None:
        type = "fetch"

    if public_id is not None and format is not None:
        public_id = public_id + "." + format

    if public_id is None and resource_type != "text" and type != "fetch":
        raise ValueError("Must supply public_id for for non-text " + layer_parameter)

    if resource_type is not None and resource_type != "image":
        components.append(resource_type)

    if type is not None and type != "upload":
        components.append(type)

    if resource_type == "text" or resource_type == "subtitles":
        if public_id is None and text is None:
            raise ValueError("Must supply either text or public_id in " + layer_parameter)

        text_options = __process_text_options(layer, layer_parameter)

        if text_options is not None:
            components.append(text_options)

        if public_id is not None:
            public_id = public_id.replace("/", ':')
            components.append(public_id)

        if text is not None:
            var_pattern = VAR_NAME_RE
            parts = filter(lambda p: p is not None, re.split(var_pattern, text))
            encoded_text = []
            for part in parts:
                if re.match(var_pattern, part):
                    encoded_text.append(part)
                else:
                    encoded_text.append(smart_escape(smart_escape(part, r"([,/])")))

            text = ''.join(encoded_text)
            components.append(text)
    elif type == "fetch":
        b64 = base64url_encode(fetch_url)
        components.append(b64)
    else:
        public_id = public_id.replace("/", ':')
        components.append(public_id)

    return ':'.join(components)


IF_OPERATORS = {
    "=": 'eq',
    "!=": 'ne',
    "<": 'lt',
    ">": 'gt',
    "<=": 'lte',
    ">=": 'gte',
    "&&": 'and',
    "||": 'or',
    "*": 'mul',
    "/": 'div',
    "+": 'add',
    "-": 'sub',
    "^": 'pow'
}

PREDEFINED_VARS = {
    "aspect_ratio": "ar",
    "aspectRatio": "ar",
    "current_page": "cp",
    "currentPage": "cp",
    "face_count": "fc",
    "faceCount": "fc",
    "height": "h",
    "initial_aspect_ratio": "iar",
    "initialAspectRatio": "iar",
    "trimmed_aspect_ratio": "tar",
    "trimmedAspectRatio": "tar",
    "initial_height": "ih",
    "initialHeight": "ih",
    "initial_width": "iw",
    "initialWidth": "iw",
    "page_count": "pc",
    "pageCount": "pc",
    "page_x": "px",
    "pageX": "px",
    "page_y": "py",
    "pageY": "py",
    "tags": "tags",
    "width": "w",
    "duration": "du",
    "initial_duration": "idu",
    "initialDuration": "idu",
    "illustration_score": "ils",
    "illustrationScore": "ils",
    "context": "ctx"
}

replaceRE = "((\\|\\||>=|<=|&&|!=|>|=|<|/|-|\\+|\\*|\\^)(?=[ _])|(\\$_*[^_ ]+)|(?<![\\$:])(" + \
            '|'.join(PREDEFINED_VARS.keys()) + "))"


def translate_if(match):
    name = match.group(0)
    return IF_OPERATORS.get(name,
                            PREDEFINED_VARS.get(name,
                                                name))


def process_custom_function(custom_function):
    if not isinstance(custom_function, dict):
        return custom_function

    function_type = custom_function.get("function_type")
    source = custom_function.get("source")
    if function_type == "remote":
        source = base64url_encode(source)

    return ":".join([function_type, source])


def process_custom_pre_function(custom_function):
    value = process_custom_function(custom_function)
    return "pre:{0}".format(value) if value else None


def process_fps(fps):
    """
    Serializes fps transformation parameter

    :param fps: A single number, a list of mixed type, a string, including open-ended and closed range values
                Examples: '24-29.97', 24, 24.973, '-24', [24, 29.97]

    :return: string
    """
    if not isinstance(fps, (list, tuple)):
        return fps

    return "-".join(normalize_expression(f) for f in fps)


def process_ki(ki):
    """
    Serializes keyframe_interval parameter
    :param ki: Keyframe interval. Should be either a string or a positive real number.
    :return: string
    """
    if ki is None:
        return None
    if isinstance(ki, string_types):
        return ki
    if not isinstance(ki, Number):
        raise ValueError("Keyframe interval should be a number or a string")
    if ki <= 0:
        raise ValueError("Keyframe interval should be greater than zero")
    return str(float(ki))


def process_conditional(conditional):
    if conditional is None:
        return conditional
    result = normalize_expression(conditional)
    return result


def normalize_expression(expression):
    if re.match(r'^!.+!$', str(expression)):  # quoted string
        return expression
    elif expression:
        result = str(expression)
        result = re.sub(replaceRE, translate_if, result)
        result = re.sub('[ _]+', '_', result)
        return result
    return expression


def __join_pair(key, value):
    if value is None or value == "":
        return None
    elif value is True:
        return key
    return u"{0}=\"{1}\"".format(key, value)


def html_attrs(attrs, only=None):
    return ' '.join(sorted([__join_pair(key, value) for key, value in attrs.items() if only is None or key in only]))


def __safe_value(v):
    if isinstance(v, bool):
        return "1" if v else "0"
    return v


def __bool_string(v):
    if isinstance(v, bool):
        return "true" if v else "false"

    return v

def __crc(source):
    return str((zlib.crc32(to_bytearray(source)) & 0xffffffff) % 5 + 1)


def __compact(array):
    return filter(lambda x: x, array)


def base64_encode_url(url):
    """
    Returns the Base64-decoded version of url.
    The method tries to unquote the url because quoting it

    :param str url:
        the url to encode. the value is URIdecoded and then
        re-encoded before converting to base64 representation

    """

    try:
        url = unquote(url)
    except Exception:
        pass
    url = smart_escape(url)
    b64 = base64.b64encode(url.encode('utf-8'))
    return b64.decode('ascii')


def base64url_encode(data):
    """
    Url safe version of urlsafe_b64encode with stripped `=` sign at the end.

    :param data: input data

    :return: Base64 URL safe encoded string
    """
    return to_string(base64.urlsafe_b64encode(to_bytes(data)))


def encode_unicode_url(url_str):
    """
    Quote and encode possible unicode url string (applicable for python2)

    :param url_str: Url string to encode

    :return: Encoded string
    """
    if six.PY2:
        url_str = urllib.quote(url_str.encode('utf-8'), ":/?#[]@!$&'()*+,;=")

    return url_str


def __json_serializer(obj):
    """JSON serializer for objects not serializable by default json code"""
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    raise TypeError("Object of type %s is not JSON serializable" % type(obj))


def is_remote_url(file):
    """Basic URL scheme check to define if it's remote URL"""
    return isinstance(file, string_types) and re.match(REMOTE_URL_RE, file)


def file_io_size(file_io):
    """
    Helper function for getting file-like object size(suitable for both files and streams)

    :param file_io: io.IOBase

    :return: size
    """
    initial_position = file_io.tell()
    file_io.seek(0, os.SEEK_END)
    size = file_io.tell()
    file_io.seek(initial_position, os.SEEK_SET)

    return size


def check_property_enabled(f):
    """
    Used as a class method decorator to check whether class is enabled(self.enabled is True)

    :param f: function to call

    :return: None if not enabled, otherwise calls function f
    """
    def wrapper(*args, **kwargs):
        if not args[0].enabled:
            return None
        return f(*args, **kwargs)

    return wrapper


def verify_api_response_signature(public_id, version, signature, algorithm=None):
    """
    Verifies the authenticity of an API response signature

    :param public_id: The public id of the asset as returned in the API response
    :param version:   The version of the asset as returned in the API response
    :param signature: Actual signature. Can be retrieved from the X-Cld-Signature header
    :param algorithm: Name of hashing algorithm to use for calculation of HMACs.
                      By default uses `cloudinary.config().signature_algorithm`

    :return: Boolean result of the validation
    """
    if not cloudinary.config().api_secret:
        raise Exception('Api secret key is empty')

    parameters_to_sign = {'public_id': public_id,
                          'version': version}

    return signature == api_sign_request(
        parameters_to_sign,
        cloudinary.config().api_secret,
        algorithm or cloudinary.config().signature_algorithm
    )


def verify_notification_signature(body, timestamp, signature, valid_for=7200, algorithm=None):
    """
    Verifies the authenticity of a notification signature

    :param body: Json of the request's body
    :param timestamp: Unix timestamp. Can be retrieved from the X-Cld-Timestamp header
    :param signature: Actual signature. Can be retrieved from the X-Cld-Signature header
    :param valid_for: The desired time in seconds for considering the request valid
    :param algorithm: Name of hashing algorithm to use for calculation of HMACs.
                      By default uses `cloudinary.config().signature_algorithm`

    :return: Boolean result of the validation
    """
    if not cloudinary.config().api_secret:
        raise Exception('Api secret key is empty')

    if timestamp < time.time() - valid_for:
        return False

    if not isinstance(body, str):
        raise ValueError('Body should be type of string')

    return signature == compute_hex_hash(
        '{}{}{}'.format(body, timestamp, cloudinary.config().api_secret),
        algorithm or cloudinary.config().signature_algorithm)


def get_http_connector(conf, options):
    """
    Used to create http connector, depends on api_proxy and disable_tcp_keep_alive configuration parameters.

    :param conf: configuration object
    :param options: additional options

    :return: ProxyManager if api_proxy is set, otherwise PoolManager object
    """
    if conf.api_proxy:
        if conf.disable_tcp_keep_alive:
            return ProxyManager(conf.api_proxy, **options)

        return TCPKeepAliveProxyManager(conf.api_proxy, **options)

    if conf.disable_tcp_keep_alive:
        return PoolManager(**options)

    return TCPKeepAlivePoolManager(**options)


def encode_list(obj):
    if isinstance(obj, list):
        return ",".join(obj)
    return obj


def safe_cast(val, casting_fn, default=None):
    """
    Attempts to cast a value to another using a given casting function
    Will return a default value if casting fails (configurable, defaults to None)

    :param val: The value to cast
    :param casting_fn: The casting function that will receive the value to cast
    :param default: The return value if casting fails

    :return: Result of casting the value or the value of the default parameter
    """
    try:
        return casting_fn(val)
    except (ValueError, TypeError):
        return default


def __id(x):
    """
    Identity function. Returns the passed in values.
    """
    return x


def unique(collection, key=None):
    """
    Removes duplicates from collection using key function

    :param collection: The collection to remove duplicates from
    :param key: The function to generate key from each element. If not passed, identity function is used
    """
    if key is None:
        key = __id

    to_return = OrderedDict()

    for element in collection:
        to_return[key(element)] = element

    return list(to_return.values())


def fq_public_id(public_id, resource_type="image", type="upload"):
    """
    Returns the fully qualified public id of form resource_type/type/public_id.

    :param public_id: The public ID of the asset.
    :type public_id: str
    :param resource_type: The type of the asset. Defaults to "image".
    :type resource_type: str
    :param type: The upload type. Defaults to "upload".
    :type type: str

    :return:
    """
    return "{resource_type}/{type}/{public_id}".format(resource_type=resource_type, type=type, public_id=public_id)