def get_schedule(self, campaign_id):
# Get the GoogleAdsService client
ga_service = self.client.get_service("GoogleAdsService")
# GAQL query to retrieve ad schedules for a specific campaign
query = f"""
SELECT
campaign_criterion.criterion_id,
campaign_criterion.ad_schedule.day_of_week,
campaign_criterion.ad_schedule.start_hour,
campaign_criterion.ad_schedule.start_minute,
campaign_criterion.ad_schedule.end_hour,
campaign_criterion.ad_schedule.end_minute
FROM campaign_criterion
WHERE
campaign_criterion.type = 'AD_SCHEDULE'
"""
# Execute the search request
try:
response = ga_service.search(customer_id=self.customer_id, query=query)
# Check if the response contains rows
if response.total_results_count == 0:
logger.info(f"No ad schedules found for campaign ID {campaign_id}.")
return []
# Process the rows
schedules = []
for row in response:
ad_schedule = row.campaign_criterion.ad_schedule
schedule_data = {
"criterion_id": row.campaign_criterion.criterion_id,
"start_time": f"{ad_schedule.start_hour}:{ad_schedule.start_minute:02d}",
"end_time": f"{ad_schedule.end_hour}:{ad_schedule.end_minute:02d}",
}
schedules.append(schedule_data)
logger.info(f"Ad Schedule: {schedule_data}")
return schedules
except Exception as e:
logger.error(f"Failed to retrieve ad schedules: {e}")
return []