overwrite_aac

sync_amz_data/public/amz_ad_client.py

@@ -1,4 +1,6 @@
 import requests
+from urllib3.util.retry import Retry
+from requests.adapters import HTTPAdapter
 import json
 import time
 from cachetools import TTLCache
@@ -6,14 +8,20 @@ from urllib.parse import urljoin
 from typing import List, Literal, Iterable, Iterator
 import gzip
 from pathlib import Path
-from logging import getLogger
+
+import logging
 
 URL_AUTH = "https://api.amazon.com/auth/o2/token"
 URL_AD_API = "https://advertising-api.amazon.com"
 
 cache = TTLCache(maxsize=10, ttl=3200)
 
-logger = getLogger(__name__)
+logger = logging.getLogger(__name__)
+
+
+class RateLimitError(Exception):
+    def __init__(self, retry_after: str = None):
+        self.retry_after = retry_after
 
 
 def gz_decompress(file_path: str, chunk_size: int = 1024 * 1024):
@@ -41,6 +49,19 @@ class BaseClient:
         if not self.data_path.exists():
             self.data_path.mkdir(parents=True)
 
+        retry_strategy = Retry(
+            total=5,  # 重试次数
+            allowed_methods=["GET", "POST"],
+            # 强制重试的状态码，在method_whitelist中的请求方法才会重试
+            status_forcelist=[429, 500, 502, 503, 504],
+            raise_on_status=False,  # 在status_forcelist中的状态码达到重试次数后是否抛出异常
+            # backoff_factor * (2 ** (retry_time-1))， 即间隔1s, 2s, 4s, 8s, ...
+            backoff_factor=1,
+        )
+        adapter = HTTPAdapter(max_retries=retry_strategy)
+        self.session = requests.session()
+        self.session.mount("https://", adapter)
+
     @property
     def access_token(self) -> str:
         try:
@@ -72,15 +93,19 @@ class BaseClient:
         head = self.auth_headers
         if headers:
             head.update(headers)
-        resp = requests.request(
+        resp = self.session.request(
             method=method,
             url=urljoin(URL_AD_API, url_path),
             headers=head,
             params=params,
             json=body,
         )
-        js = resp.json()
-        return js
+        if resp.status_code == 429:
+            raise RateLimitError(resp.headers.get("Retry-After"))
+        if resp.status_code >= 400:
+            raise Exception(resp.text)
+
+        return resp.json()
 
 
 class SPClient(BaseClient):
@@ -149,6 +174,7 @@ class SPClient(BaseClient):
             "Content-Type": "application/vnd.spKeyword.v3+json"
         }
         return self._request(url_path, method="POST", body=body, headers=headers)
+
     def iter_keywords(self, **body) -> Iterator[dict]:
         if "maxResults" not in body:
             body["maxResults"] = 100
@@ -186,26 +212,30 @@ class SPClient(BaseClient):
         }
         return self._request(url_path, method="POST", body=body)
 
-    def get_adgroup_bidrecommendation(self,campaignId:str,adGroupId:str,targetingExpressions:list,recommendationType:str="BIDS_FOR_EXISTING_AD_GROUP"):
+    def get_adgroup_bidrecommendation(
+            self, campaignId: str, adGroupId: str, targetingExpressions: list,
+            recommendationType: str = "BIDS_FOR_EXISTING_AD_GROUP"):
         url_path = "/sp/targets/bid/recommendations"
         headers = {
             "Accept": "application/vnd.spthemebasedbidrecommendation.v3+json",
             "Content-Type": "application/vnd.spthemebasedbidrecommendation.v3+json"
         }
-        body = {"campaignId":campaignId,
-                "adGroupId":adGroupId,
-                "recommendationType":recommendationType,
-                "targetingExpressions":targetingExpressions}
+        body = {
+            "campaignId": campaignId,
+            "adGroupId": adGroupId,
+            "recommendationType": recommendationType,
+            "targetingExpressions": targetingExpressions
+        }
         return self._request(url_path, method="POST", body=body, headers=headers)
 
-    def get_keyword_bidrecommendation(self,adGroupId:str,keyword:list,matchType:list):
-        keywords = list(map(lambda x:{"keyword":x[0],"matchType":x[1]},list(zip(keyword,matchType))))
-        print(keywords)
+    def get_keyword_bidrecommendation(self, adGroupId: str, keyword: list, matchType: list):
+        keywords = list(map(lambda x: {"keyword": x[0], "matchType": x[1]}, list(zip(keyword, matchType))))
         url_path = "/v2/sp/keywords/bidRecommendations"
-        body = {"adGroupId":adGroupId,
-                "keywords":keywords}
+        body = {"adGroupId": adGroupId,
+                "keywords": keywords}
         return self._request(url_path, method="POST", body=body)
 
+
 class SBClient(BaseClient):
     def get_campaigns(self, **body):
         url_path = "/sb/v4/campaigns/list"
@@ -226,13 +256,13 @@ class SBClient(BaseClient):
             body["nextToken"] = info["nextToken"]
         # logger.info(f"总共数量：{info['totalResults']}")
 
-    def get_ad_groups(self,**body):
+    def get_ad_groups(self, **body):
         url_path = "/sb/v4/adGroups/list"
         headers = {
             'Content-Type': "application/vnd.sbadgroupresource.v4+json",
             'Accept': "application/vnd.sbadgroupresource.v4+json"
         }
-        return self._request(url_path, method="POST", headers=headers,body=body)
+        return self._request(url_path, method="POST", headers=headers, body=body)
 
     def iter_adGroups(self, **body) -> Iterator[dict]:
         if "maxResults" not in body:
@@ -245,44 +275,34 @@ class SBClient(BaseClient):
                 break
             body["nextToken"] = info["nextToken"]
 
-    def get_ads(self,**body):
+    def get_ads(self, **body):
         url_path = "/sb/v4/ads/list"
-        headers = {'Content-Type': "application/vnd.sbadresource.v4+json",
-                  'Accept': "application/vnd.sbadresource.v4+json"
-                  }
-        return self._request(url_path, method="POST", headers=headers,body=body)
+        headers = {
+            'Content-Type': "application/vnd.sbadresource.v4+json",
+            'Accept': "application/vnd.sbadresource.v4+json"
+        }
+        return self._request(url_path, method="POST", headers=headers, body=body)
 
-    def iter_ads(self,**body):
+    def iter_ads(self, **body):
         if "maxResults" not in body:
             body["maxResults"] = 100
         while True:
             info: dict = self.get_ads(**body)
-            # print(info)
+            print(info)
             yield from info["ads"]
             if not info.get("nextToken"):
                 break
             body["nextToken"] = info["nextToken"]
-    def get_keywords(self,**param):
-        url_path = "/sb/keywords"
-        return self._request(url_path, method="GET",params=param)
 
-    def iter_keywords(self,**param):
-        if "startIndex" not in param:
-            param["startIndex"] = 0
-            param["count"] = 5000
-        while True:
-            info:list = self.get_keywords(**param)
-            # print(info)
-            if len(info)==0:
-                break
-            param["startIndex"] += 5000
-            yield info
+    def get_keywords(self):
+        url_path = "/sb/keywords"
+        return self._request(url_path, method="GET")
 
-    def get_targets(self,**body):
+    def get_targets(self, **body):
         url_path = "/sb/targets/list"
         return self._request(url_path, method="POST", body=body)
 
-    def iter_targets(self,**body):
+    def iter_targets(self, **body):
         if "maxResults" not in body:
             body["maxResults"] = 100
         while True:
@@ -293,12 +313,12 @@ class SBClient(BaseClient):
                 break
             body["nextToken"] = info["nextToken"]
 
-    def get_budget(self,campaignIds:list):
+    def get_budget(self, campaignIds: list):
         url_path = "/sb/campaigns/budget/usage"
-        body = {"campaignIds":campaignIds}
-        return self._request(url_path, method="POST",body=body)
+        body = {"campaignIds": campaignIds}
+        return self._request(url_path, method="POST", body=body)
 
-    def get_keyword_bidrecommendation(self,**body):
+    def get_keyword_bidrecommendation(self, **body):
         url_path = "/sb/recommendations/bids"
         return self._request(url_path, method="POST", body=body)
 
@@ -353,7 +373,7 @@ class SBClient(BaseClient):
 
     def download_report(self, report_id: str, file_path: str, decompress: bool = True) -> str:
         url = urljoin(URL_AD_API, f"/v2/reports/{report_id}/download")
-        resp = requests.get(url, headers=self.auth_headers, stream=True)
+        resp = requests.get(url, headers=self.auth_headers, stream=True, allow_redirects=True)
         logger.info(f"开始下载报告：{report_id}")
         with open(file_path, "wb") as file:
             for data in resp.iter_content(chunk_size=10 * 1024):
@@ -366,16 +386,24 @@ class SBClient(BaseClient):
         return de_file
 
 
+
+
 class SDClient(BaseClient):
     def get_campaigns(self, **params) -> List[dict]:
         url_path = "/sd/campaigns"
         return self._request(url_path, params=params)
 
+
 class Account(BaseClient):
-    def get_portfolio(self):
+    def get_portfolios(self):
         url_path = "/v2/portfolios/extended"
         return self._request(url_path)
 
+    def iter_portfolios(self):
+        yield from self.get_portfolios()
+
+
+AccountClient = Account
 
 if __name__ == '__main__':
     AWS_CREDENTIALS = {
@@ -385,56 +413,49 @@ if __name__ == '__main__':
         'profile_id': "3006125408623189"
     }
     # sp = SPClient(**AWS_CREDENTIALS)
-    # print(sp.get_keyword_bidrecommendation(adGroupId="119753215871672",keyword=["8mp security camera system","8mp security camera system"],matchType=["broad","exact"]))
+    # print(sp.get_keyword_bidrecommendation(
+    # adGroupId="119753215871672",
+    # keyword=["8mp security camera system","8mp security camera system"],
+    # matchType=["broad","exact"]))
     sb = SBClient(**AWS_CREDENTIALS)
     # print(list(sb.iter_targets()))
-    # print(sb.get_keyword_bidrecommendation(**{'campaignId': 27333596383941,'keywords':[{"matchType":'broad',"keywordText":"4k security camera system"}]}))
-    import pandas as pd
-
-    pd.set_option('display.max_columns', None)
-    import warnings
-
-    warnings.filterwarnings('ignore')
-    pd.set_option('expand_frame_repr', False)
-    # a = sb.iter_keywords()
-    a = [row for _ in list(sb.iter_keywords()) for row in _]
-    print(len(a))
-
-    # print(pd.json_normalize(a))
+    print(sb.get_keyword_bidrecommendation(**{'campaignId': 27333596383941, 'keywords': [
+        {"matchType": 'broad', "keywordText": "4k security camera system"}]}))
+    print(sb.get_budget([27333596383941]))
     # sd = SDClient(**AWS_CREDENTIALS)
     # print(sd.get_campaigns(startIndex=10, count=10))
 
-    # sb = SBClient(**AWS_CREDENTIALS)
-    # metrics = [
-    #     'applicableBudgetRuleId',
-    #     'applicableBudgetRuleName',
-    #     'attributedConversions14d',
-    #     'attributedConversions14dSameSKU',
-    #     'attributedDetailPageViewsClicks14d',
-    #     'attributedOrderRateNewToBrand14d',
-    #     'attributedOrdersNewToBrand14d',
-    #     'attributedOrdersNewToBrandPercentage14d',
-    #     'attributedSales14d',
-    #     'attributedSales14dSameSKU',
-    #     'attributedSalesNewToBrand14d',
-    #     'attributedSalesNewToBrandPercentage14d',
-    #     'attributedUnitsOrderedNewToBrand14d',
-    #     'attributedUnitsOrderedNewToBrandPercentage14d',
-    #     'campaignBudget',
-    #     'campaignBudgetType',
-    #     'campaignId',
-    #     'campaignName',
-    #     'campaignRuleBasedBudget',
-    #     'campaignStatus',
-    #     'clicks',
-    #     'cost',
-    #     'dpv14d',
-    #     'impressions',
-    #     'unitsSold14d',
-    #     'attributedBrandedSearches14d',
-    #     'topOfSearchImpressionShare']
+    sb = SBClient(**AWS_CREDENTIALS)
+    metrics = [
+        'applicableBudgetRuleId',
+        'applicableBudgetRuleName',
+        'attributedConversions14d',
+        'attributedConversions14dSameSKU',
+        'attributedDetailPageViewsClicks14d',
+        'attributedOrderRateNewToBrand14d',
+        'attributedOrdersNewToBrand14d',
+        'attributedOrdersNewToBrandPercentage14d',
+        'attributedSales14d',
+        'attributedSales14dSameSKU',
+        'attributedSalesNewToBrand14d',
+        'attributedSalesNewToBrandPercentage14d',
+        'attributedUnitsOrderedNewToBrand14d',
+        'attributedUnitsOrderedNewToBrandPercentage14d',
+        'campaignBudget',
+        'campaignBudgetType',
+        'campaignId',
+        'campaignName',
+        'campaignRuleBasedBudget',
+        'campaignStatus',
+        'clicks',
+        'cost',
+        'dpv14d',
+        'impressions',
+        'unitsSold14d',
+        'attributedBrandedSearches14d',
+        'topOfSearchImpressionShare']
     # sb.get_report(
     #     record_type="campaigns",
     #     report_date="20231008",
     #     metrics=metrics
-    # )
+    # )