Meta 마케팅 api, 파이썬 requests를 이용하여 광고 조회하기

  이전 글에서 페이스북의 데이터들을 가져오기 위한 graph api 사용법에 대해 간단히 알아봤다. 이를 기반으로 페이스북 광고 채널의 광고들을 관리하기 위한 HTTP 기반 api를 마케팅 api라고 한다. 현재 작업 환경은 파이썬이므로 파이썬의 requests 라이브러리를 활용하여 facebook 광고 채널과 통신하면서 광고 정보들을 가져올 것이다. 여기서도 access token에 대한 내용은 생략하겠다.

요구사항

캠페인 구조

fig 1. 광고캠페인 구조

  이전에 언급했다시피, 디지털 마케팅의 채널들은 fig 1과 같은 구조를 따른다. 이는 facebook 광고 역시 마찬가지다. 그렇기 때문에 다음과 같은 json 형식으로 광고 정보를 가져오려고 한다.

####### fig 2 #######
{
    <campaign info...>
    "adsets": [
    	{
    	  <adset info...>
        	"ads": [
            	{
            	  <ad info...>
                },
            .
            .
            .

조회 방법

  1.을 제외하고 모든 반환값은 '캠페인>광고세트>광고' 구조를 가진다. 그리고 목표 레벨에 해당하는 계층은 좀 더 디테일한 정보를 반환해야 한다.

1. 광고 계정 id가 주어졌을 때 해당 계정의 모든 '캠페인>광고세트>광고' 구조를 가져오기
2. 광고 계정 id, 캠페인 id가 주어졌을 때 '--level' 옵션(adset or ad)에 따라 목표 레벨이 달라진다. 없다면 목표레벨은 campaignd이고 이때는 캠페인 내부 adset의 정보까지만 가져온다.
3. 광고 계정 id, 캠페인 id가 주어지고 '--target' 옵션(ex. adset1_id{ad1_id, ad2_id}, adset2_id{ad3_id})이 주어졌을 때 target에 해당하는 광고 객체의 정보만 가져오기
   1. target에 adset의 id만 있는 경우 목표 레벨은 adset이고 해당 adset들의 디테일한 정보를 반환해야 한다.
   2. target에 adset, ad의 id가 모두 있는 경우 지정된 adset, ad의 정보들만 가져오는데, 목표레벨은 ad이므로 ad 계층은 좀 더 디테일한 정보를 반환해야한다.
   3. '--level', '--target' 옵션이 모두 있을 때는 목표 레벨이 더 하위 객체인 곳을 우선한다. 만약 두 옵션의 수준이 같다면 '--target'을 우선한다.

사용 제한

  graph api 요청은 무한정 보낼 수는 없다. 특정한 기준에 따라 사용치를 넘어가면 속도/사용 제한이 걸리기 때문이다. 명확한 기준을 찾을 수는 없지만 api 요청 횟수나 cpu 사용량에 따라 결정되는 것 같다. 그렇기 때문에 요청을 최대한 효율적으로 보낼 필요가 있다.

  사실 코드도 필요할 때마다 그때그때 요청을 보내는게 코드 구조는 더  단순할 것이다. 하지만 요청을 주고받는 오버헤드를 줄이고 사용제한을 최대한 지연시켜야 하기 때문에 사용제한은 무조건 고려해야할 부분이다.

 

환경

  우선 코드의 동작을 확인하기 위해 임시로 meta developer에서 나의 앱을 만들고 이 앱에 마케팅 api를 사용할 것이다. 광고도 캠페인 2개, 캠페인 당 광고 세트 2개, 광고세트 당 광고 2개를 미리 만들어 놓았다. 또한 api 요청은 requests 라이브러리를 사용할 것이다. 현재 프로젝트의 실제 동작과정 이러하다.

1. '커맨드 명령어 입력. 광고 계정id, 캠페인 id, --target, --level 등의 옵션이 포함된다.
2. 커맨드에 따라 목표 레벨이 설정되고 이에 따라 실행할 함수가 매핑된다(get_account, get_campaign, get_adset, get_ad)
3. 함수가 실행되면서 meta api를 요청하면서 필요한 광고 객체 정보를 가져오고 가공한다.
4. 가져온 광고 객체를 만들어 놓은 모델에 넣어준다

  여기서는 3.과정에 대해서만 알아보자.

 

코드

  어떤 방식으로 데이터를 가져올까 정말 많은 고민을 했다. 각 함수에 따라 그때그때 필요한 광고 객체 정보만 가져오려고 했는데 이렇게 해버리면 api 호출을 너무 자주 해버린다. 예를 들어 조회 방법의 1. 상황일 때, 계정 조회, 캠페인 조회, 캠페인 내부의 광고세트, 광고세트 내부의 광고들을 조회해야하는데 엣지 요청으로는 한 상위 객체에 대한 하위 객체들을 요청하는 것이 최선이므로 요청을 빈번하게 해야했고 코드도 복잡해졌다.

  이 때문에 자연스럽게 한번에 모든 계층 정보를 자져올 수 있는 중접 요청을 생각하게 되었고 이에 따라 자연스럽게 상황에 따라 필요한 모든 데이터를 먼저 가져온 다음 필요한 것만 필터링하는 방법을 생각하게 되었다. 사실 이 방법도 상황에 따라 정답이 아닐 수 있지만 이게 지금 내가 생각할 수 있는 최선이었다.

  해당 프로젝트는 팀장님이 만드신 모듈을 이용해서 구현하고 있기 때문에 일반 requests 라이브러리와 모양이 다르거나 잘 파악이 되지 않는 타입, 변수, 메서드 등이 있을 수 있다. 하지만 구조를 이해하는데는 큰 문제가 없을 것이다.

각 상황에 맞춰 필요한 데이터 가져오기

  위와 같이 생각했기 때문에 '상황에 따라 필요한 모든 데이터를 먼저 가져오기'를 먼저 해줘야 했다. 이 기능을 하는 'get_required_data()'라는 메서드를 정의해줬다

def get_required_data(conn:Session, access_token:str, account_id:str, caller_name:str, campaign_id:str=None):

    if caller_name == 'get_account':

        conn.get(account_id, access_token = access_token, fields = Fields.GET_ACCOUNT)

        res = get_all_child_of_adobject(conn.response_json(), ObjectLevel.CAMPAIGN)

        all_campaigns = list()

        for campaign in res['campaigns']:

            campaigns_with_adgroups = get_all_child_of_adobject(campaign, ObjectLevel.AD_GROUP)

            all_adgroups = [get_all_child_of_adobject(i, ObjectLevel.AD)

                            for i in campaigns_with_adgroups['adsets']]

            campaigns_with_adgroups['adsets'] = all_adgroups

            all_campaigns.append(campaigns_with_adgroups)

        res['campaigns'] = all_campaigns

    elif caller_name == 'get_campaign':

        conn.get(campaign_id, access_token = access_token, fields = Fields.GET_CAMPAIGN)

        res = get_all_child_of_adobject(conn.response_json(), ObjectLevel.AD_GROUP)

   

    elif caller_name == 'get_adgroup':

        conn.get(campaign_id, access_token = access_token, fields = Fields.GET_ADGROUP)

        res = get_all_child_of_adobject(conn.response_json(), ObjectLevel.AD_GROUP)

        all_adgroups = [get_all_child_of_adobject(i, ObjectLevel.AD) for i in res['adsets']]

        res['adsets'] = all_adgroups

   

    elif caller_name == 'get_ad':

        conn.get(campaign_id, access_token = access_token, fields = Fields.GET_AD)

        res = get_all_child_of_adobject(conn.response_json(), ObjectLevel.AD_GROUP)

        all_adgroups = [get_all_child_of_adobject(i, ObjectLevel.AD) for i in res['adsets']]

        res['adsets'] = all_adgroups

    return res

class Fields:
    ACCOUNT_INFO: Final = "id,name"
    GET_ACCOUNT: Final = 'id,name,campaigns{id,name,status,adsets{id,name,status,ads{id,name,status}}}'
    GET_CAMPAIGN: Final = 'id,name,status,adsets{id,name,status}'
    GET_ADGROUP: Final = "id,name,status,adsets{id,name,status,ads{id,name,status}}"
    GET_AD: Final = "id,name,status,adsets{id,name,status,ads{id,name,status}}"
    INSIGHT_DATA: Final = "clicks,conversions,impressions"

  원래 목표 레벨의 필드는 더 디테일한 필드값들이 들어가야하지만 여기서는 편의를 위해 모두 'id, name, status'만을 조회하는 것으로 통일하자. 인자로 caller_name이라는 값을 받는 것을 확인할 수 있다. caller_name은 위에서 언급한 목표레벨에 따라 매핑되는 메서드의 이름이다(get_account, get_campaign, ... ). 

  모든 분기가 동일하게 처음엔 conn.get(requests.get과 거의 유사)을 사용하여 원하는 정보를 가져온다. 이때의 응답값들은 conn.response.json에 담기게 된다. 두번째 코드를 보면 각 분기에 어떤 필드값이 삽입되는지 알 수 있다. caller_name이 get_campaign이라면 아마 이런 형식으로 반환값이 올 것이다.

{
  "id": "23852696963720226",
  "name": "test campaign 1",
  "status": "ACTIVE",
  "adsets": {
    "data": [
      {
        "id": "23852697435960226",
        "name": "광고 세트 1-3",
        "status": "PAUSED"
      }
      ...
    ],
    "paging": {
      "cursors": {
        "before": "QVFIUjlhQ056Y2VvWllwNkFMVEVVfTHNya2xDRGFtVWx4cThCSVFtYXRJN0VDdGJja3hoS2FFQmxEa1BpYzFB",
        "after": "QVFIUjlhQ056Y2VvWllwNkFMVEVWWXRtYXVfTHNya2xDRGFtVWx4cThCSVFtYXRJN0VDdGJja3hoS2FFQmxEa1BpYzFB"
      },
      "next": "https://graph.facebook.com/v15.0/23852696963720226/adsets?access_token={token_value}&pretty=0&fields=id%2Cname%2Cstatus&limit=1&after=QVFIUjlhQ056Y2VvWllwNkFMVEVWWXRqcVpGbGNwTmNzYBpYzFB"
    }
  }
}

  처음에는 "중첩 요청을 사용하면 원하는 전 계층에 원하는 필드값을 각각 설정해서 가져올 수 있어서 좋네 끝!"이라고 생각했는데 하나 간과한 것이 있었다. 바로 pagination을 고려해야 한다는 것이다. 지금 임의로 만든 광고들은 그 수가 적어서 pagination을 사용하지 않아도 되니 미처 고려하지 않았던 것이다. 그렇기 때문에 pagination을 고려한 특정 광고 객체의 자식 객체들을 모두 가져오는 기능이 필요하다고 생각했고 그렇게 만든 메소드가 'get_all_child_of_adobject()'이다.

{
    특정 객체 정보
    ...
    "하위 객체들": {
    	"data": [...]
        "paging": {
        	"cursors": {}
            "next": ...
    	} 
    }     
}

# 다음과 같이 변경
{
    특정 객체 정보
    ...
    "하위 객체들": [...]
    }     
}

  즉 위와 같이 "하위 객체들"의 value 값으로 바로 모든 하위 객체들의 정보가 리스트 형으로 들어가게끔 하는 것이다. 코드를 살펴보자.

def get_all_child_of_adobject(struct:dict, child_level: str):

    try: adobjects = struct[ResKeys.CAMPAIGNS] if child_level==ObjectLevel.CAMPAIGN \

        else struct[ResKeys.ADSETS] if child_level==ObjectLevel.AD_GROUP \

            else struct[ResKeys.ADS]

    except KeyError as e:

        object_name = struct[ObjectAttr.NAME]

        object_id = struct[ObjectAttr.ID]

        print(f'{object_name}({object_id})의 하위 객체가 존재하지 않습니다.')

        return

   

    all_childs = adobjects[ResKeys.DATA]

    all_childs = get_all_pagination_result(adobjects, all_childs)

   

    if child_level == ObjectLevel.CAMPAIGN:

        struct[ResKeys.CAMPAIGNS] = all_childs

    elif child_level == ObjectLevel.AD_GROUP:

        struct[ResKeys.ADSETS] = all_childs

    else:

        struct[ResKeys.ADS] = all_childs

    return struct

  우선 adobjets에 '하위 객체들'의 value 값을 넣어 준다. 그리고 all_childs에 첫 페이지의 객체 정보들을 넣어준 뒤에 모든 페이지를 순회하며 모든 자식 객체를 반환하는 get_all_pagination_result() 메서드를 통해 모든 자식 객체를 다시 all_childs에 가져온다. 그리고  '하위 객체들'의 value 값을 all_childs로 갈아 끼워준다. 

def get_all_pagination_result(response, results):

    while True:

        if 'paging' in response and 'next' in response['paging']:

            next_page = response['paging']['next']

            response = json.loads(requests.get(next_page).text)

            results.extend(response[ResKeys.DATA])

        else:

            break

   

    return results

  이 메서드는 응답값의 paging 속성 안에 'next'라는 키값이 있는지 확인하고 없을 때까지 'next'의 url로 요청을 보내고 이를 dict 형식으로 변환한 뒤 results에 붙여넣은 뒤 이를 반환해준다.

  이런 과정을 통해 'get_required_data()'가 종료되면 fig 2와 같은 형식의 dict 형식 데이터가 반환될 것이다. 하지만 여기서 필요한 것만 필터링하고 다음 단계로 넘어가야하는 경우가 있는데, 이때 사용되는 메서드가 '_filter_target()'이다.

def _filter_target(self, comm:Commands, info:dict, caller_name:str):

        if caller_name == 'get_adgroup':

            target_adgroups = list(filter(lambda x: x['id'] in comm.adgroups, info[ResKeys.ADSETS]))

            info[ResKeys.ADSETS] = target_adgroups

        elif caller_name == 'get_ad':

            target = list()

            target_adgroups = list(filter(lambda x: x['id'] in comm.target.keys(), info[ResKeys.ADSETS]))

            for adgroup in target_adgroups:

                ads_id = comm.ads(adgroup['id'])

                if ads_id:

                    target_ads = list(filter(lambda x: x['id'] in ads_id, adgroup[ResKeys.ADS]))

                    adgroup[ResKeys.ADS] = target_ads

                target.append(adgroup)

            info[ResKeys.ADSETS] = target

       

        return info

  해당 함수가 필요한 경우는 get_adgorup, get_ad가 매핑되었을 때이다. get_adgroup()인 경우에는 전체 adset의 id 중 '--target' 옵션으로 받은 adset의 id에 해당하는 adset만 필터링 한 후 교체해준다. get_ad()인 경우에는 target adset만 필터링 한 후 이 adset들을 순회하면서 --target 옵션에 있던 ad_id에 해당하는 ad들만 필터링하여 adset내부 ads value를 교체해준다. 

  정리하고보니 전체코드가 따로 없어서 처음 보는 사람은 다소 혼란스러울 수도 있을 것 같다. 하지만 중요한건 meta graph api가 어떻게 동작하는지 이해한 다음 응답값들을 적절히 조작하여 원하는 값만을 얻을 수 있다는 것이다. 정보가 부족했기에 처음으로 공식문서를 제대로 읽어보기도 했고, 동일한 기능을 하여도 요구사항이나 구조가 어떻게 변하냐에 따라 코드를 여러번 수정하는 경험도 해보았다. 아직도 리팩토링의 여지가 다분해보이긴 하지만 파이썬이라는 언어에 익숙해지고 외부 api를 깊게 이해해보는 의미있는 프로젝트였다.