1

app/api_data/platform_order/yangguang.py

@@ -1,13 +1,16 @@
 from app.api_data.platform_order.order_util import *
 from model.DataBaseUtils import MysqlUtils
 from model.DingTalkUtils import DingTalkDecorators,DingTalkUtils
-
+from model.DateUtils import DateUtils
 db = MysqlUtils()
+du = DateUtils()
 
 
 
 @DingTalkDecorators("阳光")
 def yangguang(start=None, end=None):
+    accounts = get_account("阳光")
+
     if start:
         start = start + ' 00:00:00'
         end = end + ' 23:59:59'
@@ -17,7 +20,7 @@ def yangguang(start=None, end=None):
 
     client_id = 10008097
     token = '2xa1d55tTPBjeEA8Ho'
-    accounts = get_account("阳光")
+
     if accounts.__len__() == 0:
         return
     else:
@@ -27,25 +30,51 @@ def yangguang(start=None, end=None):
         stage = i[0]
         vip_id = i[1]
         print(vip_id)
-        get_yg_vip_channel(stage, vip_id, client_id, token)
+        # get_yg_vip_channel(stage, vip_id, client_id, token)
         get_yg_data(stage, vip_id, client_id, token, start, end)
+
+    print(check())
+    parse_data()
+
+
+def get_channel_info():
+    accounts = get_account("阳光")
+    client_id = 10008097
+    token = '2xa1d55tTPBjeEA8Ho'
+    for i in accounts:
+        stage = i[0]
+        vip_id = i[1]
+        print(vip_id)
+        get_yg_vip_channel(stage, vip_id, client_id, token)
+
+
+def parse_data():
+    print(111)
+    accounts = get_account("阳光")
+
+    for i in accounts:
+        # print(i)
+        vip_id = i[1]
+        stage = i[0]
+        parse_yg_data(vip_id, stage)
+
+
+def check():
+
     x = 1
     while True:
-        a = db.quchen_text.getOne("select count(1) from yangguang_path where update_time is null")
-        print(f" vip 待处理数量 {a} 正在等待数据回调")
+        a = db.quchen_text.getOne("select count(1) from yangguang_path where type ='order' and path is null")
+        print(f" 回调接口 待处理数量 {a} ")
         if a == 0:
+            info = '回调完成'
             break
         time.sleep(60)
         x += 1
         if x > 10:
             DingTalkUtils.send('阳光订单回调延时10min', '18860455786')
+            info = '回调未完成'
             break
-
-    for i in accounts:
-        vip_id = i[1]
-        stage = i[0]
-        parse_yg_data(vip_id, stage)
-
+    return info
 
 
 def get_account(plactform, id=None):
@@ -62,4 +91,14 @@ def daily_yg():
     yangguang(st, et)
 
 if __name__ == '__main__':
-    yangguang()
+
+    # get_channel_info()
+    # exit(0)
+    # yangguang('2020-09-20','2020-10-01')
+    # exit(0)
+
+    for i in du.split_date2('2020-06-28','2020-11-03',30):
+        print(i)
+        yangguang(i[0], i[1])
+
+

app/api_data/platform_order/yuewen.py

@@ -21,7 +21,7 @@ def get_yuewen_order_task(st, et, account):
         end_time = DateUtils.str_to_stamp(et) + 86399
 
 
-    for i in split_int(start_time,end_time,3600):
+    for i in split_int(start_time,end_time,10000):
         start = i[0]
         end = i[1]
         print(ut.stamp_to_str(end))

app/crontab_task/task.py

@@ -3,6 +3,7 @@ import threading
 from app.api_data.platform_order.get_order import *
 from app.api_data.tx_ad_cost.get_cost_older import old_cost_hourly,old_cost_daily
 from app.etl.sync_to_ck_task import order_sync_ck
+from app.api_data.platform_order import yangguang
 
 def hourly():
     t1 = threading.Thread(target=huasheng)
@@ -41,6 +42,7 @@ def daily():
     st = du.get_n_days(-10)
     et = du.get_n_days(-1)
 
+    yangguang.get_channel_info()
     t1 = threading.Thread(target=huasheng,args=(st,et))
     t2 = threading.Thread(target=qiyue,args=(st,et))
     t3 = threading.Thread(target=qiyueyousheng,args=(st,et))

app/etl/UserTags/UserTagsClean.py

@@ -1,36 +1,64 @@
 """用户标签清洗"""
 from model.DataBaseUtils import CkUtils,MysqlUtils
 import pandas as pd
+from model.DateUtils import DateUtils
 import sys
 ck = CkUtils()
 db = MysqlUtils()
+du = DateUtils()
+import time
+def run_all(y=0):
+    x =100000
 
-def run():
-    sql = """
-        select channel,
-            user_id open_id,
-            platform platform_book_name,
-                count(1) recharge_count,
-               sum(amount) recharge_money,
-               max(order_time) last_recharge_time
+    while True:
+        print((y+1)*x)
 
-        from order where status=2 
-        group by channel,user_id,platform limit 100"""
-    col = ['channel','open_id','platform_book_name','recharge_count','recharge_money','last_recharge_time']
+        sql = f"""
+            select channel,
+                user_id open_id,
+                platform platform_book_name,
+                    count(1) recharge_count,
+                   sum(amount) recharge_money,
+                   max(order_time) last_recharge_time
+    
+            from order where status=2 
+            group by channel,user_id,platform limit {y*x},{(y+1)*x}"""
+        col = ['channel','open_id','platform_book_name','recharge_count','recharge_money','last_recharge_time']
+        df = ck.getData_pd(sql,col=col)
+        if df.empty:
+            break
+        # print(df)
+        df_channel = get_qc_channel_info()
+        result = pd.merge(df,df_channel,how='left',on='channel').fillna(0)
+        save_data(result)
+        y += 1
 
-    df = ck.getData_pd(sql,col=col)
-    print(df)
+def run_today():
+    sql = f"""
+           select channel,
+                    user_id open_id,
+                    platform platform_book_name,
+                        count(1) recharge_count,
+                       sum(amount) recharge_money,
+                       max(order_time) last_recharge_time
 
+                from order where status=2 and user_id in (select distinct user_id from order where date='{du.getNow()}')
+                group by channel,user_id,platform"""
+    col = ['channel', 'open_id', 'platform_book_name', 'recharge_count', 'recharge_money', 'last_recharge_time']
+    df = ck.getData_pd(sql, col=col)
+    # print(df)
     df_channel = get_qc_channel_info()
+    result = pd.merge(df, df_channel, how='left', on='channel')
+    save_data(result)
 
-    result = pd.merge(df,df_channel,how='left',on='channel')
-    print(result)
 
+def save_data(df):
     table = 't_mp_fans'
-    keys=['mp_id','open_id']
-    tags=['recharge_count','recharge_money','last_recharge_time','platform_book_name']
+    keys = ['mp_id', 'open_id']
+    tags = ['recharge_count', 'recharge_money', 'last_recharge_time', 'platform_book_name']
+    db.zx.dfUpdate2mysql(df, table, keys, tags)
+
 
-    db.zx_test.dfUpdate2mysql(result,table,keys,tags)
 
 
 def get_qc_channel_info():
@@ -42,5 +70,6 @@ def get_qc_channel_info():
 
 
 if __name__ == '__main__':
-    run()
-    # get_qc_channel_info()
+    # run_today()
+    # get_qc_channel_info()
+    run_all(10)

model/DataBaseOperation.py

@@ -448,7 +448,7 @@ class MysqlOperation:
             keys=key,
             tags=tag,
             tag_values=df[tag],
-            key_values=df[key]
+            key_values=df[key],split=1000
         )
 
     def insertorupdatemany_v3(self, df, table, keys, tags, flag=False, split=80):