create_first_commit

sync_amz_data/public/amz_ad_client_01.py

@@ -0,0 +1,440 @@
+import requests
+import json
+import time
+from cachetools import TTLCache
+from urllib.parse import urljoin
+from typing import List, Literal, Iterable, Iterator
+import gzip
+from pathlib import Path
+from logging import getLogger
+
+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__)
+
+
+def gz_decompress(file_path: str, chunk_size: int = 1024 * 1024):
+    decompressed_file = file_path.rstrip(".gz")
+    with open(decompressed_file, "wb") as pw:
+        zf = gzip.open(file_path, mode='rb')
+        while True:
+            chunk = zf.read(size=chunk_size)
+            if not chunk:
+                break
+            pw.write(chunk)
+    return decompressed_file
+
+
+class BaseClient:
+    def __init__(
+            self, lwa_client_id: str, lwa_client_secret: str, refresh_token: str = None, profile_id: str = None,
+            data_path: str = "./"
+    ):
+        self.lwa_client_id = lwa_client_id
+        self.lwa_client_secret = lwa_client_secret
+        self.refresh_token = refresh_token
+        self.profile_id = profile_id
+        self.data_path = Path(data_path)
+        if not self.data_path.exists():
+            self.data_path.mkdir(parents=True)
+
+    @property
+    def access_token(self) -> str:
+        try:
+            return cache[self.refresh_token]
+        except KeyError:
+            resp = requests.post(URL_AUTH, data={
+                "grant_type": "refresh_token",
+                "client_id": self.lwa_client_id,
+                "refresh_token": self.refresh_token,
+                "client_secret": self.lwa_client_secret,
+            })
+            if resp.status_code != 200:
+                raise Exception(resp.text)
+            js = resp.json()
+            cache[self.refresh_token] = js["access_token"]
+            self.refresh_token = js["refresh_token"]
+            return js["access_token"]
+
+    @property
+    def auth_headers(self):
+        return {
+            "Amazon-Advertising-API-ClientId": self.lwa_client_id,
+            "Amazon-Advertising-API-Scope": self.profile_id,
+            "Authorization": f"Bearer {self.access_token}",
+        }
+
+    def _request(self, url_path: str, method: str = "GET", headers: dict = None, params: dict = None,
+                 body: dict = None):
+        head = self.auth_headers
+        if headers:
+            head.update(headers)
+        resp = requests.request(
+            method=method,
+            url=urljoin(URL_AD_API, url_path),
+            headers=head,
+            params=params,
+            json=body,
+        )
+        js = resp.json()
+        return js
+
+
+class SPClient(BaseClient):
+
+    def get_campaigns(self, **body):
+        url_path = "/sp/campaigns/list"
+        headers = {
+            "Accept": "application/vnd.spcampaign.v3+json",
+            "Content-Type": "application/vnd.spcampaign.v3+json"
+        }
+        return self._request(url_path, method="POST", headers=headers, body=body)
+
+    def iter_campaigns(self, **body) -> Iterator[dict]:
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_campaigns(**body)
+            yield from info["campaigns"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+        logger.info(f"总共数量：{info['totalResults']}")
+
+    def get_ad_groups(self, **body):
+        url_path = "/sp/adGroups/list"
+        headers = {
+            "Accept": "application/vnd.spadGroup.v3+json",
+            "Content-Type": "application/vnd.spadGroup.v3+json"
+        }
+        return self._request(url_path, method="POST", body=body, headers=headers)
+
+    def iter_adGroups(self, **body) -> Iterator[dict]:
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_ad_groups(**body)
+            yield from info["adGroups"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+        logger.info(f"总共数量：{info['totalResults']}")
+
+    def get_ads(self, **body):
+        url_path = "/sp/productAds/list"
+        headers = {
+            "Accept": "application/vnd.spproductAd.v3+json",
+            "Content-Type": "application/vnd.spproductAd.v3+json"
+        }
+        return self._request(url_path, method="POST", body=body, headers=headers)
+
+    def iter_ads(self, **body) -> Iterator[dict]:
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_ads(**body)
+            yield from info["productAds"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+        logger.info(f"总共数量：{info['totalResults']}")
+
+    def get_keywords(self, **body):
+        url_path = "/sp/keywords/list"
+        headers = {
+            "Accept": "application/vnd.spKeyword.v3+json",
+            "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
+        while True:
+            info: dict = self.get_keywords(**body)
+            yield from info["keywords"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+        logger.info(f"总共数量：{info['totalResults']}")
+
+    def get_targets(self, **body):
+        url_path = "/sp/targets/list"
+        headers = {
+            "Accept": "application/vnd.sptargetingClause.v3+json",
+            "Content-Type": "application/vnd.sptargetingClause.v3+json"
+        }
+        return self._request(url_path, method="POST", body=body, headers=headers)
+
+    def iter_targets(self, **body) -> Iterator[dict]:
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_targets(**body)
+            yield from info["targetingClauses"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+        logger.info(f"总共数量：{info['totalResults']}")
+
+    def get_budget(self, campaign_ids: list):
+        url_path = "/sp/campaigns/budget/usage"
+        body = {
+            "campaignIds": campaign_ids
+        }
+        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"):
+        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}
+        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)
+        url_path = "/v2/sp/keywords/bidRecommendations"
+        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"
+        headers = {
+            "Accept": "application/vnd.sbcampaignresouce.v4+json",
+            "Content-Type": "application/vnd.sbcampaignresouce.v4+json"
+        }
+        return self._request(url_path, method="POST", body=body, headers=headers)
+
+    def iter_campaigns(self, **body) -> Iterator[dict]:
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_campaigns(**body)
+            yield from info["campaigns"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+        # logger.info(f"总共数量：{info['totalResults']}")
+
+    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)
+
+    def iter_adGroups(self, **body) -> Iterator[dict]:
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_ad_groups(**body)
+            # print(info)
+            yield from info["adGroups"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+
+    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)
+
+    def iter_ads(self,**body):
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_ads(**body)
+            # 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_targets(self,**body):
+        url_path = "/sb/targets/list"
+        return self._request(url_path, method="POST", body=body)
+
+    def iter_targets(self,**body):
+        if "maxResults" not in body:
+            body["maxResults"] = 100
+        while True:
+            info: dict = self.get_targets(**body)
+            # print(info)
+            yield from info["targets"]
+            if not info.get("nextToken"):
+                break
+            body["nextToken"] = info["nextToken"]
+
+    def get_budget(self,campaignIds:list):
+        url_path = "/sb/campaigns/budget/usage"
+        body = {"campaignIds":campaignIds}
+        return self._request(url_path, method="POST",body=body)
+
+    def get_keyword_bidrecommendation(self,**body):
+        url_path = "/sb/recommendations/bids"
+        return self._request(url_path, method="POST", body=body)
+
+    def get_report(
+            self,
+            record_type: Literal['campaigns', 'adGroups', 'ads', 'targets', 'keywords'],
+            report_date: str,
+            metrics: List[str],
+            segment: Literal['placement', 'query'] = None,
+            creative_type: Literal['video', 'all'] = "all",
+            download: bool = True
+    ):
+        """
+        @param download: 是否下载文件
+        @param record_type:
+        @param report_date: 格式为YYYYMMDD，以请求的卖家市场所对应的时区为准，超过60天的报告不可用
+        @param metrics:
+        @param segment:
+        @param creative_type:
+            None：仅包含非视频广告
+            'video'：仅包含视频广告
+            'all'：包含视频和非视频广告
+        @return:
+        """
+        url = f"/v2/hsa/{record_type}/report"
+        body = {
+            "reportDate": report_date,
+            "metrics": ",".join(metrics),
+            "creativeType": creative_type,
+            "segment": segment
+        }
+        if record_type == "ads":
+            body["creativeType"] = "all"
+        ret = self._request(url, method="POST", body=body)
+        report_id = ret["reportId"]
+        status = ret["status"]
+        if status == "FAILURE":
+            raise Exception(ret)
+        logger.info(f"创建报告成功：{ret}")
+        while status == "IN_PROGRESS":
+            logger.debug(f"报告{report_id}正在处理中...")
+            time.sleep(3)
+            ret = self._request(f"/v2/reports/{report_id}")
+            status = ret["status"]
+            if status == "FAILURE":
+                raise Exception(ret)
+        logger.info(f"报告处理完成：{ret}")
+        if download:
+            self.download_report(report_id, str(self.data_path / f"sb_{record_type}.json.gz"))
+        else:
+            return ret
+
+    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)
+        logger.info(f"开始下载报告：{report_id}")
+        with open(file_path, "wb") as file:
+            for data in resp.iter_content(chunk_size=10 * 1024):
+                file.write(data)
+        logger.info(f"报告{report_id}下载完成：{file_path}")
+        if not decompress:
+            return file_path
+        de_file = gz_decompress(file_path)
+        logger.info(f"解压完成：{de_file}")
+        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):
+        url_path = "/v2/portfolios/extended"
+        return self._request(url_path)
+
+
+if __name__ == '__main__':
+    AWS_CREDENTIALS = {
+        'lwa_client_id': 'amzn1.application-oa2-client.ebd701cd07854fb38c37ee49ec4ba109',
+        'refresh_token': "Atzr|IwEBIL4ur8kbcwRyxVu_srprAAoTYzujnBvA6jU-0SMxkRgOhGjYJSUNGKvw24EQwJa1jG5RM76mQD2P22AKSq8qSD94LddoXGdKDO74eQVYl0RhuqOMFqdrEZpp1p4bIR6_N8VeSJDHr7UCuo8FiabkSHrkq7tsNvRP-yI-bnpQv4EayPBh7YwHVX3hYdRbhxaBvgJENgCuiEPb35Q2-Z6w6ujjiKUAK2VSbCFpENlEfcHNsjDeY7RCvFlwlCoHj1IeiNIaFTE9yXFu3aEWlExe3LzHv6PZyunEi88QJSXKSh56Um0e0eEg05rMv-VBM83cAqc5POmZnTP1vUdZO8fQv3NFLZ-xU6e1WQVxVPi5Cyqk4jYhGf1Y9t98N654y0tVvw74qNIsTrB-8bGS0Uhfe24oBEWmzObvBY3zhtT1d42myGUJv4pMTU6yPoS83zhPKm3LbUDEpBA1hvvc_09jHk7vUEAuFB-UAZzlht2C1yklzQ",
+        'lwa_client_secret': 'cbf0514186db4df91e04a8905f0a91b605eae4201254ced879d8bb90df4b474d',
+        '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"]))
+    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))
+    # 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.get_report(
+    #     record_type="campaigns",
+    #     report_date="20231008",
+    #     metrics=metrics
+    # )