datainsert系列代码

sync_amz_data/public/amz_ad_client.py

@@ -134,6 +134,17 @@ class SPClient(BaseClient):
             body["nextToken"] = info["nextToken"]
         logger.info(f"总共数量：{info['totalResults']}")
 
+    def get_spbudgetrecommendation(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 get_ad_groups(self, **body):
         url_path = "/sp/adGroups/list"
         headers = {

sync_amz_data/tasks/datainsert/SP/mysql_datainsert_sp_budget_recommendation.py

@@ -0,0 +1,137 @@
+import requests
+from urllib.parse import urljoin
+from sync_amz_data.public.amz_ad_client import SPClient
+from sync_amz_data.settings import AWS_LWA_CLIENT
+import pandas as pd
+import json
+
+
+class RateLimitError(Exception):
+    def __init__(self, retry_after: str = None):
+        self.retry_after = retry_after
+
+
+def request(url_path: str, method: str = "GET", head: dict = None, params: dict = None, body: dict = None):
+    ADS = "http://192.168.1.23:8001/"
+    resp = requests.session().request(
+        method=method,
+        url=urljoin(ADS, 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()
+
+
+class SpCampaign:
+    def __init__(self, profile_id, portfolioId: list = None):
+        self.profile_id = profile_id
+        self.portfolioId = portfolioId
+        self.re_url_path = "api/ad_manage/profiles/"
+        self.camid_url_path = "api/ad_manage/spcampaigns/"
+        self.upcreate_url_path = "api/spbudgetrecommendation/list/"
+        self.heads = {'X-Token': "da4ab6bc5cbf1dfa"}
+        self.refresh_token = self.get_refresh_token()
+        self.lwa_client_id = AWS_LWA_CLIENT['lwa_client_id']
+        self.lwa_client_secret = AWS_LWA_CLIENT['lwa_client_secret']
+        self.AWS_CREDENTIALS = {
+            'lwa_client_id': self.lwa_client_id,
+            'lwa_client_secret': self.lwa_client_secret,
+            'refresh_token': self.refresh_token,
+            'profile_id': self.profile_id
+        }
+
+    def get_refresh_token(self):
+        params = {'profile_id': self.profile_id}
+        heads = self.heads
+        url_path = self.re_url_path
+        tem = request(url_path=url_path, head=heads, params=params)
+        if tem.get('data') is not None:
+            _ = tem.get('data')
+            out = _[0].get('refresh_token')
+        else:
+            out = None
+        return out
+
+    def get_campaignid(self):
+        heads = self.heads
+        url_path = self.camid_url_path
+        tem = request(url_path=url_path, head=heads)
+        out = []
+        if tem.get('data') is not None:
+            _ = tem.get('data')
+            for i in _:
+                out.append(i['campaignId'])
+        return out
+
+    def get_spcampaign_data(self):
+        tem = SPClient(**self.AWS_CREDENTIALS)
+        camid_list = self.get_campaignid()
+
+        chunks = []
+        for i in range(0, len(camid_list), 100):
+            chunk = camid_list[i:i + 100]
+            chunks.append(chunk)
+
+
+        dict_tem = tem.get_spbudgetrecommendation(camid_list)
+        list_tem = dict_tem.get('budgetRecommendationsSuccessResults')
+
+        df_spbudgetrecommendation = pd.json_normalize(list(list_tem))
+        return df_spbudgetrecommendation
+
+    def dataconvert(self):
+        df = self.get_spcampaign_data()
+        df.drop(columns='index', inplace=True)
+        df['sevenDaysMissedOpportunities.startDate'] = pd.to_datetime(
+            df['sevenDaysMissedOpportunities.startDate'], format='%Y%m%d').dt.strftime(
+            '%Y-%m-%d')
+        df['sevenDaysMissedOpportunities.endDate'] = pd.to_datetime(
+            df['sevenDaysMissedOpportunities.endDate'], format='%Y%m%d').dt.strftime(
+            '%Y-%m-%d')
+        col = ['campaignId', 'suggestedBudget', 'budgetRuleRecommendation',
+               'sevenDaysMissedOpportunities.startDate',
+               'sevenDaysMissedOpportunities.endDate',
+               'sevenDaysMissedOpportunities.percentTimeInBudget',
+               'sevenDaysMissedOpportunities.estimatedMissedImpressionsLower',
+               'sevenDaysMissedOpportunities.estimatedMissedImpressionsUpper',
+               'sevenDaysMissedOpportunities.estimatedMissedClicksLower',
+               'sevenDaysMissedOpportunities.estimatedMissedClicksUpper',
+               'sevenDaysMissedOpportunities.estimatedMissedSalesLower',
+               'sevenDaysMissedOpportunities.estimatedMissedSalesUpper']
+        old = pd.DataFrame(data=[], columns=col)
+        tem = pd.concat([old, df]).reset_index()
+        tem.drop(columns='index', inplace=True)
+        tem.columns = [i.replace(".", "_") for i in tem.columns]
+        tem.rename(columns={
+            'campaignId': 'campaign',
+            'sevenDaysMissedOpportunities_startDate': 'startDate',
+            'sevenDaysMissedOpportunities_endDate': 'endDate',
+            'sevenDaysMissedOpportunities_percentTimeInBudget': 'percentTimeInBudget',
+            'sevenDaysMissedOpportunities_estimatedMissedImpressionsLower': 'MissedImpressionsLower',
+            'sevenDaysMissedOpportunities_estimatedMissedImpressionsUpper': 'MissedImpressionsUpper',
+            'sevenDaysMissedOpportunities_estimatedMissedClicksLower': 'MissedClicksLower',
+            'sevenDaysMissedOpportunities_estimatedMissedClicksUpper': 'MissedClicksUpper',
+            'sevenDaysMissedOpportunities_estimatedMissedSalesLower': 'MissedSalesLower',
+            'sevenDaysMissedOpportunities_estimatedMissedSalesUpper': 'MissedSalesUpper',
+        }, inplace=True)
+        json_data = json.loads(tem.to_json(orient='records', force_ascii=False))
+        return json_data
+
+    def updata_create(self):
+        body = self.dataconvert()
+        heads = self.heads
+        url_path = self.upcreate_url_path
+        tem = request(url_path=url_path, head=heads, body=body, method="POST")
+        return tem
+
+
+if __name__ == '__main__':
+    a = SpCampaign(profile_id="3006125408623189")
+    out = a.updata_create()
+    # out = a.dataconvert()
+    print(out)