path: root/aw_api/soappy_toolkit/SanityCheck.py

#!/usr/bin/python
#
# Copyright 2009 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""Validation and type conversion functions."""

__author__ = 'api.sgrinberg@gmail.com (Stan Grinberg)'

import cgi
import re

from aw_api import SanityCheck as glob_sanity_check
from aw_api import Utils
from aw_api.Errors import MissingPackageError
from aw_api.soappy_toolkit import MIN_SOAPPY_VERSION
try:
  import SOAPpy
except ImportError:
  msg = 'SOAPpy v%s or newer is required.' % MIN_SOAPPY_VERSION
  raise MissingPackageError(msg)
else:
  if (map(eval, SOAPpy.version.__version__.split('.')) <
      (list(map(eval, MIN_SOAPPY_VERSION.split('.'))))):
    msg = 'SOAPpy v%s or newer is required.' % MIN_SOAPPY_VERSION
    raise MissingPackageError(msg)


def UnType(item):
  """Convert given string into untyped type.

  Args:
    item: str string to untype.

  Returns:
    untypedType string converted into untypedType.
  """
  glob_sanity_check.ValidateTypes(((item, (str, unicode)),))

  # HTML encode non-complex strings. Complex strings would be XML snippets,
  # like <networkTypes>GoogleSearch</networkTypes>.
  pattern = re.compile('<.*>|</.*>')
  if pattern.search(item) is None:
    pattern = re.compile('&#(x\w\w|\d{3});')
    result = pattern.findall(item)
    # Escape only ASCII characters.
    if not result:
      item = cgi.escape(item)

  return SOAPpy.Types.untypedType(item)


def IsUnTypedClass(obj):
  """Return True if a given object is of type SOAPpy.Types.untypedType, False
  otherwise.

  Args:
    obj: object an object to check.

  Returns:
    bool True if a given object is untyped, False otherwise.
  """
  return isinstance(obj, SOAPpy.Types.untypedType)


def ValidateAccountInfoV13(acct_info):
  """Validate AccountInfo object.

  Args:
    acct_info: dict AccountInfo object with updated values.
  """
  glob_sanity_check.ValidateTypes(((acct_info, dict),))
  for key in acct_info:
    if key == 'defaultNetworkTargeting':
      glob_sanity_check.ValidateTypes(((acct_info[key], list),))
      items = []
      for sub_key in acct_info[key]:
        glob_sanity_check.ValidateTypes(((sub_key, (str, unicode)),))
        items.append('<networkTypes>%s</networkTypes>' % sub_key)
      acct_info[key] = UnType(''.join(items))
    elif key == 'emailPromotionsPreferences':
      glob_sanity_check.ValidateTypes(((acct_info[key], dict),))
      for sub_key in acct_info[key]:
        glob_sanity_check.ValidateTypes(((acct_info[key][sub_key],
                                          (str, unicode)),))
        acct_info[key][sub_key] = UnType(acct_info[key][sub_key])
    else:
      glob_sanity_check.ValidateTypes(((acct_info[key], (str, unicode)),))


def ValidateAdGroupV13(ad_group):
  """Validate AdGroup object.

  Args:
      ad_group: dict information for the ad group.
  """
  glob_sanity_check.ValidateTypes(((ad_group, dict),))
  for key in ad_group:
    if IsUnTypedClass(ad_group[key]): continue
    glob_sanity_check.ValidateTypes(((ad_group[key], (str, unicode)),))
    ad_group[key] = UnType(ad_group[key])


def ValidateAdV13(ad):
  """Validate Ad object.

  Args:
    ad: dict information for the ad.
  """
  glob_sanity_check.ValidateTypes(((ad, dict),))
  for key in ad:
    if key == 'markupLanguages' or key == 'mobileCarriers':
      glob_sanity_check.ValidateTypes(((ad[key], list),))
      for item in ad[key]:
        glob_sanity_check.ValidateTypes(((item, (str, unicode)),))
        item = UnType(item)
    elif (key == 'image' or key == 'businessImage' or key == 'customIcon' or
          key == 'productImage' or key == 'video'):
      glob_sanity_check.ValidateTypes(((ad[key], dict),))
      for sub_key in ad[key]:
        glob_sanity_check.ValidateTypes(((ad[key][sub_key], (str, unicode)),))
        ad[key][sub_key] = UnType(ad[key][sub_key])
    else:
      glob_sanity_check.ValidateTypes(((ad[key], (str, unicode)),))
      if key == 'description1' or key == 'description2' or key == 'headline':
        ad[key] = Utils.MakeTextXMLReady(ad[key])
      ad[key] = UnType(ad[key])


def ValidateOptimizerSettingsV13(optimizer):
  """Validate BudgetOptimizerSettings or ConversionOptimizerSettings object.

  Args:
    optimizer: dict optimizer settings for this entity.

  Returns:
    str optimizer settings converted into str type.
  """
  glob_sanity_check.ValidateTypes(((optimizer, dict),))
  items = []
  for key in optimizer:
    items.append('<%s>%s</%s>' % (key, optimizer[key], key))

  return ''.join(items)


def ValidateLanguageTargetV13(language_target):
  """Validate LanguageTarget object.

  Args:
    language_target: list languages targeted by this entity.

  Returns:
    str languages targeted converted into str type.
  """
  glob_sanity_check.ValidateTypes(((language_target, list),))
  items = []
  for item in language_target:
    glob_sanity_check.ValidateTypes(((item, (str, unicode)),))
    items.append('<languages>%s</languages>' % item)

  return ''.join(items)


def ValidateGeoTargetV13(geo_target):
  """Validate GeoTarget object.

  Args:
    geo_target: dict geographic targeting rules for this entity.

  Returns:
    str geographic targeting converted into str type.
  """
  glob_sanity_check.ValidateTypes(((geo_target, dict),))
  items = []
  for key in geo_target:
    if key == 'targetAll':
      glob_sanity_check.ValidateTypes(((geo_target[key], (str, unicode)),))
      items.append('<%s>%s</%s>' % (key, geo_target[key], key))
    elif key == 'cityTargets':
      glob_sanity_check.ValidateTypes(((geo_target[key], dict),))
      items.append('<%s>' % key)
      for sub_key in geo_target[key]:
        glob_sanity_check.ValidateTypes(((geo_target[key][sub_key], list),))
        for item in geo_target[key][sub_key]:
          items.append('<%s>%s</%s>' % (sub_key, item, sub_key))
      items.append('</%s>' % key)
    elif key == 'countryTargets':
      glob_sanity_check.ValidateTypes(((geo_target[key], dict),))
      items.append('<%s>' % key)
      for sub_key in geo_target[key]:
        glob_sanity_check.ValidateTypes(((geo_target[key][sub_key], list),))
        for item in geo_target[key][sub_key]:
          items.append('<%s>%s</%s>' % (sub_key, item, sub_key))
      items.append('</%s>' % key)
    elif key == 'metroTargets':
      glob_sanity_check.ValidateTypes(((geo_target[key], dict),))
      items.append('<%s>' % key)
      for sub_key in geo_target[key]:
        glob_sanity_check.ValidateTypes(((geo_target[key][sub_key], list),))
        for item in geo_target[key][sub_key]:
          items.append('<%s>%s</%s>' % (sub_key, item, sub_key))
      items.append('</%s>' % key)
    elif key == 'regionTargets':
      glob_sanity_check.ValidateTypes(((geo_target[key], dict),))
      items.append('<%s>' % key)
      for sub_key in geo_target[key]:
        glob_sanity_check.ValidateTypes(((geo_target[key][sub_key], list),))
        for item in geo_target[key][sub_key]:
          items.append('<%s>%s</%s>' % (sub_key, item, sub_key))
      items.append('</%s>' % key)
    elif key == 'proximityTargets':
      glob_sanity_check.ValidateTypes(((geo_target[key], dict),))
      items.append('<%s>' % key)
      for sub_key in geo_target[key]:
        glob_sanity_check.ValidateTypes(((geo_target[key][sub_key], list),))
        for item in geo_target[key][sub_key]:
          items.append('<%s>' % sub_key)
          glob_sanity_check.ValidateTypes(((item, dict),))
          for sub_sub_key in item:
            glob_sanity_check.ValidateTypes(((item[sub_sub_key],
                                              (str, unicode)),))
            items.append('<%s>%s</%s>' % (sub_sub_key, item[sub_sub_key],
                                          sub_sub_key))
          items.append('</%s>' % sub_key)
      items.append('</%s>' % key)
    else:
      pass

  return ''.join(items)


def ValidateNetworkTargetV13(network_target):
  """Validate NetworkTarget object.

  Args:
    network_target: list advertising networks targeted by this entity.

  Returns:
    str adertising networks converted into str type.
  """
  glob_sanity_check.ValidateTypes(((network_target, list),))
  items = []
  for item in network_target:
    glob_sanity_check.ValidateTypes(((item, (str, unicode)),))
    items.append('<networkTypes>%s</networkTypes>' % item)

  return ''.join(items)


def ValidateAdScheduleV13(ad_schedule):
  """Validate AdSchedule object.

  Args:
    ad_schedule: dict ad scheduling setting for this entity.

  Returns:
    str ad scheduling setting converted into str type.
  """
  glob_sanity_check.ValidateTypes(((ad_schedule, dict),))
  items = []
  for key in ad_schedule:
    if key == 'status':
      glob_sanity_check.ValidateTypes(((ad_schedule[key], (str, unicode)),))
      items.append('<%s>%s</%s>' % (key, ad_schedule[key], key))
    elif key == 'intervals':
      glob_sanity_check.ValidateTypes(((ad_schedule[key], list),))
      for item in ad_schedule[key]:
        glob_sanity_check.ValidateTypes(((item, dict),))
        sub_items = ['<intervals>']
        for sub_key in item:
          glob_sanity_check.ValidateTypes(((item[sub_key], (str, unicode)),))
          sub_items.append('<%s>%s</%s>' % (sub_key, item[sub_key], sub_key))
        sub_items.append('</intervals>')
        items.append(''.join(sub_items))
    else:
      pass

  return ''.join(items)


def ValidateCampaignV13(campaign):
  """Validate Campaign object.

  Args:
    campaign: dict information for the campaign.
  """
  glob_sanity_check.ValidateTypes(((campaign, dict),))
  for key in campaign:
    if (key == 'budgetAmount' or key == 'budgetPeriod' or
        key == 'contentTargeting' or key == 'endDay' or key == 'id' or
        key == 'name' or key == 'startDay' or key == 'status'):
      glob_sanity_check.ValidateTypes(((campaign[key], (str, unicode)),))
      campaign[key] = UnType(campaign[key])
    elif (key == 'budgetOptimizerSettings' or
          key == 'conversionOptimizerSettings'):
      campaign[key] = UnType(ValidateOptimizerSettingsV13(campaign[key]))
    elif key == 'geoTargeting':
      campaign[key] = UnType(ValidateGeoTargetV13(campaign[key]))
    elif key == 'languageTargeting':
      campaign[key] = UnType(ValidateLanguageTargetV13(campaign[key]))
    elif key == 'networkTargeting':
      campaign[key] = UnType(ValidateNetworkTargetV13(campaign[key]))
    elif key == 'schedule':
      campaign[key] = UnType(ValidateAdScheduleV13(campaign[key]))
    else:
      pass


def ValidateCriterionV13(criterion):
  """Validate criterion object.

  Args:
    criterion: dict information for the criterion.
  """
  glob_sanity_check.ValidateTypes(((criterion, dict),))
  for key in criterion:
    glob_sanity_check.ValidateTypes(((criterion[key], (str, unicode)),))
    if key == 'text':
      criterion[key] = Utils.MakeTextXMLReady(criterion[key])
    criterion[key] = UnType(criterion[key])


def ValidateSeedKeywordV13(seed_keyword):
  """Validate SeedKeyword object.

  Args:
    seed_keyword: dict information for keyword to derive variations from.
  """
  glob_sanity_check.ValidateTypes(((seed_keyword, dict),))
  for key in seed_keyword:
    glob_sanity_check.ValidateTypes(((seed_keyword[key], (str, unicode)),))
    seed_keyword[key] = UnType(seed_keyword[key])


def ValidateDemographicsTargetV13(demo):
  """Validate DemographicsTarget object.

  Args:
    demo: dict demographics targeting criteria.
  """
  glob_sanity_check.ValidateTypes(((demo, dict),))
  for key in demo:
    glob_sanity_check.ValidateTypes(((demo[key], (str, unicode)),))
    demo[key] = UnType(demo[key])


def ValidateLanguageGeoTargetingV13(targeting):
  """Validate LanguageGeoTargeting object.

  Args:
    targeting: dict language and geo targeting information.
  """
  glob_sanity_check.ValidateTypes(((targeting, dict),))
  for key in targeting:
    glob_sanity_check.ValidateTypes(((targeting[key], list),))
    for sub_key in targeting[key]:
      glob_sanity_check.ValidateTypes(((sub_key, (str, unicode)),))


def ValidateKeywordTrafficRequestV13(request):
  """Validate KeywordTrafficRequest object.

  Args:
    request: dict keyword traffic request.
  """
  glob_sanity_check.ValidateTypes(((request, dict),))
  for key in request:
    glob_sanity_check.ValidateTypes(((request[key], (str, unicode)),))
    request[key] = UnType(request[key])


def ValidateKeywordRequestV13(request):
  """Validate KeywordRequest object.

  Args:
    request: dict keyword request.

  Returns:
    str keyword request converted into str type.
  """
  glob_sanity_check.ValidateTypes(((request, dict),))
  items = []
  for key in request:
    glob_sanity_check.ValidateTypes(((request[key], (str, unicode)),))
    items.append('<%s>%s</%s>' % (key, request[key], key))

  return ''.join(items)


def ValidateAdGroupRequestV13(request):
  """Validate AdGroupRequest object.

  Args:
    request: dict ad group request.

  Returns:
    str ad group request converted into str type.
  """
  glob_sanity_check.ValidateTypes(((request, dict),))
  items = []
  for key in request:
    if key == 'keywordRequests':
      items.append('<keywordRequests>')
      glob_sanity_check.ValidateTypes(((request[key], list),))
      for item in request[key]:
        items.append(ValidateKeywordRequestV13(item))
      items.append('</keywordRequests>')
    else:
      glob_sanity_check.ValidateTypes(((request[key], (str, unicode)),))
      items.append('<%s>%s</%s>' % (key, request[key], key))

  return ''.join(items)


def ValidateCampaignRequestV13(request):
  """Validate CampaignRequest object.

  Args:
    request: dict campaign request.

  Returns:
    str campaign request converted into str type.
  """
  glob_sanity_check.ValidateTypes(((request, dict),))
  items = []
  for key in request:
    if key == 'adGroupRequests':
      glob_sanity_check.ValidateTypes(((request[key], list),))
      items.append('<adGroupRequests>')
      for item in request[key]:
        items.append(ValidateAdGroupRequestV13(item))
      items.append('</adGroupRequests>')
    elif key == 'geoTargeting':
      glob_sanity_check.ValidateTypes(((request[key], dict),))
      items.append(('<geoTargeting>%s</geoTargeting>'
                    % ValidateGeoTargetV13(request[key])))
    elif key == 'id':
      glob_sanity_check.ValidateTypes(((request[key], (str, unicode)),))
      items.append('<%s>%s</%s>' % (key, request[key], key))
    elif key == 'languageTargeting':
      glob_sanity_check.ValidateTypes(((request[key], list),))
      items.append('<%s>%s</%s>' % (key,
                                    ValidateLanguageTargetV13(request[key]),
                                    key))
    elif key == 'networkTargeting':
      glob_sanity_check.ValidateTypes(((request[key], list),))
      items.append('<%s>%s</%s>' % (key, ValidateNetworkTargetV13(request[key]),
                                    key))
    else:
      pass

  return ''.join(items)


def ValidateDefinedReportJobV13(job, name_space):
  """Validate DefinedReportJob object.

  Args:
    job: dict report job object.
    name_space: str namespace to use for this ReportJob.

  Returns:
    instance untyped instance of ReportJob.
  """
  items = []
  for key in job:
    if (key == 'adWordsType' or key == 'crossClient' or key == 'endDay' or
        key == 'includeZeroImpression' or key == 'keywordType' or
        key == 'name' or key == 'selectedReportType' or key == 'startDay'):
      glob_sanity_check.ValidateTypes(((job[key], (str, unicode)),))
      items.append('<%s>%s</%s>' % (key, job[key], key))
    else:
      glob_sanity_check.ValidateTypes(((job[key], list),))
      for item in job[key]:
        glob_sanity_check.ValidateTypes(((item, (str, unicode)),))
        items.append('<%s>%s</%s>' % (key, item, key))
  # Explicitly set job's namespace and type.
  job = UnType(''.join(items))
  job._setAttr('xmlns:impl', name_space)
  job._setAttr('xsi3:type', 'impl:DefinedReportJob')

  return job


# TODO(api.sgrinberg): Add validation for Get/Mutate calls.
def ValidateOperation(operation):
  """Validate Operation object.

  Args:
    operation: dict operation object.
  """
  pass


def ValidateSelector(selector):
  """Validate Selector object.

  Args:
    selector: dict selector object.
  """
  pass