import base64 import hashlib import hmac import logging from collections import namedtuple import argparse import requests from requests.models import PreparedRequest try: from urllib.parse import urlunparse, urlparse, unquote_plus, parse_qs except ImportError: from urlparse import urlunparse, urlparse, parse_qs from urllib import unquote_plus logger = logging.getLogger('User Data API') ''' This script generates signed url by given parameters, and sending get request the elementary parameters you must supply in are: siteid, bkuid, bksecretkey, in addition should use at least one of the next parameters to supply for identify the user for getting or sending data about: -userid -puserid -adid -idfa In addition you can add more optional parameters which let you get/send more or more specific information by using one of the next options: *puserid, phint , target, filterbycampids *puserid must be followed by pfield or be known in advance to Oracle ''' # the next parameter is tuple of ("arg shortcut name", "arg name", helper") cmd_args = namedtuple('cmd_args', ['short', 'arg_name', 'helper']) required_parameters = [ cmd_args("-s", "--siteid", "The unique site identifier generated when you created your " "container"), cmd_args("-b", "--bkuid", "Your web service user key"), cmd_args("-k", "--bksecretkey", "your web service private key"), ] optional_parameters = [ cmd_args("-u", "--userid", "The obfuscated Oracle Data Cloud unique user ID " "that you received via an ID swap, SDT delivery, or JSON Return " "tag."), cmd_args("-p", "--puserid", "Your partner-based unique user ID (PUUID), " "which may be an email hash, account ID hash, " "cookie ID, " "or other identifier.\nThe PUUIDs you pass " "must be ID swapped " "to BKUUIDs or you will get 404 user not " "found errors"), cmd_args("-f", "--pfield", "The type of key associated with the partner-based " "UUID you passed in the puserid field."), cmd_args("-d", "--adid", " enables you to send and get the data linked to this " "mobile advertising ID, which operates in a primary ID " "space"), cmd_args("-e", "--idfa", " enables you to send and get the data linked to this " "mobile advertising ID, which operates in a primary ID " "space"), cmd_args("-c", "--campIds", "A list of comma-delimited campaign IDs" "This field is used for filtering the " "categories included in the user data API " "response based on the specified campaign IDs " "that have " "targeted and won the user."), cmd_args("-t", "--target", "indicate whether the ability to use a set of " "conditions in a schedule that determine " "whether a tag is fired"), ] optional_multiply_parameters = [ cmd_args("-n", "--phint", " A key-value pair representing the user's attributes " "and behavior, which you included in your rules") ] def sign_string(bk_secret_key, str_to_sign): h = hmac.new(bytes(bk_secret_key.encode('utf-8')), bytes( str_to_sign.encode('utf-8')), hashlib.sha256) return base64.standard_b64encode(h.digest()) def generate_string_to_sign(url, method): parsed_url = urlparse(url) query_arr = parse_qs(parsed_url.query) query_values = ''.join([''.join(v) for v in query_arr.values()]) return method + parsed_url.path + query_values def get_parsed_args_dict(): parser = argparse.ArgumentParser() named_required = parser.add_argument_group('required arguments') named = parser.add_argument_group('extra optional arguments') named_multiply = parser.add_argument_group('positional multiply arguments') for current_arg_detail in optional_parameters: named.add_argument(current_arg_detail.short, current_arg_detail.arg_name, help=current_arg_detail.helper) for current_arg_detail in required_parameters: named_required.add_argument(current_arg_detail.short, current_arg_detail.arg_name, help=current_arg_detail.helper, required=True) for current_arg_detail in optional_multiply_parameters: named_multiply.add_argument(current_arg_detail.short, current_arg_detail.arg_name, help=current_arg_detail.helper, action='append', nargs='*') args = parser.parse_args() return vars(args) def filter_args_dict(args_dict): return {k: v for k, v in args_dict.items() if v} def build_url_by_args(): args_in_use = filter_args_dict(get_parsed_args_dict()) # handle special fields which are not related to query part of url siteid = args_in_use.pop("siteid", None) bkuid = args_in_use.pop("bkuid", None) bksecretkey = args_in_use.pop("bksecretkey", None) url_before_sign = urlunparse( ('https', 'api.tags.bluekai.com', '/getdata/' + siteid + '/v1.2', None, None, None)) req = PreparedRequest() req.prepare_url(url_before_sign, args_in_use) return req.url, bkuid, bksecretkey def get_signed_url(url_before_sign, bk_uid, bk_secret_key): if not bk_uid or not bk_secret_key: raise ValueError('bkuid or bksecretkey are required and missing') url_before_sign = unquote_plus(url_before_sign) string_to_sign = generate_string_to_sign(url_before_sign, "GET") signature = sign_string(bk_secret_key, string_to_sign) params = {'bkuid': bk_uid, "bksig": signature} req = PreparedRequest() req.prepare_url(url_before_sign, params) return req.url def do_request(url): result = requests.get(url) logger.info("The request you create is\n{}\n".format(url)) raw_data = result.content logger.info("the request result is:\n {}\n".format(str(raw_data))) return raw_data def main(): logging.basicConfig(level=logging.INFO, format='%(name)s - %(levelname)s - %(' 'message)s') print('\nGenerates a signed url using the given arguments and sends a ' 'GET request.') url_by_param, bk_uid, bk_secret_key = build_url_by_args() signed_url = get_signed_url(url_by_param, bk_uid, bk_secret_key) do_request(signed_url) if __name__ == "__main__": main()