# -*- codeing = utf-8 -*- # @Time : 2023/4/11 19:08 # @Author : Clown # @File : demo_feiyu_ad_tk_refresh.py # @Software : PyCharm import pymysql import json import requests from datetime import datetime,timedelta import time def linkTomySql(host, passwd, db_name, port): '''连接至数据库返回【db】,v2新增local_infile=1 打开文件导入权限''' try: # 本地连接为：localhost 服务器连接为：124.222.188.59 db = pymysql.connect ( host=host, user="root", passwd=passwd, db=db_name, port=port, charset='utf8mb4', local_infile=1) print ('\nconnect to mysql server 成功') print ('---------------------------------------') except: print ("\ncould not connect to mysql server") db = "连接失败" return db def save_oauth_to_db(db,rsp_data,brandName): lastCatchTime = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor = db.cursor() sql = '''INSERT INTO feiyu_ad_tk_refresh VALUES (%s,%s,%s,%s);''' values = (lastCatchTime,brandName,rsp_data,0) cursor.execute(sql,values) db.commit() cursor.close() def select_secret_by_app_id(db,app_id): cursor = db.cursor () sql = f'''SELECT * FROM feiyu_ad_app_id_sct WHERE app_id = {app_id};''' cursor.execute (sql) out = cursor.fetchall ()[0] secret =out[2] brandName = out[1] cursor.close () return secret,brandName def get_access_token_by_auth_code(app_id,auth_code): """ auth_code换取access-token详见https://ad.oceanengine.com/openapi/doc/index.html?id=1696710505596940 :param auth_code: :return: """ host = '124.222.188.59' passwd = '111...Clown' db_name = 'zuzu_data' port = 63306 db = linkTomySql (host, passwd, db_name, port) secret,brandName = select_secret_by_app_id (db, app_id) open_api_url_prefix = "https://ad.oceanengine.com/open_api/" uri = "oauth2/access_token/" url = open_api_url_prefix + uri headers = {'Content-Type':'application/json'} data = { "app_id": app_id, "secret": secret, "grant_type": "auth_code", "auth_code": auth_code } rsp = requests.post(url, json=data, headers=headers) rsp_data = rsp.text save_oauth_to_db(db,rsp_data,brandName) # 将access_token等信息进行存储 db.close () return # 选择所有有效账户 def selectAllUsers(db): cursor = db.cursor () sql = f'''SELECT * FROM feiyu_ad_app_id_sct WHERE state = 1;''' cursor.execute (sql) out = cursor.fetchall () list_out = [] for row in out: dict_out = {'brandName': row[1], 'appid': row[0], 'secret':row[2], 'userName':row[3]} list_out.append (dict_out) cursor.close () return list_out # 选择品牌下的token信息 def selectAccessToken(db,brandName): cursor = db.cursor () sql = f'''SELECT * FROM feiyu_ad_tk_refresh WHERE brandName = '{brandName}' ORDER BY lastCatchTime DESC;''' cursor.execute (sql) out = cursor.fetchall ()[0] json_data = out[2] cursor.close () return json_data # 查询已授权账户信息 def getAdvertiser_ids(json_data): open_api_url_prefix = "https://ad.oceanengine.com/open_api/" uri = "oauth2/advertiser/get/" url = open_api_url_prefix + uri data = { "access_token": json_data['access_token'] } headers = {'Content-Type':'application/json'} rsp = requests.get (url, json=data, headers = headers).json() print(rsp) advertiser_data = rsp['data'] advertiser_ids = [] for info in rsp['data']['list']: advertiser_id = info['advertiser_id'] advertiser_ids.append(advertiser_id) return advertiser_data,advertiser_ids # 独立刷新token def refresh_access_token(db,brandName,appid,secret): json_data = selectAccessToken(db,brandName) json_data = json.loads(json_data)['data'] open_api_url_prefix = "https://ad.oceanengine.com/open_api/" uri = "oauth2/refresh_token/" refresh_token_url = open_api_url_prefix + uri data = { "appid": appid, "secret": secret, "grant_type": "refresh_token", "refresh_token": json_data['refresh_token'], } rsp = requests.post(refresh_token_url, json=data).json() save_oauth_to_db (db, json.dumps (rsp, ensure_ascii=False), brandName) json_data = selectAccessToken (db, brandName) json_data = json.loads (json_data)['data'] advertiser_data, advertiser_ids = getAdvertiser_ids (json_data) rsp['data']['advertiser_ids'] = advertiser_ids rsp['advertiser_data'] = advertiser_data time.sleep(2) save_oauth_to_db(db,json.dumps(rsp,ensure_ascii=False),brandName) # 批量更新密钥 def refreshAllFeiYuUsersToken(db): Users = selectAllUsers (db) for userInfo in Users: try: refresh_access_token (db, userInfo['brandName'], userInfo['appid'], userInfo['secret']) except Exception as e: print(e) print(f'【{userInfo["brandName"]}】token 更新失败 ') # 获取资源账户id def get_advertiser(advertiser_id,accessToken): out_list = [] open_api_url_prefix = "https://ad.toutiao.com/open_api/" uri = "2/majordomo/advertiser/select/" url = open_api_url_prefix + uri params = { "advertiser_id": advertiser_id } headers = {"Access-Token": accessToken} try: rsp = requests.get (url, json=params, headers=headers).json () rsp_list = rsp['data']['list'] for rsp_0 in rsp_list: advertiser_id = rsp_0['advertiser_id'] out_list.append(advertiser_id) except: rsp = {'message':'error'} return rsp,out_list # 获取广告主们的详细信息 def get_advertiser_info(userName,advertisers_id,accessToken): open_api_url_prefix = "https://ad.oceanengine.com/open_api/" uri = "2/advertiser/info/" url = open_api_url_prefix + uri params = { "advertiser_ids": advertisers_id, "fields": ["id", "name", "note"] } headers = {"Access-Token": accessToken} rsp = requests.get (url, json=params, headers=headers) rsp_data = rsp.json () # print(rsp_data) out_list = [] brand_list = ['浆小白','粥小鲜','楼兰说'] brand_dict = {'浆小白':{'brandName':'浆小白','list':[]}, '粥小鲜':{'brandName':'粥小鲜','list':[]}, '楼兰说':{'brandName':'楼兰说','list':[]}} for advertiser_info in rsp_data['data']: try: brand_dict[advertiser_info['note']]['list'].append(advertiser_info['id']) except: ... for brand in brand_list: brandName = brand_dict[brand]['brandName'] out_list_0 = brand_dict[brand]['list'] if len(out_list_0) > 0: dict_out = {'userName': userName, 'accessToken': accessToken, 'brandName': brandName, 'advertiser_ids': out_list_0, 'routeName': 'feiyu'} out_list.append(dict_out) # print(out_list) return out_list # 获取所有飞鱼投流账户信息 def getAllFeiYuUsersList(db): Users = selectAllUsers (db) out_list = [] for userInfo in Users: userName = userInfo['userName'] brandName = userInfo['brandName'] json_data = selectAccessToken (db, brandName) json_data = json.loads (json_data)['data'] accessToken = json_data['access_token'] advertisers_id = json_data['advertiser_ids'] for advertiser_id in advertisers_id: rsp,out_list_0 = get_advertiser (advertiser_id, accessToken) #获取总帐号下的分账号 if brandName == '综合': out_list_z = get_advertiser_info (userName, out_list_0, accessToken) out_list = out_list + out_list_z else: if rsp['message'] == 'OK' and len(out_list_0)>0: dict_out = {'userName':userName, 'accessToken':accessToken, 'brandName':brandName, 'advertiser_ids':out_list_0, 'routeName':'feiyu'} out_list.append(dict_out) return out_list if __name__ == '__main__': host = '124.222.188.59' passwd = '111...Clown' db_name = 'zuzu_data' port = 63306 db = linkTomySql (host, passwd, db_name, port) advertiser_ids = ['1762396620978184', '1762396621515784', '1762396622070791'] accessToken = 'c06ac76994e55d4951223e38d174b2591efb7065' out_list = getAllFeiYuUsersList(db) print(out_list) db.close () if 1 == 0: open_api_url_prefix = "https://ad.oceanengine.com/open_api/" uri = "2/majordomo/advertiser/select/" url = open_api_url_prefix + uri params = { "advertiser_id": 1668398135393287 } headers = {"Access-Token": "f1c777bd288b700ee9fa25cac83c941277f061c1"} rsp = requests.get (url, json=params, headers=headers) rsp_data = rsp.json() print(json.dumps(rsp_data,ensure_ascii=False))