Yandex.Direct API client (see http://api.yandex.com/direct/)

README.md

Usage

Instantiate with a path to the directory containing the SSL certificate and key (issued by Yandex.Direct).

from directmanager import DirectManager

manager = DirectManager('~/.certificates')
manager.set_cid(2139432)
umanager.minimum(1.2, 0.05)

License

This work is licensed under a Creative Commons Attribution 3.0 Unported License.

directmanager.py

#!/usr/bin/python3

import ssl
import http.client
import json

class Connection():
    """Manages an HTTPS connecton to Yandex.Direct JSON API v4."""

    HOST = 'soap.direct.yandex.ru'
    METHOD = 'POST'
    PATHNAME = '/json-api/v4/'
    ENCODING = 'utf-8'

    # Create a connection immediately if a certificate path provided.
    def __init__(self, path):
        if path is None:
            self.h = None
        else:
            self.set_cert_path(path)
            self.connect()

    # Store the certificate files path.
    def set_cert_path(self, path):
        """Sets the certificate files' path."""

        self.cert_path = path

    def connect(self):
        self.h = self.create()

    def create(self):
        """Creates the HTTPS connection using certificate and key files."""

        # SSL certificate files.
        cert = self.cert_path + '/cert.crt'
        key  = self.cert_path + '/private.key'
        ca   = self.cert_path + '/cacert.pem'

        # SSL context.
        context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
        context.verify_mode = ssl.CERT_REQUIRED
        context.load_cert_chain(cert, key)
        context.load_verify_locations(ca)

        # Create a connection.
        return http.client.HTTPSConnection(
            self.HOST,
            http.client.HTTPS_PORT,
            context=context
        )

    # Close the connection.
    def close(self):
        """Closes the HTTPS connection."""

        return self.h.close()

    # Proxy the API.
    def send(self, method, param):
        """
        Calls an API method with optional parameters
        by sending a POST-request via the HTTPS connection.

        Returns a native object, loaded from a JSON string.
        """

        data = {'method': method}

        if param:
            data['param'] = param

        data = json.dumps(data)

        self.h.request(self.METHOD, self.PATHNAME, data)

        # Parse the JSON response.
        resp = json.loads(
            self.h.getresponse().read().decode(self.ENCODING)
        );

        data = resp.get('data')

        # Return the data or print errors.
        if data is None:
            print(
                'Error',
                resp.get('error_code'),
                resp.get('error_str'),
                resp.get('error_detail')
            )
        else:
            return data

class DirectManager():
    # The price minimum.
    MIN = 0.01

    # Create a connection.
    def __init__(self, cert_path):
        self.api = Connection(cert_path)

    def set_cid(self, cid):
        self.cid = [cid]

    # Extract fields by name.
    def get_fields(self, table, field):
        return [item.get(field) for item in table]

    # Get active phrases and update prices.
    def process(self, fields, calc):
        filters = {'IsActive': ['Yes']}

        # Get banner IDs.
        banners = self.api.send('GetBanners', {
            'Filter': filters,
            'CampaignIDS': self.cid
        })
        banners = self.get_fields(banners, 'BannerID')

        # Always load the current price.
        fields.append('Price')

        # Get prices.
        phrases = self.api.send('GetBannerPhrasesFilter', {
            'BannerIDS': banners,
            'FieldsNames': fields
        })

        to_update = []

        for p in phrases:
            # Calculate a new price.
            price = calc(p)

            # Compare the current and new prices.
            if price != p.get('Price'):
                to_update.append({
                    'CampaignID' : p.get('CampaignID'),
                    'BannerID'   : p.get('BannerID'),
                    'PhraseID'   : p.get('PhraseID'),
                    'Price'      : price
                })

        # Do the update if necessary.
        length = len(to_update)

        if length > 0:
            if self.api.send('UpdatePrices', to_update) is 1:
                print('Success:', length, 'prices updated.')
        else:
            print('Checked', len(phrases), 'prices, nothing to update.')

        self.api.close()


    # Strategies.

    # The minimal expence strategy.
    def minimum(self, maxlim, step):
        fields = ['Min', 'PremiumMin']

        def calc(p):
            min_ = p.get('Min', self.MIN)
            pr_min = p.get('PremiumMin', self.MIN)

            if min_ > pr_min:
                min_ = pr_min

            if min_ > maxlim:
                min_ = self.MIN
            else:
                min_ += step

            return min_

        return self.process(fields, calc)

    # The premium position strategy.
    def premium(self, maxlim, step):
        fields = ['Min', 'PremiumMin']

        def calc(p):
            min_ = p.get('Min', self.MIN)
            pr_min = p.get('PremiumMin', self.MIN)

            if pr_min > maxlim:
                if maxlim < min_:
                    pr_min = min_
                else:
                    pr_min = self.MIN
            else:
                pr_min += step

            return pr_min

        return self.process(fields, calc)