如何使用facebook_business Python SDK实现批量请求规避限流

原代码在广告数量较多的账号触发限流，核心原因是遍历每个广告单独请求线索，导致API调用次数暴增，超出Facebook的速率限制。以下是优化后的实现，通过账号级批量获取线索+限流重试机制彻底解决这个问题：

import pandas as pd
from facebook_business.api import FacebookAdsApi
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.lead import Lead
import time
from requests.exceptions import HTTPError

# 配置信息
access_token = "my_token"
account_id = "my_ad_account_id"
start_date = '2023-07-01' 
end_date = '2023-07-31' 

# 初始化API
FacebookAdsApi.init(access_token=access_token)

def fetch_leads_with_retry(ad_account, fields, params, max_retries=3, delay=5):
    """带指数退避重试的线索获取函数，处理429限流错误"""
    for attempt in range(max_retries):
        try:
            return list(ad_account.get_leads(fields=fields, params=params))
        except HTTPError as e:
            if e.response.status_code == 429:
                print(f"触发限流，{delay}秒后重试（第{attempt+1}/{max_retries}次）")
                time.sleep(delay)
                delay *= 2  # 指数退避，每次等待时间翻倍
            else:
                raise
    raise Exception("重试次数耗尽，无法获取线索数据")

# 定义需要的字段和时间筛选条件
lead_fields = ['campaign_name', "created_time", "field_data"]
lead_params = {
    'filtering': [
        {'field': 'time_created', 'operator': 'GREATER_THAN', 'value': start_date},
        {'field': 'time_created', 'operator': 'LESS_THAN', 'value': end_date}
    ],
    'limit': 1000  # 单次请求最大返回数量，可根据需求调整
}

# 直接从广告账号批量获取所有符合条件的线索（无需遍历单个广告）
ad_account = AdAccount(account_id)
leads = fetch_leads_with_retry(ad_account, lead_fields, lead_params)

# 整理线索数据为字典格式
leads_list = []
for lead in leads:
    lead_dic = {
        "campaign_name": lead["campaign_name"],
        "created_time": lead["created_time"]
    }
    # 解析自定义字段数据，处理空值情况
    for field in lead["field_data"]:
        lead_dic[field["name"]] = field["values"][0] if field["values"] else None
    leads_list.append(lead_dic)

print(leads_list)

关键优化点说明 #

1. 减少API调用次数：
   原代码遍历每个广告单独调用get_leads()，广告数量越多请求次数越多。优化后直接通过AdAccount.get_leads()一次性获取账号下所有符合时间条件的线索，将请求次数从N次压缩到1~几次（分页场景），从根源上降低限流风险。

2. 限流自动重试：
   加入指数退避重试逻辑，遇到429限流错误时自动等待并重试，无需人工干预，提升代码稳定性。

3. 分页支持（可选）：
   如果线索数量超过limit设置的1000条，需要分页获取所有数据，可添加以下分页函数：

   def fetch_all_leads(ad_account, fields, params):
       leads = []
       after = None
       while True:
           params['after'] = after
           batch = fetch_leads_with_retry(ad_account, fields, params)
           if not batch:
               break
           leads.extend(batch)
           # 获取下一页游标
           paging = batch[-1].get('paging', {})
           after = paging.get('cursors', {}).get('after')
           if not after:
               break
           time.sleep(1)  # 分页请求间加入短暂延迟，避免触发限流
       return leads
   
   # 使用分页函数替代原获取逻辑
   leads = fetch_all_leads(ad_account, lead_fields, lead_params)
   

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