1

app/api_data/cost_util.py

@@ -17,6 +17,69 @@ max_workers = 10
 count = []
 t = du.get_n_days(-10)
 
+
+def get_campaign(account_id, access_token, flag,campaign_ids,dt):
+    path = 'campaigns/get'
+    fields = ('campaign_id', 'campaign_name', 'configured_status', 'campaign_type', 'promoted_object_type',
+              'daily_budget','budget_reach_date','created_time','last_modified_time','speed_mode','is_deleted')
+    url = 'https://api.e.qq.com/v1.3/' + path
+    li = []
+    page = 1
+
+    while True:
+        parameters = {
+            'access_token': access_token,
+            'timestamp': int(time.time()),
+            'nonce': str(time.time()) + str(random.randint(0, 999999)),
+            'fields': fields,
+            "filtering": [{
+                "field": "campaign_id",
+                "operator": "IN",
+                "values":
+                    campaign_ids.split(',')
+            }],
+            "account_id": account_id,
+            "page": page,
+            "page_size": 100,
+            "is_deleted": False
+        }
+
+        for k in parameters:
+            if type(parameters[k]) is not str:
+                parameters[k] = json.dumps(parameters[k])
+
+        while True:
+            r = requests.get(url, params=parameters).json()
+            code = r['code']
+            if code == 11017:
+                time.sleep(61)
+            else:
+                break
+        # print(r)
+        total_page = r['data']['page_info']['total_page']
+
+        if page > total_page:
+            break
+        else:
+            page += 1
+
+        if r.get("data"):
+            for i in r['data']['list']:
+                li.append((str(i['campaign_id']), i['campaign_name'], i['configured_status'], i['campaign_type'],
+                           i['promoted_object_type'],i['daily_budget'],i.get('budget_reach_date'),
+                           DateUtils.stamp_to_str(i['created_time']),
+                           DateUtils.stamp_to_str(i['last_modified_time']),i.get('speed_mode'),i.get('is_deleted'),
+                           account_id, flag, dt))
+    # print(li)
+    """mp 没有 speed_mode，is_deleted，budget_reach_date"""
+    if li.__len__() > 0:
+        print(f"{account_id}有计划:", li.__len__())
+        sql = "replace into campaign_info values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
+        db.quchen_text.executeMany(sql, li)
+    db.close()
+
+
+
 def get_adcreatives(account_id,access_token,flag,adc_ids,dt):  # 获取创意
 
     url = 'https://api.e.qq.com/v1.1/adcreatives/get'

app/api_data/get_cost.py

@@ -99,6 +99,23 @@ def video(dt):
 		executor.submit(video_info_get, account[0], account[1], account[3])
 	executor.shutdown()
 
+def campaign(dt):
+	sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(campaign_id)  from ad_info  a
+	left join (
+	select account_id,access_token,'MP' type  from advertiser_vx where  (name !='' or name is not null) union 
+	select account_id,access_token,'GDT' type from advertiser_qq where  (name !='' or name is not null)
+	) b on a.account_id=b.account_id
+	where a.dt='{dt}'
+	GROUP BY b.account_id,b.access_token,b.type having b.account_id is not null """
+
+	accounts = db.quchen_text.getData(sql)
+
+	executor = ThreadPoolExecutor(max_workers=max_workers)
+	for account in accounts:
+		executor.submit(get_campaign, account[0], account[1], account[2], account[3],dt)
+	executor.shutdown()
+
+
 def run(dt):
 	"""
 		1.拉取有消耗的广告
@@ -142,7 +159,8 @@ if __name__ == '__main__':
 	# day()
 	# run('2021-05-10')
 	# adcreative('2021-05-11')
-	video('2021-05-14')
+	# video('2021-05-14')
+	campaign('2021-05-14')
 	# for dt in list(reversed(du.getDateLists('2020-04-08','2021-04-09'))):
 	# 	print(dt)
 	# 	ad(dt)

app/api_data/get_cost_older.py

@@ -0,0 +1,166 @@
+import sys
+import requests
+import hashlib
+import time
+import json
+import pymysql
+import logging
+import random
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
+from model.DateUtils import DateUtils
+from model.DataBaseUtils import MysqlUtils
+db = MysqlUtils()
+logging.getLogger().setLevel(logging.WARNING)
+du = DateUtils()
+
+def md5value(s):
+	md5 = hashlib.md5()
+	md5.update(s.encode("utf-8"))
+	return md5.hexdigest()
+
+
+def daily_reports_get(access_token, account_id, st, et,level, fields):
+    interface = 'daily_reports/get'
+    url = 'https://api.e.qq.com/v1.1/' + interface
+
+    common_parameters = {
+        'access_token': access_token,
+        'timestamp': int(time.time()),
+        'nonce': str(time.time()) + str(random.randint(0, 999999)),
+    }
+
+    parameters = {
+        "account_id": account_id,
+        "level": level,
+        "date_range":
+            {
+                "start_date": st,
+                "end_date": et
+            },
+        "page": 1,
+        "page_size": 1000,
+        "fields": fields
+    }
+    parameters.update(common_parameters)
+
+    for k in parameters:
+        if type(parameters[k]) is not str:
+            parameters[k] = json.dumps(parameters[k])
+
+    r = requests.get(url, params=parameters).json()
+    if r['code'] != 0:
+        print(r['code'], r['message'])
+
+    return r
+
+
+def get_q_data(y,li,st,et):
+    c = daily_reports_get(y[2], y[0],st,et, "REPORT_LEVEL_ADVERTISER", (
+        'date', 'view_count', 'valid_click_count', 'ctr', 'cpc', 'cost', 'web_order_count', 'web_order_rate',
+        'web_order_cost', 'follow_count', 'order_amount', 'order_roi', 'platform_page_view_count',
+        'web_commodity_page_view_count', 'from_follow_uv'))
+    # print(c)
+    if 'data' in c.keys() and len(c["data"]["list"]) > 0:
+        for d in c['data']['list']:
+            d['account_id'] = y[0]
+            print(d['account_id'], d["cost"])
+        x = tuple(d.values())
+        li.append(x)
+
+def get_v_data(y,li,st,et):
+    c = daily_reports_get(y[2], y[0],st,et, "REPORT_LEVEL_ADVERTISER_WECHAT", (
+        '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'))
+    # print(c)
+
+    if 'data' in c.keys() and len(c["data"]["list"])>0:
+        for d in c['data']['list']:
+            d['account_id'] = y[0]
+            print(d['account_id'], d["cost"])
+        x = tuple(d.values())
+        li.append(x)
+        # print(len(li))
+
+
+
+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"
+	a= db.quchen_text.getData(sql)
+	return a
+
+def get_qq_list():
+	sql = "select account_id,'',access_token,refresh_token,name," \
+		  "ifnull(stage,''),ifnull(pitcher,''),ifnull(platform,''),ifnull(book,'') from advertiser_qq"
+	a = db.quchen_text.getData(sql)
+	return a
+
+
+def get_daily_vx(st,et):
+	token_list_v = get_vx_list()
+	print("获取vx账号：",token_list_v.__len__())
+
+	time1 = time.time()
+	executor = ThreadPoolExecutor(max_workers=20)
+	li=[]
+	for y in token_list_v:
+		executor.submit(get_v_data,y,li,st,et)
+	executor.shutdown()
+	print('get_daily_vx:', len(li), 'cost:', int(time.time()-time1))
+
+	mysql_insert_daily_vx(li)
+
+def get_daily_qq(st,et):
+	token_list_q =  get_qq_list()
+	print("获取qq账号：",token_list_q.__len__())
+	time1 = time.time()
+	li=[]
+	executor = ThreadPoolExecutor(max_workers=10)
+	for x in token_list_q:
+
+		executor.submit(get_q_data,x,li,st,et)
+	executor.shutdown()
+	print('get_qq_order:', len(li), 'cost:', int(time.time()-time1))
+	mysql_insert_daily_qq(li)
+
+
+def mysql_insert_daily_vx(data):
+
+    b ="""replace into daily_vx (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 mysql_insert_daily_qq(data):
+    a = """replace into daily_qq (date,view_count,valid_click_count,ctr,cpc,cost,web_order_count,web_order_rate,
+	web_order_cost,follow_count,order_amount,order_roi,platform_page_view_count,web_commodity_page_view_count,
+	from_follow_uv,account_id) values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"""
+    db.quchen_text.executeMany(a,data)
+
+
+def run(st,et):
+	get_daily_vx(st,et)
+	get_daily_qq(st,et)
+
+def old_cost_hourly():
+    st = et =du.getNow()
+    run(st, et)
+
+def daily():
+    st = du.get_n_days(-10)
+    et = du.get_n_days(-1)
+    run(st, et)
+
+if __name__ == '__main__':
+
+    run()
+
+
+
+
+
+
+

app/api_data/get_order.py

@@ -207,4 +207,4 @@ if __name__ == '__main__':
     # yangguang()
     # huasheng("2021-04-12",'2021-05-10')
     # yueweng("2021-05-11", '2021-05-11')
-    hourly()
+    hourly()

app/api_data/platform_order/yuewen.py

@@ -53,7 +53,7 @@ def get_yuewen_order_task(st, et, account):
         params['sign'] = sign
         # print(params)
         response_result_json = requests.get(url=url, params=params).json()
-        # print(response_result_json)
+        print(response_result_json)
 
         # print(len(response_result_json["data"]["list"]))
 
@@ -101,4 +101,4 @@ def get_yuewen_order_task(st, et, account):
 
 if __name__ == '__main__':
     a = "mqud82950@163.com,74ca754515fa253c8ab790603cebc2ee"
-    get_yuewen_order_task('2021-05-06', '2021-05-06', a.split(','))
+    get_yuewen_order_task('2021-05-14', '2021-05-14', a.split(','))

app/hourlyRun.py

@@ -4,6 +4,7 @@ from app.etl import data_stat_run
 from model.DateUtils import DateUtils
 from app.api_data import get_cost
 from app.etl.dw import dw_image_cost_day
+from app import task
 du = DateUtils()
 import time
 from model.DingTalkUtils import DingTalkUtils
@@ -12,8 +13,8 @@ from model.DingTalkUtils import DingTalkUtils
 if __name__ == '__main__':
 
     st = time.time()
-    # 订单数据
-    get_order.hourly()
+    # 订单数据和消耗数据拉取
+    task.hourly()
 
     # 数据处理
     data_stat_run.hourly()
@@ -22,7 +23,7 @@ if __name__ == '__main__':
     get_cost.hourly()  # 广告相关消耗数据
     dw_image_cost_day.hourly()
 
-    DingTalkUtils.send(f"小时任务耗时{int(time.time()-st)}秒",phone="18860455786")
+    # DingTalkUtils.send(f"小时任务耗时{int(time.time()-st)}秒",phone="18860455786")
 
 
 

app/task.py

@@ -0,0 +1,35 @@
+
+from multiprocessing import Process
+from app.api_data.get_order import *
+from app.api_data.get_cost_older import old_cost_hourly
+def hourly():
+    p1 = Process(target=huasheng)
+    p2 = Process(target=qiyue)
+    p3 = Process(target=qiyueyousheng)
+    p4 = Process(target=wending)
+    p5 = Process(target=zhangdu)
+    p6 = Process(target=zhangzhongyun)
+    p7 = Process(target=yueweng)
+    p8 = Process(target=youshuge)
+    p9 = Process(target=yangguang)
+    p10 = Process(target=old_cost_hourly)
+    p1.start()
+    p1.join()
+    p2.start()
+    p2.join()
+    p3.start()
+    p3.join()
+    p4.start()
+    p4.join()
+    p5.start()
+    p5.join()
+    p6.start()
+    p6.join()
+    p7.start()
+    p7.join()
+    p8.start()
+    p8.join()
+    p9.start()
+    p9.join()
+    p10.start()
+    p10.join()

model/DateUtils.py

@@ -291,7 +291,7 @@ class DateUtils:
         r = self.now - timedelta(hours=n)
         return r if flag else r.strftime("%Y-%m-%d %H:00:00")
 
-    def get_n_minutes_ago(self, n=1, string=True):
+    def get_n_minutes_ago(self, n=1):
         """
         get n minutes ago
         default 1 minutes ago
@@ -302,10 +302,8 @@ class DateUtils:
         :param n:
         :return:
         """
-        if string:
-            return (self.now - timedelta(minutes=n)).strftime("%Y-%m-%d %H:%M")
-        else:
-            return self.now - timedelta(minutes=n)
+        return (self.now - timedelta(minutes=n)).strftime("%Y-%m-%d %H:%M:%S")
+
 
     def get_n_pre_month_last_day(self, n=0, flag=False):
         """