Headline Testing using v2 REST API

headline_testing.py

import datetime
import os
import threading
import urllib

import pytz
import requests
from bravado import requests_client
from bravado.client import SwaggerClient


class ApiKeyHeaderAuthenticator(requests_client.Authenticator):
    """?api_key authenticator.
    This authenticator adds a header to specify an API key.
    :param host: Host to authenticate for.
    :param api_key: API key.
    :param param_name: Header name specifying the API key.
    """

    def __init__(self, host, api_key, param_name='api_key'):
        super(ApiKeyHeaderAuthenticator, self).__init__(host)
        self.param_name = param_name
        self.api_key = api_key

    def apply(self, request):
        request.headers[self.param_name] = self.api_key
        return request



print('Welcome to the command-line Headling Testing tool!')
token = os.environ.get('ACCESS_TOKEN') 
if token is None:
    token = input('Enter Optimizely OAuth Token: ')
http_client = requests_client.RequestsClient()
http_client.authenticator = ApiKeyHeaderAuthenticator(
    host='api.optimizely.com',
    api_key='Bearer {}'.format(token),
    param_name='Authorization'
)
client = SwaggerClient.from_url(
        'https://api.optimizely.com/v2/swagger.json',
        http_client=http_client,
)

Project = client.get_model('Project')
Experiment = client.get_model('Experiment')
Page = client.get_model('Page')
Variation = client.get_model('Variation')
Action = client.get_model('Action')
Metric = client.get_model('Metric')
Change = client.get_model('Change')


def get_projects():
    client = SwaggerClient.from_url(
            'https://api.optimizely.com/v2/swagger.json',
            http_client=http_client,
            config={
                'also_return_response': True,
            }
    )
    projects, response = client.Projects.list_projects().result()
    while 'rel=next' in response.headers['link']:
        next_page_rel = response.headers['link'].split(',')[0]
        next_page_url = next_page_rel.split(';')[0].strip('<').strip('>')
        next_page_url = urllib.parse.urlparse(next_page_url)
        page_params = urllib.parse.parse_qs(next_page_url.query)
        more_projects, response = client.Projects.list_projects(
                per_page=int(page_params['per_page'][0]),
                page=int(page_params['page'][0])).result()
        projects += more_projects
    return projects


def run_headline_test():
    projects = get_projects()
    print('Here are your projects:')
    for p in projects:
        if p.status == 'active':
            print('{}: {}'.format(p.id, p.name))
    project_id = input('Enter the project id you want to use (or nothing to create a new one): ')
    if project_id:
        project = [i for i in filter(lambda p: p if p.id == int(project_id) else None, projects)][0]
    else:
        name = input('What do you want to call it? ')
        p = Project(name=name)
        project, response = client.Projects.create_project(body=p).result()
        print('Created new project named "{}" with id "{}"'.format(project.name, project.id))

    article_url = input('Page URL: ')
    original_headline = input('Original Headline: ')
    headline_variation = input('Variation Headline: ')
    print('Thanks for the info! Creating your new experiment now...')

    # Create the page for the article.
    # I need to set this if I'm targeting a single URL???
    conditions = '["and", ["or", {"match_type": "simple", "type": "url", "value": "%s"}]]'
    conditions = conditions % article_url
    activation_code = 'function pollingFn() { return document.documentElement.textContent || document.documentElement.innerText.indexOf("%s"); }' % original_headline
    page = Page(name='{}: {}'.format(article_url, original_headline),
                project_id=project.id, edit_url=article_url,
                page_type='single_url', activation_type='polling',
                activation_code=activation_code,
                platform='web', category='other', conditions=conditions)
    page = client.Pages.create_page(body=page).result()
    print('Created a page for {} with ID {}'.format(article_url, page.id))

    action = Action(page_id=page.id, changes=[])
    variation1 = Variation(name=original_headline, weight=5000, actions=[action])

    code = '''
    var utils = window['optimizely'].get('utils');
    utils.waitForElement('body').then(function(body){
        body.innerHTML = body.innerHTML.replace("%s", "%s");
    });
    ''' % (original_headline, headline_variation)
    #code = code.format(original_headline, headline_variation)
    change = Change(async=False, dependencies=[],
                    type='custom_code', value=code) 
    action = Action(page_id=page.id, changes=[change])
    variation2 = Variation(name=headline_variation, weight=5000, actions=[action])

    metric = Metric(
        aggregator='unique',
        event_id=page.id,
        scope='session'
    )
    experiment = Experiment(project_id=project.id,
                            name='Headline Testing - {}'.format(original_headline),
                            variations=[variation1, variation2], status='active',
                            metrics=[metric])
    experiment = client.Experiments.create_experiment(body=experiment).result()
    print('Created experiment with ID {} - Campaign ID {}'.format(experiment.id, experiment.campaign_id))


def get_results():
    projects = client.Projects.list_projects().result()
    for p in projects:
        if p.status == 'active':
            print('{}: {}'.format(p.id, p.name))
    project_id = int(input('Enter projcet ID to get experiments: '))
    experiments = client.Experiments.list_experiments(project_id=project_id).result()
    for e in experiments:
        print('{}: {}'.format(e.id, e.name))
    experiment_id = int(input('Enter experiment ID to get results: '))
    results = client.Experiments.get_experiment_results(experiment_id=experiment_id,
            start_time=datetime.datetime.strptime('2016-10-20', '%Y-%m-%d').replace(tzinfo=pytz.utc)).result()
    print('Number of people reached: {}'.format(results.reach.total_count))

    variation_results = results.metrics[0].variation_results
    variant = None
    baseline = None
    for k, v in variation_results.items():
        if v.is_baseline:
            baseline = v
        else:
            variant = v
    # I think I can also do this via treatment_count and baseline_count, but I
    # *think* treatment_count would contain the total across all variations.
    print('Number of people in baseline "{}": {}'.format(
        baseline.name,
        results.reach.variations[baseline.variation_id].count))
    print('Number of people in Variation "{}": {}'.format(
        variant.name,
        results.reach.variations[variant.variation_id].count))

    baseline_conclusive = getattr(baseline.lift, 'is_most_conclusive', False)
    variant_conclusive = getattr(variant.lift, 'is_most_conclusive', False)
    if not baseline_conclusive and not variant_conclusive:
        print('Experiment is not yet conclusive.')
    else:
        print('Experiment is conclusive!')
        # TODO: Print winner


choice = input('Do you want to:\n1. Run a headline test.\n2. Get results.\nChoice (1 or 2): ')
choices = {
    '1': run_headline_test,
    '2': get_results,
}
choices[choice]()

requirements.txt

attrs==16.2.0
bravado==8.4.0
bravado-core==4.5.1
cffi==1.8.3
crochet==1.5.0
cryptography==1.5.2
fido==4.1.0
idna==2.1
jsonschema==2.5.1
pyasn1==0.1.9
pyasn1-modules==0.0.8
pycparser==2.14
pyOpenSSL==16.1.0
python-dateutil==2.5.3
pytz==2016.7
PyYAML==3.12
requests==2.11.1
rfc3987==1.3.7
service-identity==16.0.0
simplejson==3.8.2
six==1.10.0
strict-rfc3339==0.7
swagger-spec-validator==2.0.2
Twisted==16.4.1
webcolors==1.5
wheel==0.26.0
yelp-bytes==0.3.0
yelp-encodings==0.1.3
zope.interface==4.3.2