如何避免Facebook Marketing API调用超限并优化请求次数

原代码的核心问题 #

1. 单广告单次Insights请求：遍历每个广告单独调用/insights接口，调用次数随广告数量直接翻倍
2. 重复查询广告系列目标：同一广告系列被多个广告关联时，重复发起campaign查询请求
3. 未利用嵌套字段查询：拆分多次请求获取广告ID、广告详情，没有通过Graph API的嵌套字段一次性拉取关联数据
4. 分页逻辑错误：不同接口复用同一个cursor，导致分页处理混乱

优化措施 #

1. 批量获取Insights数据：直接通过广告账户批量请求所有目标广告的曝光数据，一次请求覆盖多个广告
2. 缓存广告系列目标：用字典缓存已查询过的campaign objective，避免重复调用
3. 嵌套字段一次性拉取：通过ad_sets的嵌套字段直接获取广告的ID、名称、关联campaign信息，减少请求次数
4. 独立处理分页：每个接口的分页cursor单独管理，确保分页逻辑正确

优化后的完整代码 #

import requests
import datetime

ad_account_id = 'act_11111'
access_token = 'ABC'
ad_set_ids = ['one', 'two']

# 定义需要的字段：广告ID、名称、关联campaign的ID和objective
ad_fields = 'id,name,campaign{id,objective}'
# 时间范围参数
time_range = {
    "since": datetime.datetime(2023, 1, 1).date().isoformat(),
    "until": datetime.datetime(2023, 4, 30).date().isoformat()
}

# 缓存已查询的campaign目标，避免重复调用
campaign_objective_cache = {}

def get_all_ads_from_ad_sets():
    """批量获取指定广告组下的所有广告及关联campaign信息"""
    all_ads = []
    has_next_page = True
    cursor = ''
    
    while has_next_page:
        # 一次性拉取广告组下的广告，以及广告关联的campaign字段
        url = f'https://graph.facebook.com/v16.0/?ids={",".join(ad_set_ids)}'
        params = {
            'fields': f'ads{{{ad_fields}}}',
            'access_token': access_token,
            'after': cursor
        }
        response = requests.get(url, params=params).json()
        
        for ad_set_id in response.keys():
            if 'ads' in response[ad_set_id]:
                ad_data = response[ad_set_id]['ads']['data']
                all_ads.extend(ad_data)
                
                # 更新分页cursor
                if 'paging' in response[ad_set_id]['ads'] and 'cursors' in response[ad_set_id]['ads']['paging']:
                    cursor = response[ad_set_id]['ads']['paging']['cursors'].get('after', '')
                else:
                    has_next_page = False
    return all_ads

def get_batch_insights(ad_ids):
    """批量获取多个广告的每日曝光数据"""
    if not ad_ids:
        return {}
    
    url = f'https://graph.facebook.com/v16.0/{ad_account_id}/insights'
    params = {
        'level': 'ad',
        'fields': 'ad_id,time_range,impressions',
        'time_range': str(time_range).replace("'", "\""),
        'filtering': f'[{{"field":"ad.id","operator":"IN","value":["{"\",\"".join(ad_ids)}"]}}]',
        'access_token': access_token,
        'breakdowns': 'day'  # 按天拆分数据
    }
    response = requests.get(url, params=params).json()
    
    # 整理成ad_id为key的字典，方便匹配
    insights_map = {}
    if 'data' in response:
        for item in response['data']:
            ad_id = item['ad_id']
            if ad_id not in insights_map:
                insights_map[ad_id] = []
            insights_map[ad_id].append({
                'day': item['time_range']['date_start'],
                'impressions': item['impressions']
            })
    return insights_map

# 主流程
if __name__ == '__main__':
    # 1. 获取所有广告及关联campaign信息
    ads_list = get_all_ads_from_ad_sets()
    
    # 2. 提取所有广告ID，批量获取insights
    ad_ids = [ad['id'] for ad in ads_list]
    insights_data = get_batch_insights(ad_ids)
    
    # 3. 整合数据并输出
    for ad in ads_list:
        ad_id = ad['id']
        ad_name = ad['name']
        # 从嵌套字段获取campaign信息，或缓存中读取
        campaign = ad.get('campaign', {})
        campaign_id = campaign.get('id')
        objective = campaign.get('objective')
        
        # 如果嵌套字段没拿到，查缓存或请求一次
        if not objective and campaign_id:
            if campaign_id in campaign_objective_cache:
                objective = campaign_objective_cache[campaign_id]
            else:
                # 仅当缓存中没有时才请求
                campaign_url = f'https://graph.facebook.com/v16.0/{campaign_id}?fields=objective&access_token={access_token}'
                campaign_res = requests.get(campaign_url).json()
                objective = campaign_res.get('objective', 'Unknown')
                campaign_objective_cache[campaign_id] = objective
        
        # 匹配曝光数据
        daily_impressions = insights_data.get(ad_id, [])
        for item in daily_impressions:
            print(f"Ad ID: {ad_id}")
            print(f"Ad Name: {ad_name}")
            print(f"Objective: {objective}")
            print(f"Day: {item['day']}")
            print(f"Impressions: {item['impressions']}")
            print("-" * 30)

优化效果说明 #

• API调用次数从O(N+M)（N为广告数，M为重复campaign数）降到O(1+K)（K为分页请求数，通常远小于广告数）
• 利用嵌套字段直接拉取广告关联的campaign信息，避免额外请求
• 批量Insights请求一次性获取所有广告的每日数据，彻底消除单广告请求的开销
• 缓存机制避免重复查询相同campaign的目标

内容的提问来源于stack exchange，提问作者Lax Mandis