soomgogather.google.googleads의 소스 코드
from google.ads.googleads.client import GoogleAdsClient
from marshmallow import Schema, ValidationError, fields, validate
[문서]class GoogleAds:
"""GoogleAds search_stream_request
GoogleAds를 사용한다는 것은 GoogleAds의 API를 사용한다는 의미이기 때문에 Class 생성시 GoogleAds서비스까지 생성한다.
query data 요청시 parameter는 다른 soomgo-gather와의 프로토콜을 동일하게 하기 위해 json 형식을 사용한다.
Google developer key를 인스턴스의 기본설정 값(GOOGLE_ADS_CONFIGURATION_FILE_PATH)이나
google developer 키파일(.yaml) 또는 dict를 사용하여 GoogleAds 서비스 객체를 생성한다.
생성한 GoogleAds 인스턴스를 사용하여 쿼리를 통해 원하는 데이터에 대한 통계 정보를 얻을 수 있다.
https://developers.google.com/google-ads/api/reference/rpc/v9/SearchGoogleAdsStreamRequest
사용 예시) dict를 사용하여 GoogleAds 서비스 인스턴스를 생성한다.
키워드 퍼포먼스의 어제 데이터의 클릭수를 받아온다.
.. code-block:: python
>>> from soomgogather.google import GoogleAds
>>> customer_id = "123445678"
>>> query = "SELECT metrics.clicks FROM keyword_view WHERE segments.date DURING YESTERDAY"
>>> params = {
... 'query': query,
... 'customer_id': customer_id,
>>> }
>>> credentials_dict = {
... 'developer_token': '<<PUT YOUR DEVELOPER TOKEN>>',
... 'refresh_token': '<<PUT YOUR REFRESH TOKEN>>',
... 'client_id': '<<PUT YOUR CLIENT ID>>',
... 'client_secret': '<<PUT YOUR CLIENT SECRET>>',
... 'use_proto_plus': True,
... 'login_customer_id': '<<PUT YOUR LOGIN CUSTOMER ID>>',
>>> }
>>> # If using file
>>> # credentials_dict = { 'key_file':'<<PUT YOUR KEY FILE(.yaml) PATH>>', }
>>> service = GoogleAds(credentials=credentials_dict)
>>> stream = service.search_stream_request(params)
>>> for batch in stream:
... for row in batch.results:
... matrics = row.matrics
... print(matrics.clicks)
"""
def __init__(self, credentials=None, version="v8"):
self.client = self._create_client(credentials, version)
self.service = self.client.get_service("GoogleAdsService")
class _GoogleAdsSchema(Schema):
query = fields.Str(required=True)
customer_id = fields.Str(required=True)
summary_row_setting = fields.Str(
validate=validate.OneOf(
['UNSPECIFIED', 'UNKNOWN', 'NO_SUMMARY_ROW', 'SUMMARY_ROW_WITH_RESULTS', 'SUMMARY_ROW_ONLY']
)
)
def _get_params(self, params):
if "summary_row_setting" in params:
try:
params["summary_row_setting"] = params["summary_row_setting"].upper()
except AttributeError as err:
raise ValueError(f"summary_row_setting is a string type.: {err}")
try:
return self._GoogleAdsSchema().load(params)
except ValidationError as err:
raise ValueError(f"incorrect parameters: {err}")
def _create_client(self, credentials, version):
if credentials is None:
_client = GoogleAdsClient.load_from_storage(version=version)
elif 'key_file' in credentials:
_client = GoogleAdsClient.load_from_storage(path=credentials.get('key_file'), version=version)
else:
_client = GoogleAdsClient.load_from_dict(credentials, version=version)
return _client
def _get_summary(self, params):
return getattr(
self.client.get_type('SummaryRowSettingEnum').SummaryRowSetting, params['summary_row_setting']
).value
[문서] def search_stream_request(self, params={}):
"""전달한 필터와 조건에 맞는 GoogleAds 정보를 받아온다.
:param params: stream request를 위한 매개변수, query, customer_id는 필수
:type params: dict
**params:**
- *query* (`str`) : 데이터를 추출할 쿼리
- *customer_id* (`str`) : GoogleAds customer ID
- *summary_row_setting* (`str, optional`) - summary row에 대한 설정
- UNSPECIFIED : 명시되지 않음.
- UNKNOWN : 반환 요약 행의 unknown 값을 표시.
- NO_SUMMARY_ROW : 요약 행을 반환하지 않음.
- SUMMARY_ROW_WITH_RESULTS : 결과와 함께 요약 행을 반환. 요약 행은 마지막 배치에서만 반환(마지막 배치에는 결과가 포함되지 않음).
- SUMMARY_ROW_ONLY : 요약 행만 반환하고 결과는 반환하지 않음.
"""
params = self._get_params(params)
search_request = self.client.get_type("SearchGoogleAdsStreamRequest")
search_request.customer_id = params.get('customer_id')
search_request.query = params.get('query')
if "summary_row_setting" in params:
search_request.summary_row_setting = self._get_summary(params=params)
stream = self.service.search_stream(search_request)
return stream