MOD:添加头条数据采集

app/api_data/tx_ad_cost/get_cost_older.py

@@ -8,6 +8,8 @@ from concurrent.futures import ThreadPoolExecutor
 from model.DateUtils import DateUtils
 from model.DataBaseUtils import MysqlUtils
 from model.DingTalkUtils import DingTalkUtils
+from six import string_types
+from six.moves.urllib.parse import urlencode, urlunparse
 
 db = MysqlUtils()
 du = DateUtils()
@@ -48,7 +50,7 @@ def daily_reports_get(access_token, account_id, st, et, level, fields, err_num=0
         if type(parameters[k]) is not str:
             parameters[k] = json.dumps(parameters[k])
 
-    r = requests.get(url, params=parameters,timeout=5).json()
+    r = requests.get(url, params=parameters, timeout=5).json()
     logging.info('account_id: {} 开始获取消耗数据'.format(account_id))
     if r['code'] != 0:
         logging.warning(
@@ -89,6 +91,60 @@ def get_v_data(y, li, st, et):
             li.append(x)
 
 
+def get_tt_data(account_info, li, st, et):
+    def build_url_ad(path, query=""):
+        # type: (str, str) -> str
+        """
+        Build request URL
+        :param path: Request path
+        :param query: Querystring
+        :return: Request URL
+        """
+        scheme, netloc = "https", "ad.oceanengine.com"
+        return urlunparse((scheme, netloc, path, "", query, ""))
+
+    page_num = 1
+    advertiser_ids = account_info[1]
+    for advertiser_id in advertiser_ids:
+        while True:
+            # account_info
+            my_args = {
+                "start_date": st,
+                "end_date": et,
+                "page_size": 100,
+                "page": page_num,
+                # "agent_id" : "1708974248093789",
+                'advertiser_id': advertiser_id,
+                # "start_date": "%s"
+            }
+            PATH = "/open_api/2/report/advertiser/get/"
+            args = json.loads(json.dumps(my_args))
+            query_string = urlencode({k: v if isinstance(v, string_types) else json.dumps(v) for k, v in args.items()})
+            url = build_url_ad(PATH, query_string)
+            headers = {
+                "Access-Token": account_info[0],
+            }
+            rsp = requests.get(url, headers=headers)
+            '''
+            date,cost,view_count,valid_click_count,
+        ctr,official_account_follow_rate,order_amount,
+        order_roi,order_count,order_rate,order_unit_price,
+        web_order_cost,first_day_order_amount,first_day_order_count,account_id
+            '''
+            result = rsp.json()
+            for _ in result['data']['list']:
+                campaign_info = (_['stat_datetime'][:10], _['cost'] * 100, _['show'], _['click'],
+                                 _['ctr'], None, None,
+                                 None, None, None, None,
+                                 None, None, None, advertiser_id)
+                li.append(campaign_info)
+            total_page = result['data']['page_info']['total_page']
+            if page_num > total_page or page_num == total_page:
+                break
+            else:
+                page_num = page_num + 1
+
+
 def get_vx_list():
     sql = "select account_id,wechat_account_id,access_token,refresh_token,name," \
           "ifnull(stage,''),ifnull(pitcher,''),ifnull(platform,''),ifnull(book,'') from advertiser_vx"
@@ -117,6 +173,16 @@ def mysql_insert_daily_qq(data):
     db.quchen_text.executeMany(a, data)
 
 
+def mysql_insert_daily_tt(data):
+    b = """replace into daily_tt (date,cost,view_count,valid_click_count,ctr,
+    official_account_follow_rate,order_amount,
+	order_roi,order_count,order_rate,order_unit_price,
+	web_order_cost,first_day_order_amount,
+	first_day_order_count,account_id)
+	 values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"""
+    db.quchen_text.executeMany(b, data)
+
+
 def get_daily_vx(st, et):
     token_list_v = get_vx_list()
     logging.info("获取vx账号：" + str(token_list_v.__len__()))
@@ -145,6 +211,57 @@ def get_daily_qq(st, et):
     mysql_insert_daily_qq(li)
 
 
+def get_daily_tt(st, et):
+    def refresh_access_token(appid, secret, refresh_token):
+        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": refresh_token,
+        }
+        rsp = requests.post(refresh_token_url, json=data)
+        # print(rsp.text)
+        rsp_data = rsp.json()
+        new_refresh_token = rsp_data['data']['refresh_token']
+        sql = f'''
+        update bytedance_login_info
+        set refresh_token='{new_refresh_token}'
+        where appid='{appid}' 
+        '''
+        db.quchen_text.execute(sql)
+        return rsp_data['data']['access_token']
+
+    # 1.获取refresh_token
+    sql = '''
+    select appid,secret,refresh_token from bytedance_login_info
+    '''
+    accounts_info = db.quchen_text.getData(sql)
+
+    # 2.刷新refresh_token,并获取最新的access_token
+    for account_info in accounts_info:
+
+        appid, secret, refresh_token = account_info
+        access_token = refresh_access_token(appid, secret, refresh_token)
+        # 3.获取agent_id
+        sql = f'''
+        select distinct(advertiser_id) from bytedance_pitcher_change
+        where appid='{appid}'
+        '''
+        advertiser_ids = db.quchen_text.getData(sql)
+        logging.info("获取头条账号：" + str(advertiser_ids.__len__()))
+        advertiser_ids = [_[0] for _ in advertiser_ids]
+        # token,adv_ids
+        account_info = (access_token, advertiser_ids)
+        time1 = time.time()
+        li = []
+        get_tt_data(account_info, li, st, et)
+        logging.info('get_tt_order:' + str(len(li)) + 'cost:' + str(int(time.time() - time1)))
+        mysql_insert_daily_tt(li)
+
+
 def run(st, et):
     logging.info('微信消耗数据拉取,开始')
     get_daily_vx(st, et)
@@ -152,7 +269,9 @@ def run(st, et):
     logging.info('qq消耗数据拉取,开始')
     get_daily_qq(st, et)
     logging.info('qq消耗数据拉取,结束')
-
+    logging.info('头条消耗数据拉取,开始')
+    get_daily_tt(st, et)
+    logging.info('头条消耗数据拉取,结束')
 
 def old_cost_hourly():
     st = et = du.getNow()
@@ -170,7 +289,7 @@ def old_cost_daily():
 if __name__ == '__main__':
     # run()
     # old_cost_daily()
-    st = du.get_n_days(-6)
-    et = du.get_n_days(-1)
+    st = du.get_n_days(-30)
+    et = du.get_n_days(0)
     print(st, et)
-    run(st, et)
+    get_daily_tt(st, et)