sync_amz_data/sync_amz_data/public/amz_ad_client.py

import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
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
import s3fs
from s3fs import S3FileSystem

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 = 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):
    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)

        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=2,
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session = requests.session()
        self.session.mount("https://", adapter)

    @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 = self.session.request(
            method=method,
            url=urljoin(URL_AD_API, url_path),
            headers=head,
            params=params,
            json=body,
        )
        if resp.status_code == 429:
            raise RateLimitError(resp.headers.get("Retry-After"))
        if resp.status_code >= 400:
            raise Exception(resp.text)

        return resp.json()

    def get_profilesInfo(self):
        url_path = "/v2/profiles"
        return self._request(url_path)

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_budgetrecommendation(self, campaign_ids):
        url_path = "/sp/campaigns/budgetRecommendations"
        body = {
            "campaignIds": campaign_ids
        }
        headers = {
            "Accept": "application/vnd.budgetrecommendation.v3+json",
            "Content-Type": "application/vnd.budgetrecommendation.v3+json"
        }
        return self._request(url_path, method="POST", headers=headers, body=body)

    def iter_budgetrecommendation(self,campaign_ids):
        for i in range(0,len(campaign_ids),100):
            campaign_id = campaign_ids[i:i+100]
            info: list = self.get_budgetrecommendation(campaign_id)
            yield from info["budgetRecommendationsSuccessResults"]


    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))))
        url_path = "/v2/sp/keywords/bidRecommendations"
        body = {"adGroupId": adGroupId,
                "keywords": keywords}
        return self._request(url_path, method="POST", body=body)

    def get_targets_bid_recommendations(self,campaignId:str=None,
                                             adGroupId:str=None,
                                             asins:list=None,
                                             bid:float=None,
                                             keyword:str=None,
                                             userSelectedKeyword:bool=False,
                                             matchType:Literal["BROAD","EXACT","PHRASE"]="BROAD",
                                             recommendationType:Literal['KEYWORDS_FOR_ASINS','KEYWORDS_FOR_ADGROUP']="KEYWORDS_FOR_ASINS",
                                             sortDimension:Literal["CLICKS","CONVERSIONS","DEFAULT"]="DEFAULT",
                                             locale:Literal["ar_EG" ,"de_DE", "en_AE", "en_AU", "en_CA", "en_GB", "en_IN", "en_SA", "en_SG", "en_US",
                                                           "es_ES", "es_MX", "fr_FR", "it_IT", "ja_JP", "nl_NL", "pl_PL", "pt_BR", "sv_SE", "tr_TR", "zh_CN"]="en_US"):
        url_path = "/sp/targets/keywords/recommendations"
        body = {
                "recommendationType": recommendationType,
                "targets": [
                {
                "matchType": matchType,
                "keyword": keyword,
                "bid": bid,
                "userSelectedKeyword": userSelectedKeyword
                }
                ],
                "maxRecommendations": "200",
                "sortDimension": sortDimension,
                "locale": locale
                }
        if adGroupId is not None:
            body["campaignId"]=campaignId
            body["adGroupId"]= adGroupId
        else:
            body['asins'] = asins
        return self._request(url_path, method="POST", body=body)

    def get_v3_report(self,
                      groupby:list,
                      columns:list,
                      startDate:str,
                      endDate:str,
                      reportType: Literal['spCampaigns','spAdvertisedProduct' ,'spPurchasedProduct', 'spTargeting', 'spSearchTerm'],
                      timeUnit="DAILY",
                      download=True):
        """
        @param groupby: 聚合条件,[campaign,adGroup, searchTerm,purchasedAsin,campaignPlacement,targeting,searchTerm,advertiser,asin]
        columns: 需要获取的字段
        """
        url_path = "/reporting/reports"
        headers = {
            "Content-Type":"application/vnd.createasyncreportrequest.v3+json"
            }
        body = {
                "name":"SP campaigns report",
                "startDate":startDate,
                "endDate":endDate,
                "configuration":{
                    "adProduct":"SPONSORED_PRODUCTS",
                    "groupBy":groupby,
                    "columns":columns,
                    "reportTypeId":reportType,
                    "timeUnit":timeUnit,
                    "format":"GZIP_JSON"
                }
            }
        ret = self._request(url_path,method="POST",headers=headers,body=body)
        # print(ret)
        report_id = ret["reportId"]
        status = ret["status"]
        if status == "FAILURE":
            raise Exception(ret)
        logger.info(f"创建报告成功：{ret}")
        while status in ["PROCESSING","PENDING"]:
            logger.debug(f"报告{report_id}正在处理中...")
            time.sleep(4)
            try:
                ret = self._request(f"/reporting/reports/{report_id}")
            except:
                time.sleep(15)
                ret = self._request(f"/reporting/reports/{report_id}")
            print(ret)
            status = ret["status"]
            if status == "FAILURE":
                raise Exception(ret)
        logger.info(f"报告处理完成：{ret}")
        if download:
            pid = self.profile_id
            report_info = {'groupby': groupby,
                           'columns': columns,
                           'startDate': startDate,
                           'endDate': endDate,
                           'reportType': reportType,
                           'timeUnit': timeUnit,
                           'download': download}
            reportrel= self.download_v3_report(report_info,ret['url'],f"s3://reportforspsbsd/zosi/us/sp/{str(groupby)}_{startDate}_{endDate}_{reportType}_{str(pid)}.json.gz")
            return reportrel
        else:
            return ret

    def download_v3_report(self,report_info, url, file_path: str, decompress: bool = True) -> str:
        resp = requests.get(url, stream=True, allow_redirects=True)
        # print(resp)
        if resp.status_code in [200, 207]:
            kwargs = {'region_name': 'us-east-1', 'endpoint_url': "https://s3.amazonaws.com",
                      'aws_access_key_id': 'AKIARBAGHTGORIFN44VQ',
                      'aws_secret_access_key': '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'}
            s3_ = S3FileSystem(client_kwargs=kwargs)
            # print()
            with s3_.open(file_path, 'wb') as f:
                for data in resp.iter_content(chunk_size=10 * 1024):
                    f.write(data)
            if not decompress:
                return file_path
            with s3_.open(file_path, 'rb') as f:  # 读取s3数据
                data = gzip.GzipFile(fileobj=f, mode='rb')
                de_file = json.load(data)
            logger.info(f"解压完成：{de_file}")
            # print(de_file)
            return de_file
        else:
            logger.info(f"状态码{resp.status_code},开始重试")
            self.get_v3_report(report_info['groupby'], report_info['columns'], report_info['startDate'],
                               report_info['endDate'],
                               report_info['reportType'], report_info['timeUnit'], report_info['download'])

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 get_campaign_v3(self, campaignId):
        if campaignId is None:
            url_path = f'/sb/campaigns'
        else:
            url_path = f'/sb/campaigns/{campaignId}'
        return self._request(url_path, method="GET")

    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 get_keyword(self,keywordid):
        url_path = f'/sb/keywords/{keywordid}'
        return self._request(url_path,method="GET")

    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_v3_report(self,
                      groupby:list,
                      columns:list,
                      startDate:str,
                      endDate:str,
                      reportType: Literal['sbCampaigns', 'sbPurchasedProduct', 'sbTargeting', 'sbSearchTerm'],
                      timeUnit="DAILY",
                      download=True):
        """
        Now about reportType is only sbPurchasedProduct available.
        @param groupby: 聚合条件
        @param columns: 需要获取的字段[campaign,purchasedAsin,targeting,searchTerm]
        @param startDate: 请求开始的日期
        @param endDate: 请求结束的日期
        @param reportType: 广告类型
        @param timeUnit: 时间指标-[DAILY, SUMMARY]
        @param download: 下载报告
        """
        url_path = "/reporting/reports"
        headers = {
            "Content-Type":"application/vnd.createasyncreportrequest.v3+json"
            }
        body = {
                "name":"SB campaigns report",
                "startDate":startDate,
                "endDate":endDate,
                "configuration":{
                    "adProduct":"SPONSORED_BRANDS",
                    "groupBy":groupby,
                    "columns":columns,
                    "reportTypeId":reportType,
                    "timeUnit":timeUnit,
                    "format":"GZIP_JSON"
                }
            }
        ret = self._request(url_path,method="POST",headers=headers,body=body)
        # print(ret)
        report_id = ret["reportId"]
        status = ret["status"]
        if status == "FAILURE":
            raise Exception(ret)
        logger.info(f"创建报告成功：{ret}")
        while status in ["PROCESSING","PENDING"]:
            logger.debug(f"报告{report_id}正在处理中...")
            time.sleep(4)
            try:
                ret = self._request(f"/reporting/reports/{report_id}")
            except:
                time.sleep(15)
                ret = self._request(f"/reporting/reports/{report_id}")
            print(ret)
            status = ret["status"]
            if status == "FAILURE":
                raise Exception(ret)
        logger.info(f"报告处理完成：{ret}")
        if download:
            pid = self.profile_id
            report_info = {'groupby': groupby,
                           'columns': columns,
                           'startDate': startDate,
                           'endDate': endDate,
                           'reportType': reportType,
                           'timeUnit': timeUnit,
                           'download': download}
            reportrel= self.download_v3_report(report_info,ret['url'],f"s3://reportforspsbsd/zosi/us/sb/{startDate}_{endDate}_{reportType}_{str(groupby)}_{str(pid)}.json.gz")
            return reportrel
        else:
            return ret

    def download_v3_report(self, report_info,url, file_path: str, decompress: bool = True) -> str:
        resp = requests.get(url, stream=True, allow_redirects=True)
        # print(resp)
        if resp.status_code in [200, 207]:
            kwargs = {'region_name': 'us-east-1', 'endpoint_url': "https://s3.amazonaws.com",
                      'aws_access_key_id': 'AKIARBAGHTGORIFN44VQ',
                      'aws_secret_access_key': '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'}
            s3_ = S3FileSystem(client_kwargs=kwargs)
            # print()
            with s3_.open(file_path, 'wb') as f:
                for data in resp.iter_content(chunk_size=10 * 1024):
                    f.write(data)
            if not decompress:
                return file_path
            with s3_.open(file_path, 'rb') as f:  # 读取s3数据
                data = gzip.GzipFile(fileobj=f, mode='rb')
                de_file = json.load(data)
            logger.info(f"解压完成：{de_file}")
            # print(de_file)
            return de_file
        else:
            logger.info(f"状态码{resp.status_code},开始重试")
            self.get_v3_report(report_info['groupby'], report_info['columns'], report_info['startDate'],
                               report_info['endDate'],
                               report_info['reportType'], report_info['timeUnit'], report_info['download'])

    def get_v2_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(4)
            try:
                ret = self._request(f"/v2/reports/{report_id}")
            except:
                time.sleep(15)
                ret = self._request(f"/v2/reports/{report_id}")
            print(ret)
            status = ret["status"]
            if status == "FAILURE":
                raise Exception(ret)
        logger.info(f"报告处理完成：{ret}")
        if download:
            pid = self.profile_id
            report_info = {"record_type": record_type, "report_date": report_date, "metrics": metrics,
                           "segment": segment, "creative_type": creative_type, "download": download}
            reportrel= self.download_v2_report(report_info,report_id, f"s3://reportforspsbsd/zosi/us/sb/{str(report_date)}_{record_type}_{creative_type}_{segment}_{str(pid)}.gz")
            return reportrel
        else:
            return ret

    def download_v2_report(self,report_info,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, allow_redirects=True)
        if resp.status_code in [200, 207]:
            logger.info(f"开始下载报告：{report_id}")
            kwargs = {'region_name': 'us-east-1', 'endpoint_url': "https://s3.amazonaws.com",
                      'aws_access_key_id': 'AKIARBAGHTGORIFN44VQ',
                      'aws_secret_access_key': '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'}
            s3_ = S3FileSystem(client_kwargs=kwargs)
            # print()
            with s3_.open(file_path, 'wb') as f:
                for data in resp.iter_content(chunk_size=10 * 1024):
                    f.write(data)

            logger.info(f"报告{report_id}下载完成：{file_path}")
            if not decompress:
                return file_path
            with s3_.open(file_path, 'rb') as f:  # 读取s3数据
                data = gzip.GzipFile(fileobj=f, mode='rb')
                de_file = json.load(data)
            logger.info(f"解压完成：{de_file}")
            # print(de_file)
            return de_file
        else:
            logger.info(f"状态码{resp.status_code},开始重试")
            self.get_v2_report(report_info['record_type'], report_info['report_date'], report_info['metrics'],
                               report_info['segment'], report_info['creative_type'], report_info['download'])

class SDClient(BaseClient):
    def get_campaigns(self, **params) -> List[dict]:
        url_path = "/sd/campaigns"
        return self._request(url_path, params=params)

    def get_campaigns_extended(self, **params) -> List[dict]:
        url_path = "/sd/campaigns/extended"
        return self._request(url_path, params=params)

    def get_adGroups(self,**params):
        url_path = '/sd/adGroups'
        return self._request(url_path, params=params)

    def iter_adGroups(self,**param):
        if "startIndex" not in param:
            param["startIndex"] = 0
            param["count"] = 5000
        while True:
            info:list = self.get_adGroups(**param)
            # print(info)
            if len(info) == 0:
                break
            param["startIndex"] += 5000
            yield info

    def get_ads(self,**params):
        url_path = '/sd/productAds'
        return self._request(url_path, params=params)

    def iter_ads(self,**param):
        if "startIndex" not in param:
            param["startIndex"] = 0
            param["count"] = 5000
        while True:
            info:list = self.get_ads(**param)
            # print(info)
            if len(info) == 0:
                break
            param["startIndex"] += 5000
            yield info

    def get_targets(self,**params):
        url_path = '/sd/targets'
        return self._request(url_path, params=params)

    def iter_targets(self,**param):
        if "startIndex" not in param:
            param["startIndex"] = 0
            param["count"] = 5000
        while True:
            info:list = self.get_targets(**param)
            # print(info)
            if len(info) == 0:
                break
            param["startIndex"] += 5000
            yield info

    def get_budget(self, campaignIds: list):
        url_path = "/sd/campaigns/budget/usage"
        body = {"campaignIds": campaignIds}
        return self._request(url_path, method="POST", body=body)

    def get_target_bidrecommendation(self,tactic:str,products:list,typeFilter:list,themes:dict,locale:str='en_US'):#
        url_path = '/sd/targets/recommendations'
        headers ={
            'Content-Type':"application/vnd.sdtargetingrecommendations.v3.3+json",
            'Accept':"application/vnd.sdtargetingrecommendations.v3.3+json"
        }
        # "tactic":"T00020",
        # "products":[{"asin":"B00MP57IOY"}],
        # "typeFilter":["PRODUCT"],
        # "themes":{"product":[{"name":"TEST","expression":[{"type":"asinBrandSameAs"}]}]}
        body = {
            "tactic":tactic,
            "products":products,
            "typeFilter":typeFilter,
            "themes":themes
                 }

        return self._request(url_path, method="POST", headers=headers,body=body,params={"locale":locale})

    def get_v3_report(self,
                      groupby:list,
                      columns:list,
                      startDate:str,
                      endDate:str,
                      reportType: Literal['sdCampaigns', 'sdPurchasedProduct', 'sdTargeting', 'sdSearchTerm'],
                      timeUnit="DAILY",
                      download=True):
        """
        Now about reportType is only sbPurchasedProduct available.
        @param groupby: 聚合条件
        @param columns: 需要获取的字段[campaign,purchasedAsin,targeting,searchTerm]
        @param startDate: 请求开始的日期
        @param endDate: 请求结束的日期
        @param reportType: 广告类型
        @param timeUnit: 时间指标-[DAILY, SUMMARY]
        @param download: 下载报告
        """
        url_path = "/reporting/reports"
        headers = {
            "Content-Type":"application/vnd.createasyncreportrequest.v3+json"
            }
        body = {
                "name":"SD campaigns report",
                "startDate":startDate,
                "endDate":endDate,
                "configuration":{
                    "adProduct":"SPONSORED_DISPLAY",
                    "groupBy":groupby,
                    "columns":columns,
                    "reportTypeId":reportType,
                    "timeUnit":timeUnit,
                    "format":"GZIP_JSON"
                }
            }
        ret = self._request(url_path,method="POST",headers=headers,body=body)
        # print(ret)
        report_id = ret["reportId"]
        status = ret["status"]
        if status == "FAILURE":
            raise Exception(ret)
        logger.info(f"创建报告成功：{ret}")
        while status in ["PROCESSING","PENDING"]:
            logger.debug(f"报告{report_id}正在处理中...")
            time.sleep(4)
            try:
                ret = self._request(f"/reporting/reports/{report_id}")
            except:
                time.sleep(15)
                ret = self._request(f"/reporting/reports/{report_id}")
            print(ret)
            status = ret["status"]
            if status == "FAILURE":
                raise Exception(ret)
        logger.info(f"报告处理完成：{ret}")
        if download:
            pid = self.profile_id
            report_info = {'groupby':groupby,
                      'columns':columns,
                      'startDate':startDate,
                      'endDate':endDate,
                      'reportType': reportType,
                      'timeUnit':timeUnit,
                      'download':download}
            reportrel= self.download_v3_report(report_info,ret['url'],f"s3://reportforspsbsd/zosi/us/sd/{startDate}_{endDate}_{reportType}_{str(groupby)}_{str(pid)}.json.gz")
            return reportrel
        else:
            return ret

    def download_v3_report(self,report_info, url, file_path: str, decompress: bool = True) -> str:
        resp = requests.get(url, stream=True, allow_redirects=True)
        # print(resp)
        if resp.status_code in [200,207]:
            kwargs = {'region_name': 'us-east-1', 'endpoint_url': "https://s3.amazonaws.com",
                      'aws_access_key_id': 'AKIARBAGHTGORIFN44VQ',
                      'aws_secret_access_key': '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'}
            s3_ = S3FileSystem(client_kwargs=kwargs)
            # print()
            with s3_.open(file_path, 'wb') as f:
                for data in resp.iter_content(chunk_size=10 * 1024):
                    f.write(data)
            if not decompress:
                return file_path
            with s3_.open(file_path, 'rb') as f:  # 读取s3数据
                data = gzip.GzipFile(fileobj=f, mode='rb')
                de_file = json.load(data)
            logger.info(f"解压完成：{de_file}")
            # print(de_file)
            return de_file
        else:
            logger.info(f"状态码{resp.status_code},开始重试")
            self.get_v3_report(report_info['groupby'],report_info['columns'],report_info['startDate'],report_info['endDate'],
                               report_info['reportType'],report_info['timeUnit'],report_info['download'])

    def get_v2_report(
            self,
            record_type: Literal['campaigns', 'adGroups', 'productAds', 'targets', 'asins'],
            report_date: str,
            metrics: List[str],
            segment: Literal['matchedTarget'] = None,
            tactic: Literal['T00020', 'T00030'] = None,
            download: bool = True
    ):
        """
        @param download: 是否下载文件
        @param record_type:
        @param report_date: 格式为YYYYMMDD，以请求的卖家市场所对应的时区为准，超过60天的报告不可用
        @param metrics:
        @param segment:
        @param tactic:
            T00020: contextual targeting
            T00030: audience targeting
        @return:
        """
        url = f"/sd/{record_type}/report"
        body = {
            "reportDate": report_date,
            "metrics": ",".join(metrics),
            "tactic": tactic,
            "segment": segment
        }
        ret = self._request(url, method="POST", body=body)
        report_id = ret["reportId"]
        status = ret["status"]
        print(ret)
        if status == "FAILURE":
            raise Exception(ret)
        logger.info(f"创建报告成功：{ret}")
        while status == "IN_PROGRESS":
            logger.debug(f"报告{report_id}正在处理中...")
            time.sleep(4)
            try:
                ret = self._request(f"/v2/reports/{report_id}")
            except:
                time.sleep(15)
                ret = self._request(f"/v2/reports/{report_id}")
            print(ret)
            status = ret["status"]
            if status == "FAILURE":
                raise Exception(ret)
        logger.info(f"报告处理完成：{ret}")
        if download:
            pid = self.profile_id
            report_info = {"record_type":record_type,"report_date":report_date,"metrics":metrics,"segment":segment,"tactic":tactic,"download":download}
            reportrel= self.download_v2_report(report_info,report_id, f"s3://reportforspsbsd/zosi/us/sd/{str(report_date)}_{record_type}_{tactic}_{segment}_{str(pid)}.gz")
            return reportrel
        else:
            return ret

    def download_v2_report(self, report_info,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, allow_redirects=True)
        # print(resp.status_code)
        if resp.status_code in [200,207]:
            logger.info(f"开始下载报告：{report_id}")
            kwargs = {'region_name': 'us-east-1', 'endpoint_url': "https://s3.amazonaws.com",
                      'aws_access_key_id': 'AKIARBAGHTGORIFN44VQ',
                      'aws_secret_access_key': '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'}
            s3_ = S3FileSystem(client_kwargs=kwargs)
            # print()
            with s3_.open(file_path, 'wb') as f:
                for data in resp.iter_content(chunk_size=10 * 1024):
                    # print(resp.text)
                    f.write(data)

            logger.info(f"报告{report_id}下载完成：{file_path}")
            if not decompress:
                return file_path
            with s3_.open(file_path, 'rb') as f:  # 读取s3数据
                data = gzip.GzipFile(fileobj=f, mode='rb')
                de_file = json.load(data)
            logger.info(f"解压完成：{de_file}")
            # print(de_file)
            return de_file
        else:
            logger.info(f"状态码{resp.status_code},开始重试")
            self.get_v2_report(report_info['record_type'],report_info['report_date'],report_info['metrics'],
                               report_info['segment'],report_info['tactic'],report_info['download'])


class Account(BaseClient):
    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 = {
        '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 = SDClient(**AWS_CREDENTIALS)

    # sb.get_v3_report(groupby=["campaign"],columns=["impressions","date"],startDate="2023-10-28",endDate="2023-10-28",reportType="sdCampaigns",timeUnit="DAILY")
    # print(list(sb.iter_budgetrecommendation(['147691704431878', '60271227965408', '218847416529861', '31392773399718', '262035857074479', '198816492559645', '142984587494761', '56817060975858', '25915710734770', '129767234792339', '157524678908837', '49318230260950', '108960112862154', '62467550507341', '251456127331427', '191378898823474', '120392223446402', '247773821977107', '280377532791660', '184950658810783', '164014978241334', '123919526909988', '130880921982647', '126821145840591', '64491219404125', '38337331569585', '116386348992750', '225036289048302', '156661647012185', '144776397901336', '135236490839193', '169772039957761', '80512764023579', '34385013996533', '120697809283682', '246829191189710', '209056522772792', '232902564083581', '3932823953797', '279860689033033', '148947795709523', '97303001527474', '273793489896841', '166028784537215', '39300113616652', '170015910542201', '273987458393412', '171417956384778', '179418644846675', '93362211056849', '183219549121837', '92109046525620', '209610737481328', '64551171635322', '104421263246026', '107129361457199', '180558158817613', '196152550852335', '274383816868050', '153176862530690', '202668080336520', '139339123933891', '216562245832724', '44126930182871', '158793558529257', '163377434993147', '129760407316857', '259206875182868', '254478421786009', '113623886210537', '123264383028090', '114995189965872', '126327624499318', '111423039176174', '218824374284099', '268581491758278', '221198600183221', '252229497958387', '25969190496597', '147213408548844', '120873185867848', '121127240307802', '149837836567172', '244566073561396', '170971926269997', '50598109145873', '94661978287830', '172459323917375', '79239463046520', '161538254020266', '122877215020077', '19390096189319', '84671881754842', '56035811352399', '127311973972957', '99871114075939', '216044477480148', '212849641880903', '96517925473279', '140235725339419', '141444557365069', '2246672580150', '268090880021982', '94042224869659', '49225748581620', '32725632149299', '40908978446751', '146895818081903', '119917780655231', '146458404145911', '118070860722352', '52022434573022', '272598548490634', '16095673755650', '172664047341294', '256277301042945', '196836379347636', '155301894644769', '274223966125410', '248132117356305', '57660814157093', '50226094984793', '28888390047403', '102108408531327', '30079789635585', '223870235275688', '184420890768530', '29862567915039', '39565569376716', '5130702823897', '193159277950903', '155681125793546', '147688196248459', '246947393077054', '31511403651405', '102765494942726', '37856299838393', '231975314936513', '241347601457737', '59068161406145', '195628702544128', '225472432237809', '247179558013703', '114349375196504', '247845805126534', '50893503004023', '21955509658149', '66932634825184', '281441197839505', '255188169284953', '127289804144979', '168277283490230', '67066099773166', '14095630154088', '14246020074735', '272850527640233', '107029528476360', '174968043962800', '221862209153056', '116700370772313', '104020924872149', '97962041521658', '173158222344275', '185251252451477', '74364105565262', '210932538222999', '63297563429758', '276793053074864', '69715214871405', '95970937392401', '32561213134196', '33142883019280', '79216188082798', '59643351291635', '76327385728260', '116202606958791', '167409692560139', '171577662027638', '41421288964538', '150057207183605', '154717958881150', '114788039648066', '107895719250913', '264030622380992', '172196786467310', '197934421091867', '278388674751612', '45379253028975', '122119878478906', '4533827464348', '130815119156456', '137297479866563', '250767198691934', '139253473761708', '112336508594346', '132816688267091', '57709263376416', '249118660318975', '196638816684572', '92891389830198', '208322940506617', '186709333461759', '185422701325753', '270743781107196', '150694054972074', '193071867166263', '279972459852484', '119252950300630', '124990749954236', '266097963051545', '234638916539784', '20605146781735', '70768902397459', '154986591065926', '159501779813069', '162103896356568', '460593835599747', '410212175498241', '557726418497539', '500143110670441', '355416669121026', '431780944586985', '404895228651677', '388038108990987', '464118197356560', '521980826814295', '469329788053854', '326468917818944', '552954913725893', '542546450513278', '325523075677132', '392468912549092', '527886948735997', '329385433739622', '485978120289879', '325681364822312', '555651735003802', '303081891898256', '290254871141701', '485921949741399', '403591067235790', '499317623967079', '293479361575582', '495483931857167', '559648871033557', '336635682793609', '471666122828785', '537556291117721', '523813506535076', '497250432979707', '442860218481252', '448893131612583', '281694643275712', '433377458106838', '442612845985546', '338829853627078', '392584497224138', '496308094986157', '418568230991874', '378377235258783', '478649462904301', '369874181350855', '420881761417240', '360520776254966', '479975368002715', '390542812871179', '395983847529920', '390131732881184', '383692739831773', '444758167627464', '453566341896795', '496145304023336', '427176120966955', '310665337034129', '477718522757455', '535119828991333', '375580793899192', '464596753380107', '327731861279726', '294079129793082', '360342567843115', '310372350013643', '235356299301347', '265522205087323', '228344360452595', '264342228103465', '122100911164709', '232859308822086', '45261323402141', '65662811257626', '234144786032766', '219449168841685', '128805550207790', '52164236957007', '120413947884525', '247101003653911', '22887864793841', '66188387334596', '102602610195475', '14284441019308', '125676608715613', '208268030289407', '237637021820467', '83135684198446', '154786784931406', '34854308519962', '8788604610896', '46715264141017', '139624985843170', '220476381163737', '109100736225759', '56729203013140', '88373528260613', '3312332033772', '37834945448724', '94120697716118', '193841489621149', '176450446850186', '166966792230306', '168345583834458', '63645396093211', '21250682621308', '43999955361946', '91246816921037', '256738400830651', '141723939372941', '2925041027398', '209321026962101', '260728226512366', '139367584185686', '264968517312373', '46556032756649', '55837615659599', '6217208882388', '212383697581072', '13384963604097', '220553842518820', '216909098017681', '164205523483164', '140262174434441', '204728710865941', '257987760095645', '74010928042922', '218130080477762', '40247744447213', '243863605109393', '160210698808714', '23666771936859', '209074182215992', '183440625321762', '10008326799771', '251040907550889', '50479853741108', '98121074839083', '182666986837920', '75110020202471', '277378379542982', '137193580729176', '84426787297480', '238858183340317', '270834245781899']
    # )))
    # print(list(sb.iter_campaigns(**{"stateFilter": {
    #                     "include": ["ARCHIVED","PAUSED","ENABLED"]
    #                     }})))
    # print(sb.get_keyword_bidrecommendation(**{'campaignId': 27333596383941, 'keywords': [
    #     {"matchType": 'broad', "keywordText": "4k security camera system"}]}))
    # a = list(sd.iter_targets(**{"campaignIdFilter":"257424912382921"})) #list(sd.iter_targets())#
    # print(a,len(a))

    # sb = SBClient(**AWS_CREDENTIALS)

    # print(sd.get_v2_report(record_type="campaigns",report_date="20231020",tactic="T00030",metrics=['impressions']))