删除 'sync_amz_data/public/adstoken_OAUTH.py'

sync_amz_data/public/adstoken_OAUTH.py

@@ -1,197 +0,0 @@
-import json
-import requests
-import sys
-import time
-import os
-# from myapp.temp_get_budget_black_five.info import *
-
-#"C:\Users\ansjer\Documents\WXWork\1688858253398958\Cache\File\2023-08\autobid\autobid\info\token_info.json"
-class Ads_Api_Token(object):
-
-    def __init__(self, authorization_code='-1', profileId='-1', shop_name='-1') -> None:
-        sub_dir = os.path.abspath(os.path.dirname(__file__))
-        self.return_url = 'https://www.amazon.com/'
-        self.save_json_file_name = r"./token_info.json"# r"{sub_dir}\info\token_info.json".format(sub_dir=sub_dir)
-        self.lwa_info_file_name = r"./lwa_info.json"# r"{sub_dir}\info\lwa_info.json".format(sub_dir=sub_dir)
-        self.headers = {'Content-Type': 'application/x-www-form-urlencoded'}
-        self.authorization_url = 'https://api.amazon.com/auth/o2/token'
-        self.authorization_code = authorization_code
-
-        # self.region = 'NA'
-        lwa_info = self._read_lwa_info()
-        self.lwa_client_id = lwa_info['lwa_client_id']
-        self.lwa_client_secret = lwa_info['lwa_client_secret']
-        self.access_token = self._get_access_token()
-
-        # 检查错误
-        if shop_name == '-1' or profileId == '-1':
-            print('错误：【shop_name或profileId没有初始化】')
-            print('作为广告商，你的客户列表：')
-            self._get_profile()
-            print('\n确认客户区域、名字等,选择name和profileId重新初始化shop_name和profileId')
-            sys.exit()
-        # print(self._get_profileId(name=shop_name),type(self._get_profileId(name=shop_name)))
-        if self._get_profileId(name=shop_name) != profileId:
-            print('错误：【检查profileId与shop_name不匹配】')
-            print('作为广告商，你的客户列表：')
-            self._get_profile()
-
-            print('\n确认客户区域、名字等,选择name和profileId重新初始化shop_name和profileId')
-            sys.exit()
-
-        self.shop_name = shop_name
-        self.profileId = profileId
-
-    def _read_lwa_info(self) -> str:
-        with open(self.lwa_info_file_name, 'r') as file_obj:
-            lwa_info = json.load(file_obj)
-        file_obj.close()
-        return lwa_info
-
-    def _read_token(self) -> str:
-        with open(self.save_json_file_name, 'r') as file_obj:
-            token = json.load(file_obj)
-        file_obj.close()
-        return token
-
-    def _write_token(self, token: dict) -> None:
-        """转换过期时间到秒级时间戳写入token
-
-        Args:
-            token (dict): token
-        """
-        token['expires_in'] += int(time.time())
-        with open(self.save_json_file_name, 'w') as file_obj:
-            json.dump(token, file_obj)
-
-    def _check_token_json(self, token: dict) -> bool:
-        """检查token合法性
-
-        Args:
-            token (dict): token
-
-        Returns:
-            bool: 是否合法
-        """
-        if 'access_token' in token and 'refresh_token' in token and 'expires_in' in token:
-            return True
-        else:
-            self._print_authorization_grant_info(error_type=1)
-
-    def _print_authorization_grant_info(self, error_type='-1'):
-        """打印授权的授权信息,获取授权code
-
-        Args:
-            error_type (str, optional): 0为找不到文件，1为文件内容不通过. Defaults to '-1'.
-        """
-        url = 'https://www.amazon.com/ap/oa?client_id=%s&scope=advertising::campaign_management&response_type=code&redirect_uri=%s' % (
-        self.lwa_client_id, self.return_url)
-        if error_type == 0:
-            print('==' * 10 + '未检测' + self.save_json_file_name + '==' * 10)
-        if error_type == 1:
-            print('==' * 10 + '内容不通过' + self.save_json_file_name + '==' * 10)
-        print('\t\t1.打开登陆地址' + url)
-        print('\t\t2.点击允许并登陆zhangside账号')
-        print('\t\t3.将重定向的url中的code保存下来')
-        print('\t\t4.将code填入初始化Token的authorization_code')
-        print('==' * 30)
-        sys.exit()
-
-    def _first_get_token(self, authorization_code: str) -> str:
-        """从授权的授权code申请token
-
-        Args:
-            authorization_code (str): 授权的授权code
-
-        Returns:
-            str: access_token
-        """
-        data = 'grant_type=authorization_code&code=%s&redirect_uri=%s&client_id=%s&client_secret=%s' % (
-        authorization_code, self.return_url, self.lwa_client_id, self.lwa_client_secret)
-        res = requests.post(self.authorization_url, data=data, headers=self.headers)
-        res = res.json()
-        self._check_token_json(res)
-        self._write_token(res)
-        return res['access_token']
-
-    def _refresh_token(self, limit_time_minute=3) -> str:
-        """刷新token，如果在授权时间还有多余limit_time_minute分钟不刷新直接返回access_token
-
-        Args:
-            limit_time_minute (int, optional): 剩余时间分钟. Defaults to 15.
-
-        Returns:
-            str: access_token
-        """
-        token = self._read_token()
-        self._check_token_json(token)
-        token_timestamp = token['expires_in']
-        if token_timestamp - int(time.time()) > limit_time_minute * 60:
-            return token['access_token']
-        data = 'grant_type=refresh_token&client_id=%s&refresh_token=%s&client_secret=%s' % (
-        self.lwa_client_id, token['refresh_token'], self.lwa_client_secret)
-        res = requests.post(self.authorization_url, data=data, headers=self.headers)
-        res = res.json()
-        self._check_token_json(res)
-        self._write_token(res)
-        return res['access_token']
-
-    def _get_access_token(self) -> str:
-        """根据状态获取token
-
-        Returns:
-            str: access_token
-        """
-        # 检查状态
-        if self.authorization_code == '-1':
-            if not os.path.exists(self.save_json_file_name):
-                self._print_authorization_grant_info(error_type=0)
-            token = self._read_token()
-            self._check_token_json(token)
-            access_token = self._refresh_token()
-            return access_token
-        else:
-            print('填入了code，获取授权的授权,下一次可不用填入code')
-            access_token = self._first_get_token(self.authorization_code)
-            return access_token
-
-    def _get_profileId(self, name):
-        profiles_url = 'https://advertising-api.amazon.com/v2/profiles'
-        data = {
-            'Amazon-Advertising-API-ClientId': self.lwa_client_id,
-            'Authorization': 'Bearer ' + self.access_token
-        }
-        res = requests.get(profiles_url, headers=data)
-        res = res.json()
-        profileId = '-1'
-        for customer_info in res:
-            if customer_info['accountInfo']['name'] == name:
-                profileId = customer_info['profileId']
-        return str(profileId)
-
-    def _get_profile(self):
-        profiles_url = 'https://advertising-api.amazon.com/v2/profiles'
-        data = {
-            'Amazon-Advertising-API-ClientId': self.lwa_client_id,
-            'Authorization': 'Bearer ' + self.access_token
-        }
-        res = requests.get(profiles_url, headers=data)
-        res = res.json()
-        print(res)
-
-
-if __name__ == "__main__":
-    # authorization_code = 'ANzRQnVlaRziQlNxrHGG'
-    access_token = Ads_Api_Token(shop_name = 'ZosiDirect',profileId='3006125408623189')
-    tken = access_token._get_access_token()
-    print(type(tken),tken)
-# {'profileId': 420999963953676,
-#  'countryCode': 'BR',
-#  'currencyCode': 'BRL',
-#  'dailyBudget': 999999999.0,
-#  'timezone': 'America/Sao_Paulo',
-#  'accountInfo': {'marketplaceStringId': 'A2Q3Y263D00KWC',
-#                  'id': 'A252W7I0ACJHS1',
-#                  'type': 'seller',
-#                  'name': 'ZosiDirect',
-#                  'validPaymentMethod': False}}