123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320 |
- 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__)
- def shop_infos(profile_id):
- resp_rel = requests.get("http://192.168.1.225/api/ad_manage/profiles/",headers={"X-Token": "da4ab6bc5cbf1dfa"})
- data = resp_rel.json().get("data")
- profile_info = {}
- for info in data:
- if info.get("profile_id") in [int(profile_id),str(profile_id)]:
- profile_info["profile_id"] = profile_id
- profile_info["refresh_token"] = info.get("refresh_token")
- profile_info["account_name"] = info.get("account_name")
- profile_info["advertiser_id"] = info.get("advertiser_id")
- profile_info["country_code"] = info.get("country_code")
- profile_info["marketplace_str_id"] =info.get("marketplace_str_id")
- profile_info["time_zone"] = info.get("time_zone")
- return profile_info
- return resp_rel.text
- 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 iter_adgroup_bidrecommendation(self,campaignId: str, adGroupId: str, targetingExpressions: list,
- recommendationType: str = "BIDS_FOR_EXISTING_AD_GROUP"):
- for i in range(0,len(targetingExpressions),100):
- try:
- info = self.get_adgroup_bidrecommendation(campaignId,adGroupId,targetingExpressions[i:i+100],recommendationType)
- yield from info['bidRecommendations']
- except:
- # print("空值")
- return iter([])
- 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_bidrecommendationList(self,adGroupId,expressions):
- url_path = "/v2/sp/targets/bidRecommendations"
- body = {
- "adGroupId": adGroupId,
- "expressions": expressions
- }
- return self._request(url_path,method="POST",body=body)
- def iter_bidrecommendationList(self, adGroupId, expressions):
- for i in range(0,len(expressions), 10):
- try:
- info = self.get_bidrecommendationList(adGroupId, expressions[i:i+10])
- out = info['recommendations']
- for i in out:
- i['adGroupId'] = adGroupId
- yield from out
- except:
- # print("空值")
- return iter([])
- def get_campaignNegativekeyword(self,**body):
- url_path = '/sp/campaignNegativeKeywords/list'
- headers = {
- "Accept": "application/vnd.spCampaignNegativeKeyword.v3+json",
- "Content-Type": "application/vnd.spCampaignNegativeKeyword.v3+json"
- }
- return self._request(url_path,method="POST",body=body,headers=headers)
- def iter_campaignNegativekeyword(self,**body):
- if "maxResults" not in body:
- body["maxResults"] = 100
- while True:
- info: dict = self.get_campaignNegativekeyword(**body)
- yield from info["campaignNegativeKeywords"]
- if not info.get("nextToken"):
- break
- body["nextToken"] = info["nextToken"]
- def get_negativekeyword(self,**body):
- url_path = '/sp/negativeKeywords/list'
- headers = {
- "Accept": "application/vnd.spNegativeKeyword.v3+json",
- "Content-Type": "application/vnd.spNegativeKeyword.v3+json"
- }
- return self._request(url_path,method="POST",body=body,headers=headers)
- def iter_negativekeyword(self,**body):
- if "maxResults" not in body:
- body["maxResults"] = 100
- while True:
- info: dict = self.get_negativekeyword(**body)
- yield from info["negativeKeywords"]
- if not info.get("nextToken"):
- break
- body["nextToken"] = info["nextToken"]
- def get_campaignNegativetargeting(self, **body):
- url_path = '/sp/campaignNegativeTargets/list'
- headers = {
- "Accept": "application/vnd.spCampaignNegativeTargetingClause.v3+json",
- "Content-Type": "application/vnd.spCampaignNegativeTargetingClause.v3+json"
- }
- return self._request(url_path, method="POST", body=body, headers=headers)
- def iter_campaignNegativetargeting(self, **body):
- if "maxResults" not in body:
- body["maxResults"] = 100
- while True:
- info: dict = self.get_campaignNegativetargeting(**body)
- yield from info["campaignNegativeTargetingClauses"]
- if not info.get("nextToken"):
- break
- body["nextToken"] = info["nextToken"]
- def get_negativetargeting(self, **body):
- url_path = '/sp/negativeTargets/list'
- headers = {
- "Accept": "application/vnd.spNegativeTargetingClause.v3+json",
- "Content-Type": "application/vnd.spNegativeTargetingClause.v3+json"
- }
- return self._request(url_path, method="POST", body=body, headers=headers)
- def iter_negativetargeting(self, **body):
- if "maxResults" not in body:
- body["maxResults"] = 100
- while True:
- info: dict = self.get_negativetargeting(**body)
- yield from info["negativeTargetingClauses"]
- if not info.get("nextToken"):
- break
- body["nextToken"] = info["nextToken"]
- 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"):
- """
- adGroupId与asins只能选择一个参数填写。
- @param campaignId: pass
- @param adGroupId: 如果recommendationType为KEYWORDS_FOR_ADGROUP时,为必填项
- @param asins:如果recommendationType为KEYWORDS_FOR_ASINS时,为必填项
- @param userSelectedKeyword:是否参考选择的keyword
- @param recommendationType:类型选择,必填
- @param locale:pass
- """
- headers = {"Accept":"application/vnd.spkeywordsrecommendation.v3+json",
- "Content-Type":"application/vnd.spkeywordsrecommendation.v3+json"}
- 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,headers=headers)
- def get_v3_report(self,
- groupby:list,
- columns:list,
- startDate:str,
- endDate:str,
- reportType: Literal['spCampaigns','spAdvertisedProduct' ,'spPurchasedProduct', 'spTargeting', 'spSearchTerm'],
- timeUnit="DAILY",
- download=True,
- Rewrite=False):
- """
- @param groupby: 聚合条件,[campaign,adGroup, searchTerm,purchasedAsin,campaignPlacement,targeting,searchTerm,advertiser,asin]
- columns: 需要获取的字段
- """
- #######################################################################################################################################################
- if Rewrite:
- 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, None,
- f"s3://reportforspsbsd/zosi/us/sp/{str(groupby)}_{startDate}_{endDate}_{reportType}_{str(pid)}.json.gz")
- return reportrel
- ##############################################################################################################################################
- 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,Rewrite=False) -> str:
- ##################################################################################
- if Rewrite:
- 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)
- 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
- ##################################################################################
- resp = requests.get(url, stream=True, allow_redirects=True)
- # print(resp)
- if resp.status_code in [200, 207]:
- try:
- 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
- except:
- try:
- with open(f"{report_info['groupby']}_{report_info['startDate']}_{report_info['endDate']}_{report_info['reportType']}_{str(self.profile_id)}.json.gz", 'wb') as f:
- for data in resp.iter_content(chunk_size=10 * 1024):
- f.write(data)
- if not decompress:
- return file_path
- with open(f"{report_info['groupby']}_{report_info['startDate']}_{report_info['endDate']}_{report_info['reportType']}_{str(self.profile_id)}.json.gz", '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
- except:
- logger.info(f"过期开始重试")
- 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'])
- 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,
- Rewrite=False):
- """
- 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: 下载报告
- """
- ##############################################################################################
- if Rewrite:
- 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, None,
- f"s3://reportforspsbsd/zosi/us/sb/{startDate}_{endDate}_{reportType}_{str(groupby)}_{str(pid)}.json.gz")
- return reportrel
- ###############################################################################################
- 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,Rewrite=False) -> str:
- ###########################################################################################
- if Rewrite:
- 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)
- 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
- #############################################################################################
- resp = requests.get(url, stream=True, allow_redirects=True)
- # print(resp)
- if resp.status_code in [200, 207]:
- try:
- 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
- except:
- try:
- with open(
- f"{report_info['groupby']}_{report_info['startDate']}_{report_info['endDate']}_{report_info['reportType']}_{str(self.profile_id)}.json.gz",
- 'wb') as f:
- for data in resp.iter_content(chunk_size=10 * 1024):
- f.write(data)
- if not decompress:
- return file_path
- with open(
- f"{report_info['groupby']}_{report_info['startDate']}_{report_info['endDate']}_{report_info['reportType']}_{str(self.profile_id)}.json.gz",
- '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
- except:
- logger.info(f"过期开始重试")
- 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'])
- 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,
- Rewrite=False
- ):
- """
- @param download: 是否下载文件
- @param record_type:
- @param report_date: 格式为YYYYMMDD,以请求的卖家市场所对应的时区为准,超过60天的报告不可用
- @param metrics:
- @param segment:
- @param creative_type:
- None:仅包含非视频广告
- 'video':仅包含视频广告
- 'all':包含视频和非视频广告
- @return:
- """
- ############################################################################################
- if Rewrite:
- 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, None,
- f"s3://reportforspsbsd/zosi/us/sb/{str(report_date)}_{record_type}_{creative_type}_{segment}_{str(pid)}.gz"
- )
- return reportrel
- #############################################################################################
- 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,Rewrite=False) -> str:
- ################################################################################
- if Rewrite:
- 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, '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
- ################################################################################
- 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]:
- try:
- 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
- except:
- try: #s3://reportforspsbsd/zosi/us/sb/{str(report_date)}_{record_type}_{creative_type}_{segment}_{str(pid)}.gz
- # report_info = {"record_type": record_type, "report_date": report_date, "metrics": metrics,
- # "segment": segment, "creative_type": creative_type, "download": download}
- with open(
- f"{report_info['record_type']}_{report_info['report_date']}_{report_info['metrics']}_{report_info['segment']}_{report_info['creative_type']}_{str(self.profile_id)}.json.gz",
- 'wb') as f:
- for data in resp.iter_content(chunk_size=10 * 1024):
- f.write(data)
- if not decompress:
- return file_path
- with open(
- f"{report_info['record_type']}_{report_info['report_date']}_{report_info['metrics']}_{report_info['segment']}_{report_info['creative_type']}_{str(self.profile_id)}.json.gz",
- '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
- except:
- logger.info(f"过期开始重试")
- 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'])
- 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,
- Rewrite=False):
- """
- 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: 下载报告
- """
- ###############################################################################################
- if Rewrite:
- 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, None,
- f"s3://reportforspsbsd/zosi/us/sd/{startDate}_{endDate}_{reportType}_{str(groupby)}_{str(pid)}.json.gz")
- return reportrel
- ##################################################################################################
- 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,Rewrite=False) -> str:
- #######################################################################
- if Rewrite:
- 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)
- 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
- # logger.info(f"解压完成:{de_file}")
- # print(de_file)
- #########################################################################
- resp = requests.get(url, stream=True, allow_redirects=True)
- # print(resp)
- if resp.status_code in [200, 207]:
- try:
- 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
- except:
- try:
- with open(
- f"{report_info['groupby']}_{report_info['startDate']}_{report_info['endDate']}_{report_info['reportType']}_{str(self.profile_id)}.json.gz",
- 'wb') as f:
- for data in resp.iter_content(chunk_size=10 * 1024):
- f.write(data)
- if not decompress:
- return file_path
- with open(
- f"{report_info['groupby']}_{report_info['startDate']}_{report_info['endDate']}_{report_info['reportType']}_{str(self.profile_id)}.json.gz",
- '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
- except:
- logger.info(f"过期开始重试")
- 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'])
- 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,
- Rewrite=False
- ):
- """
- @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]:
- try:
- 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
- except:
- try:
- with open(
- f"{report_info['record_type']}_{report_info['report_date']}_{report_info['metrics']}_{report_info['segment']}_{report_info['tactic']}_{str(self.profile_id)}.json.gz",
- 'wb') as f:
- for data in resp.iter_content(chunk_size=10 * 1024):
- f.write(data)
- if not decompress:
- return file_path
- with open(
- f"{report_info['record_type']}_{report_info['report_date']}_{report_info['metrics']}_{report_info['segment']}_{report_info['tactic']}_{str(self.profile_id)}.json.gz",
- '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
- except:
- logger.info(f"过期开始重试")
- self.get_v2_report(report_info['record_type'], report_info['report_date'], report_info['metrics'],
- report_info['segment'], report_info['tactic'], report_info['download'])
- 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)
- rel = sp.iter_negativekeyword()
- print(list(rel))
- # print(rel)
|