Created
August 25, 2016 23:24
-
-
Save aparakian/cfb55967826f1beb9221026b637a963e 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 copy import copy, deepcopy | |
| import logging | |
| from celery import subtask | |
| from ads_app.api.objects import AdSet, Ad | |
| from .base import BaseFacebookCampaignHandler | |
| from sjp_app.ads.utils import get_ad_object_job_id | |
| from sjp_app.utils.campaign_actions import log_campaign_action, ACTIONS | |
| from w4.documents.mongo_job import MongoJob | |
| class FBCampaignUpdater(BaseFacebookCampaignHandler): | |
| """ | |
| This class is used to update a campaign and its related objects. It can | |
| pause ads / adsets, change the campaign budget, bid, etc... | |
| PLEASE MAKE SURE THAT NO CODE IS DUPLICATED BETWEEN THIS CLASS | |
| AND THE Pusher. IF YOU END UP DUPLICATING CODE, IT MEANS YOU'RE | |
| DOING SOMETHING WRONG: EITHER CREATE A UTILS METHOD THAT IS USED | |
| BY BOTH THIS CLASS AND THE Pusher OR EMPOWER OUR OBJECTS IN THE | |
| `ads_app.api.objects` MODULE. | |
| IF YOU'RE LOST AND CANNOT FIND A WAY TO DO THINGS CLEANLY, ASK | |
| THE HELP OF A SENIOR DEVELOPER. | |
| This class is organized as follow: | |
| - campaign related methods | |
| - adset related methods | |
| - ads related methods | |
| - subroutines | |
| Please respect this organization. | |
| """ | |
| def __init__(self, sjp_campaign): | |
| """ | |
| :param sjp_campaign: a SjpCampaign object | |
| """ | |
| super(FBCampaignUpdater, self).__init__(sjp_campaign) | |
| self.fb_campaign_id = sjp_campaign.fb_campaign_id | |
| # adsets related methods | |
| def update_adsets_bid(self): | |
| """ | |
| Update the bid type and value of all adsets in campaign | |
| and send event to datadog | |
| """ | |
| adsets = self.api.get_adset_for_campaign(self.fb_campaign_id) | |
| params = AdSet.get_bid_parameters( | |
| self.sjp_campaign.billing_event, | |
| self.sjp_campaign.optimization_goal, | |
| self.sjp_campaign.bid_value, | |
| self.sjp_campaign.automatic_bid | |
| ) | |
| params.update(AdSet.get_promoted_object(self.sjp_campaign.conversion_event)) | |
| self._update_batch_of_objects(adsets, params) | |
| change = "updating adsets bid" | |
| title = "Bid changed for campaign", | |
| text = "New bid type: %s. New bid value: %s cents." % ( | |
| self.sjp_campaign.billing_event, self.sjp_campaign.bid_value | |
| ) | |
| self.check_errors_and_alert(change, title, text, adsets) | |
| def update_adsets_placement(self): | |
| ''' | |
| Update the placement of all adsets in campaign and send event to datadog | |
| ''' | |
| # we need to read the current targeting to update only the relevant part | |
| adsets = self.api.get_adset_for_campaign(self.fb_campaign_id, fields=[AdSet.Field.targeting]) | |
| params = { | |
| AdSet.Field.targeting: {"page_types": self.sjp_campaign.placement} | |
| } | |
| self._update_batch_of_objects(adsets, params) | |
| change = "updating adsets placement" | |
| title = "Placement changed for campaign" | |
| text = "New placement: %s." % self.sjp_campaign.placement | |
| self.check_errors_and_alert(change, title, text, adsets) | |
| def pause_adsets_on_inactive_jobs(self, check_filters=False): | |
| """ | |
| Pause every adset that relates to an inactive | |
| job, using batch. | |
| :param check_filters: check if your jobs are in campaign filters | |
| """ | |
| adsets = self.api.get_adset_for_campaign(self.fb_campaign_id) | |
| adsets_to_be_paused = self.filter_ad_object_of_inactive_jobs(adsets, check_filters=check_filters) | |
| if adsets_to_be_paused: | |
| with log_campaign_action(ACTIONS.PAUSE_ADSETS, self.sjp_campaign): | |
| self.pause_batch_of_adsets(adsets_to_be_paused) | |
| def update_adsets_end_date(self, end_date): | |
| """ | |
| Update every end_date of each active adset of a campaign. | |
| """ | |
| adsets = self.api.get_adset_for_campaign(self.fb_campaign_id) | |
| params = {AdSet.Field.end_time: end_date.strftime('%Y-%m-%d')} | |
| self._update_batch_of_objects(adsets, params) | |
| change = "updating adsets end_date" | |
| title = "Updated campaign end_date to %s." % self.sjp_campaign.end_date.strftime('%Y-%m-%d') | |
| text = "All adsets of campaign %s had their end_date updated." % self.sjp_campaign.fb_campaign_id | |
| self.check_errors_and_alert(change, title, text, adsets) | |
| def pause_batch_of_adsets(self, adsets): | |
| ''' | |
| Pause all the given adsets and send stats to datadog | |
| ''' | |
| params = {AdSet.Field.status: AdSet.Status.paused} | |
| self._update_batch_of_objects(adsets, params) | |
| change = "pausing adsets" | |
| title = "Adset has been paused for campaign" | |
| self.check_errors_and_alert(change, title, adsets) | |
| # launch redistribution right away | |
| subtask("redistribute_campaign_budget").apply_async(kwargs={'sjp_campaign_id': str(self.sjp_campaign.id)}) | |
| @log_campaign_action(ACTIONS.REDISTRIBUTE_BUDGET) | |
| def update_adsets_daily_budget(self, budget): | |
| """ | |
| Update the daily_budget to the given budget for all active adset in the campaign | |
| Args: | |
| - (int) budget : new daily budget (>0, in cents) | |
| """ | |
| if budget <= 0: | |
| raise ValueError('Budget should be > 0') | |
| adsets = self.api.get_adset_for_campaign(self.fb_campaign_id) | |
| self._update_batch_of_objects(adsets, {AdSet.Field.daily_budget: str(budget)}) | |
| change = "updating adsets daily budget" | |
| self.check_errors_and_alert(change, adsets) | |
| # ads related methods | |
| @log_campaign_action(ACTIONS.ADD_IMAGES) | |
| def add_new_images_to_campaign(self, new_images): | |
| ''' | |
| Update images of all adsets in campaign and send event to datadog | |
| ''' | |
| # create the image adlabels | |
| image_adlabels = self.create_image_adlabels(new_images) | |
| # FIXME: for the carousel ads, we should make sure we add the image in each | |
| # carousel, so the grouping by job_id won't work, instead we should just | |
| # take all the carousel ads and add the image inside | |
| ads = self.api.get_ads_for_campaign(self.fb_campaign_id, fields=[ | |
| Ad.Field.name, | |
| Ad.Field.adset, | |
| "creative{object_story_spec}", | |
| Ad.Field.adlabels, | |
| ]) | |
| # this is a small trick to get 1 ad per (job, adset) | |
| # (a job can be advertised with several adsets) | |
| ads_by_job = { | |
| (ad.adset["id"], get_ad_object_job_id(ad)): ad | |
| for ad in ads | |
| if get_ad_object_job_id(ad) | |
| } | |
| batch = self.api.new_batch() | |
| new_ads = [] | |
| for existing_ad in ads_by_job.itervalues(): | |
| for image_params in image_adlabels: | |
| ad = copy(existing_ad) | |
| ad.replace_image(image_params["image"].facebook_img_hash, image_params["adlabel"]) | |
| ad.remote_create(adset=ad.adset, batch=batch) | |
| new_ads.append(ad) | |
| batch.execute() | |
| change = "adding new images" | |
| title = "New images added for campaign" | |
| text = "%d ads created." % len(new_ads) | |
| self.check_errors_and_alert(change, title, text, new_ads) | |
| @log_campaign_action(ACTIONS.REMOVE_IMAGES) | |
| def remove_images_from_campaign(self, removed_images): | |
| """ | |
| Remove images from a campaign. We simply retrieve all the ads, | |
| all the creatives of these ads, and compare the name of the | |
| creatives to to the list of google_ids we have. | |
| :param removed_images: a list of image google_ids | |
| """ | |
| # get the image adlabels. we can't filter on the /adlabel route | |
| # so it's by far the easiest way | |
| image_adlabels = self.create_image_adlabels(removed_images) | |
| ads = self.api.account.get_ads_by_labels(params={ | |
| "ad_label_ids": [params["adlabel"].get_id() for params in image_adlabels], | |
| }, fields=[ | |
| Ad.Field.campaign_id, | |
| Ad.Field.effective_status, | |
| ]) | |
| # Restrict to our campaign and to active ads | |
| ads_to_be_paused = [ | |
| ad | |
| for ad in ads | |
| if ( | |
| ad[Ad.Field.campaign_id] == str(self.fb_campaign_id) and | |
| ad[Ad.Field.effective_status] in [Ad.EffectiveStatus.active, Ad.EffectiveStatus.campaign_paused] | |
| ) | |
| ] | |
| if ads_to_be_paused: | |
| # NB: we do not delete the adlabels because images are cross-campaigns | |
| self.pause_batch_of_ads(ads_to_be_paused, "image removed") | |
| def pause_ads_on_inactive_jobs(self, check_filters=False): | |
| """ | |
| Pause every ads that relates to an inactive | |
| job, using batch. | |
| :param check_filters: check if your jobs are in campaign filters | |
| """ | |
| ads = self.api.get_ads_for_campaign(self.fb_campaign_id) | |
| ads_to_be_paused = self.filter_ad_object_of_inactive_jobs(ads, check_filters=check_filters) | |
| if ads_to_be_paused: | |
| with log_campaign_action(ACTIONS.PAUSE_ADS, self.sjp_campaign): | |
| return self.pause_batch_of_ads(ads_to_be_paused) | |
| def pause_batch_of_ads(self, ads, reason=None): | |
| ''' | |
| Pause all the given adsets and send stats to datadog | |
| ''' | |
| params = {Ad.Field.status: Ad.Status.paused} | |
| warnings = self._update_batch_of_objects(ads, params) | |
| # FIXME: we don't do stats because this is only called manually (bundled jobs) | |
| change = "pausing ads" | |
| title = "Ad has been paused for campaign" | |
| text = "Reason: %s." % reason | |
| tags = ['reason:%s' % reason] | |
| self.check_errors_and_alert(change, title, text, tags, ads) | |
| # subroutines | |
| def filter_ad_object_of_inactive_jobs(self, objects, check_filters=False): | |
| """ | |
| Filter a list of object and return the one tied to inactive jobs. If | |
| `check_filters` is `True` this list will also include objects tied to | |
| jobs that are no longer in the campaign filter. The objects are supposed | |
| to be either `Ad` or `AdSet` instances. | |
| :param objects: a list of objects to verify | |
| :param check_filters: check if the related ad object job is in campaign filters | |
| """ | |
| stale_objects = [] | |
| if check_filters: | |
| job_ids_in_filters = self.sjp_campaign.get_active_jobs_in_filters() | |
| for obj in objects: | |
| job_id = get_ad_object_job_id(obj) | |
| if not job_id: | |
| logging.debug("[Updater] No job_id found") | |
| continue | |
| logging.debug("[Updater] Job id found %s" % job_id) | |
| job = MongoJob.objects.filter(job_id=job_id).first() | |
| if not job: | |
| logging.debug("[Updater] No job found with job_id %s" % job_id) | |
| continue | |
| if not job.is_active() or (check_filters and job.job_id not in job_ids_in_filters): | |
| logging.debug("[Updater] Pausing adset [%s] %s" % (obj.get_id(), obj[obj.Field.name])) | |
| stale_objects.append(obj) | |
| return stale_objects | |
| def _update_batch_of_objects(self, adsets, params): | |
| """ | |
| Update a batch of adsets given some parameters. | |
| :param adsets: an iterable of adsets to update | |
| :param params: the params to set on the adsets | |
| """ | |
| batch = self.api.new_batch() | |
| for adset in adsets: | |
| adset.remote_update(params=deepcopy(params), batch=batch) | |
| batch.execute() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment