七悦优化

app/api_data/platform_order/QiYue.py

@@ -76,7 +76,6 @@ def get_qiyue_order_task(start, end, account):
                     x['book_name'],
                     x['id'],
                     x['state'],
-                    x['id'],
                     x['user_id'],
                     x['wechat_app_id'],
                     x['book_keywords'],
@@ -98,20 +97,22 @@ def get_qiyue_order_task(start, end, account):
 
 
 # 获取七悦用户信息
-def get_qiyue_user_info(account):
+def get_qiyue_user_info(account,date=None):
     order_url = "https://api.zhangwenwenhua.com" + "/v1/users"
     stage = account[0]
     token = account[1]
     page = 1
     size = 100
     freq = 0
+    dt = "&date=" + date if date else ''
+
     li = []
     while True:
         timestamp = int(time.time())
         url = order_url + "?" + "token=" + str(token) + "&timestamp=" + str(timestamp) + "&page=" + str(
-            page) + "&size=" + str(size)
+            page) + "&size=" + str(size) +"&date=" + dt
         r = requests.get(url=url).json()
-        # print(r)
+        print(r)
 
         freq += 1
         if freq == 59:
@@ -150,9 +151,9 @@ def get_qiyue_user_info(account):
 
 
 if __name__ == '__main__':
-    st ='2020-11-01'
-    et ='2021-01-01'
-    db.quchen_text.execute(f'DELETE from ods_order where platform="七悦" and date>="{st}" and date<="{et}"')
+    st ='2021-06-07'
+    et ='2021-06-07'
+
     for account in get_account("七悦"):
         get_qiyue_order_task(st, et, account)
         # get_qiyue_user_info(account)

app/api_data/platform_order/order_util.py

@@ -234,9 +234,9 @@ def save_order2(order_list):
 
 
     db.quchen_text.executeMany("""replace into ods_order(date,stage,platform,channel,channel_id,user_id,
-                               order_time,reg_time,amount,from_novel,order_id,status,platform_order_id,
+                               order_time,reg_time,amount,from_novel,order_id,status,
                                platform_user_id,wechat_app_id,book_tags,order_type,trade_no,transaction_no) 
-                               values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", order_list)
+                               values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", order_list)
     print("入库成功")
 
 def get_wd_account_siteid_list(account):

app/etl/UserTags/UserTagsClean.py

@@ -7,69 +7,51 @@ ck = CkUtils()
 db = MysqlUtils()
 du = DateUtils()
 import time
-def run_all(y=0):
-    x =100000
-
-    while True:
-        print((y+1)*x)
-
-        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
-
-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)
-
 
 def save_data(df):
     table = 't_mp_fans'
     keys = ['mp_id', 'open_id']
-    tags = ['recharge_count', 'recharge_money', 'last_recharge_time', 'platform_book_name']
+    tags = ['recharge_count', 'recharge_money', 'last_recharge_time',
+            'platform_book_name', 'platform_book_user_id', 'members', 'book_coin','channel_id']
     db.zx.dfUpdate2mysql(df, table, keys, tags)
 
 
 
+def QiyueUserInfo():
+    # 书城用户信息
+    sql1 = """select 
+                        user_id platform_book_user_id,
+                        open_id,
+                        channel_id,
+                        wechat_app_id ,
+                        coin+free_coin book_coin,
+                        if(vip_endtime=0,0,1) members  from platform_user_info where platform='七悦'"""
+
+    df1 = db.quchen_text.getData_pd(sql1)
+
+    # 用户订单信息
+    sql2 = """select channel_id,
+                  user_id open_id,
+                  count(1) recharge_count,
+                  sum(amount) recharge_money,
+                  max(order_time) last_recharge_time
+                  from order where status=2 and platform='七悦'
+                  group by channel_id,user_id"""
+
+    df2 = ck.getData_pdv2(sql2)
 
-def get_qc_channel_info():
-    sql ="""select cast(id as char) mp_id,nick_name channel from t_mp_account"""
-    df = db.zx.getData_pd(sql)
-    return df
-    # print(df)
+    # 查询公众号在本地的id
+    sql3 = """select cast(id as char) mp_id,appid wechat_app_id  from t_mp_account"""
+    df3 = db.zx.getData_pd(sql3)
 
+    df = df1.merge(df2,how='left',on=['channel_id','open_id']).merge(df3,how='left',on='wechat_app_id')
+    df['platform_book_name'] = '七悦'
+    print(df)
+    save_data(df)
 
 
 if __name__ == '__main__':
     # run_today()
     # get_qc_channel_info()
-    run_all(10)
+    # run_all(19)
+    QiyueUserInfo()

app/etl/sync_to_ck_task.py

@@ -12,10 +12,9 @@ dt = DateUtils()
 def order_sync_ck(ymd):
     log.debug("sync order")
     col = """date,stage,platform,channel,channel_id,user_id,order_time,reg_time,amount,from_novel,order_id,status,
-    platform_order_id,platform_user_id,wechat_app_id,book_tags,order_type"""
+    platform_user_id,wechat_app_id,book_tags,order_type"""
     sql =f"""select  DATE_FORMAT(date,"%Y-%m-%d") date,
             stage,platform,channel,channel_id,user_id,order_time,reg_time,amount,from_novel,order_id,status,
-    ifnull(platform_order_id,''),
     ifnull(platform_user_id,''),
     ifnull(wechat_app_id,''),
     ifnull(book_tags,''),

model/DataBaseUtils.py

@@ -9,6 +9,7 @@ import yaml
 import os
 import pandas as pd
 from clickhouse_driver.client import Client
+import json
 log = logger()
 
 
@@ -131,8 +132,16 @@ class CkUtils:
         df=pd.DataFrame(data,columns=col)
         return df
 
+    def getData_pdv2(self,sql):
+        data = self.client.execute_iter(sql,with_column_types=True)
+        columns = [column[0] for column in next(data)]
+        df = pd.DataFrame.from_records(data, columns=columns)
+        return df
 
+        # print(df.to_json(orient='records'))
 
+    def getData_json(self,sql):
+        return self.getData_pdv2(sql).to_json(orient='records')
 
     def getColumns(self,table,is_list=False):
         data=self.execute("desc "+table)