Skip to content
Branch: master
Find file Copy path
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
1232 lines (955 sloc) 40.4 KB
# 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
import six.moves.urllib.parse
from numbers import Number
from six import iteritems
import cloudinary
from cloudinary import auth_token
from cloudinary.compat import PY3, to_bytes, to_bytearray, to_string, string_types, urlparse
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 """
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-]+)*;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 = [
def build_array(arg):
if isinstance(arg, list):
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 ",".join([str(i) for i in array])
def encode_dict(arg):
if isinstance(arg, dict):
if PY3:
items = arg.items()
items = arg.iteritems()
return "|".join((k + "=" + v) for k, v in items)
return arg
def encode_context(context):
: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, v.replace("=", "\\=").replace("|", "\\|"))) for k, v in iteritems(context))
def json_encode(value):
Converts value to a json encoded string
:param value: value to be encoded
:return: JSON encoded string
return json.dumps(value, default=__json_serializer, separators=(',', ':'))
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
if options.get("type", "upload") != "fetch":
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
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))
start_offset = norm_auto_range_value(options.pop("start_offset", None))
end_offset = norm_range_value(options.pop("end_offset", None))
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))
simple_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",
for param, option in simple_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))))
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):
transformation = ",".join(sorted_params)
transformations = base_transformations + [transformation]
if responsive_width:
responsive_width_transformation = cloudinary.config().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'modifier') is not None:
modifier = 'p'
return'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']
return out_param
def process_radius(param):
if param is None:
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 str(param)
def cleanup_params(params):
return dict([(k, __safe_value(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")
params = cleanup_params(params)
params["signature"] = api_sign_request(params, api_secret)
params["api_key"] = api_key
return params
def api_sign_request(params_to_sign, api_secret):
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 hashlib.sha1(to_bytes(to_sign + api_secret)).hexdigest()
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
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'[\./]', 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
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
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.
if cdn_domain is true uses res-[1-5] 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] for http
if secure_cdn_domain is true uses cloudname-res-[1-5] 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 + "" \
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-" + shard + "",
prefix = "https://" + secure_distribution
elif cname:
subdomain = "a" + shard + "." if cdn_subdomain else ""
prefix = "http://" + subdomain + cname
subdomain = cloud_name + "-res" if private_cdn else "res"
if cdn_subdomain:
subdomain = subdomain + "-" + shard
prefix = "http://" + subdomain + ""
if shared_domain:
prefix += "/" + cloud_name
return prefix
def merge(*dict_args):
result = None
for dictionary in dict_args:
if dictionary is not None:
if result is None:
result = dictionary.copy()
return result
def cloudinary_url(source, **options):
original_source = source
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)
cdn_subdomain = options.pop("cdn_subdomain", cloudinary.config().cdn_subdomain)
secure_cdn_subdomain = options.pop("secure_cdn_subdomain",
cname = options.pop("cname", cloudinary.config().cname)
shorten = options.pop("shorten", cloudinary.config().shorten)
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)
secure_distribution = options.pop("secure_distribution",
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)
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)
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]))
signature = "s--" + to_string(
hashlib.sha1(to_bytes(to_sign + api_secret)).digest())[0:8]) + "--"
prefix = unsigned_download_url_prefix(
source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain,
cname, secure, secure_distribution)
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 cloudinary_api_url(action='upload', **options):
cloudinary_prefix = options.get("upload_prefix", cloudinary.config().upload_prefix)\
or ""
cloud_name = options.get("cloud_name", cloudinary.config().cloud_name)
if not cloud_name:
raise ValueError("Must supply cloud_name")
resource_type = options.get("resource_type", "image")
return encode_unicode_url("/".join([cloudinary_prefix, "v1_1", cloud_name, resource_type, action]))
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)
scale_transformation = {"crop": "scale", "width": width}
url_options = 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(,]
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("", **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):
params = options.copy()
cloudinary_params = sign_request(archive_params(**params), options)
return cloudinary_api_url("generate_archive", **options) + "?" + \
urlencode(bracketize_seq(cloudinary_params), True)
def download_zip_url(**options):
new_options = options.copy()
return download_archive_url(**new_options)
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()
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(
"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_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):
elif isinstance(headers, dict):
headers = [k + ": " + v for k, v in headers.items()]
return headers
return "\n".join(headers)
def build_upload_params(**options):
params = {
"timestamp": now(),
"transformation": generate_transformation_string(**options)[0],
"public_id": options.get("public_id"),
"callback": options.get("callback"),
"format": options.get("format"),
"type": options.get("type"),
"backup": options.get("backup"),
"faces": options.get("faces"),
"image_metadata": options.get("image_metadata"),
"exif": options.get("exif"),
"colors": options.get("colors"),
"headers": build_custom_headers(options.get("headers")),
"eager": build_eager(options.get("eager")),
"use_filename": options.get("use_filename"),
"unique_filename": options.get("unique_filename"),
"discard_original_filename": options.get("discard_original_filename"),
"invalidate": options.get("invalidate"),
"notification_url": options.get("notification_url"),
"eager_notification_url": options.get("eager_notification_url"),
"eager_async": options.get("eager_async"),
"proxy": options.get("proxy"),
"folder": options.get("folder"),
"overwrite": options.get("overwrite"),
"tags": options.get("tags") and ",".join(build_array(options["tags"])),
"allowed_formats": options.get("allowed_formats") and ",".join(
"face_coordinates": encode_double_array(options.get("face_coordinates")),
"custom_coordinates": encode_double_array(options.get("custom_coordinates")),
"context": encode_context(options.get("context")),
"moderation": options.get("moderation"),
"raw_convert": options.get("raw_convert"),
"quality_override": options.get("quality_override"),
"quality_analysis": options.get("quality_analysis"),
"ocr": options.get("ocr"),
"categorization": options.get("categorization"),
"detection": options.get("detection"),
"similarity_search": options.get("similarity_search"),
"background_removal": options.get("background_removal"),
"upload_preset": options.get("upload_preset"),
"phash": options.get("phash"),
"return_delete_token": options.get("return_delete_token"),
"auto_tagging": options.get("auto_tagging") and str(options.get("auto_tagging")),
"responsive_breakpoints": generate_responsive_breakpoints_string(
"async": options.get("async"),
"access_control": options.get("access_control") and json_encode(
return params
def __process_text_options(layer, layer_parameter):
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:
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) and layer.startswith("fetch:"):
layer = {"url": layer[len('fetch:'):]}
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 = layer.get("url")
components = list()
if text is not None and resource_type is None:
resource_type = "text"
if fetch and resource_type is None:
resource_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 resource_type != "fetch":
raise ValueError("Must supply public_id for for non-text " + layer_parameter)
if resource_type is not None and resource_type != "image":
if type is not None and type != "upload":
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:
if public_id is not None:
public_id = public_id.replace("/", ':')
if text is not None:
var_pattern = VAR_NAME_RE
match = re.findall(var_pattern, text)
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(smart_escape(smart_escape(part, r"([,/])")))
text = ''.join(encoded_text)
# text = text.replace("%2C", "%252C")
# text = text.replace("/", "%252F")
elif resource_type == "fetch":
b64 = base64_encode_url(fetch)
public_id = public_id.replace("/", ':')
return ':'.join(components)
"=": 'eq',
"!=": 'ne',
"<": 'lt',
">": 'gt',
"<=": 'lte',
">=": 'gte',
"&&": 'and',
"||": 'or',
"*": 'mul',
"/": 'div',
"+": 'add',
"-": 'sub'
"aspect_ratio": "ar",
"current_page": "cp",
"face_count": "fc",
"height": "h",
"initial_aspect_ratio": "iar",
"initial_height": "ih",
"initial_width": "iw",
"page_count": "pc",
"page_x": "px",
"page_y": "py",
"tags": "tags",
"width": "w"
replaceRE = "((\\|\\||>=|<=|&&|!=|>|=|<|/|-|\\+|\\*)(?=[ _])|" + '|'.join(PREDEFINED_VARS.keys()) + ")"
def translate_if(match):
name =
return IF_OPERATORS.get(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 __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
url = unquote(url)
except Exception:
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(), os.SEEK_END)
size = file_io.tell(), 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
You can’t perform that action at this time.