Created
August 25, 2016 22:10
-
-
Save aparakian/c56d116ea1a4d868d17c923f85c1ad28 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from collections import Counter | |
| from ads_app.api import AdsApi, get_error_list | |
| from ads_app.api.objects import AdLabel | |
| from sjp_app.errors import CampaignWarning | |
| from sjp_app.utils import stats | |
| class BaseFacebookCampaignHandler(object): | |
| def __init__(self, sjp_campaign): | |
| super(BaseFacebookCampaignHandler, self).__init__() | |
| self.api = AdsApi() | |
| self.sjp_campaign = sjp_campaign | |
| self.campaign_id = str(sjp_campaign.id) | |
| self.fb_campaign = sjp_campaign.fb_campaign | |
| def create_image_adlabels(self, images): | |
| """ | |
| """ | |
| batch = self.api.new_batch() | |
| image_adlabels = [] | |
| for image in images: | |
| adlabel = AdLabel(parent_id=self.api.account_id) | |
| adlabel.remote_create(params={ | |
| AdLabel.Field.name: AdLabel.IMAGE_LABEL_PREFIX % image.google_id | |
| }, batch=batch) | |
| image_adlabels.append({"image": image, "adlabel": adlabel}) | |
| batch.execute() | |
| return image_adlabels | |
| def check_errors_and_alert(self, event_params, *object_list): | |
| """ | |
| """ | |
| successes, errors = self._compute_errors(*object_list) | |
| text = self._process_alert_text(event_params.get('text', ""), successes, errors) | |
| stats._send_ads_sjp_event_to_datadog(self.sjp_campaign, event_params['title'], text) | |
| if len(errors) > len(successes): | |
| raise Exception # TBD | |
| if errors: | |
| raise CampaignWarning( | |
| "Warning during %s for %s" % (event_params['title'], self.sjp_campaign), | |
| [e.error for e in errors] | |
| ) | |
| def _process_alert_text(self, text, successes, errors): | |
| """ | |
| """ | |
| success_summary = Counter([s.__class__.__name__ for s in successes]) | |
| error_summary = Counter([e.__class__.__name__ for e in errors]) | |
| text += " Successes: %s" % ('%s %s,' % (v, k) for k, v in success_summary) | |
| text += " Errors: %s" % ('%s %s,' % (v, k) for k, v in error_summary) | |
| return text | |
| def _compute_errors(self, *object_list): | |
| """ | |
| """ | |
| successes = [] | |
| errors = [] | |
| for o in object_list: | |
| if isinstance(o, list): | |
| s, e = self._compute_errors(o) | |
| successes.extend(s) | |
| errors.extend(e) | |
| if o.has_error: | |
| errors.append(o) | |
| else: | |
| successes.append(o) | |
| return successes, errors |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment