1

app/api_data/get_order.py

@@ -95,9 +95,9 @@ def daily():
     yangguang(et, et)
     huasheng(et, et)
 
-    # 昨日新增账户 自动往前跑10天数据
-    yangguang(st, et, new=True)
-    huasheng(st, et, new=True)
+    # 每晚往前拉10天的数据
+    yangguang(st, et)
+    huasheng(st, et)
 
 
 

app/etl/data_stat_run.py

@@ -5,7 +5,7 @@ from data_stat_task import *
 from sync_to_ck_task import *
 from app.etl.dm.dm_pitcher_daily_overview import dm_pitcher_daily_overview
 from app.etl.dw.dw_channel_daily import dw_daily_channel
-from app.etl.dw.dw_pitcher_daily import dw_pitcher_daily,dw_pitcher_trend
+from app.etl.dw.dw_pitcher_daily import dw_pitcher_trend
 from app.etl.ods.ods_book_info_daily import book_info_daily
 log=logger()
 du = DateUtils()
@@ -29,7 +29,6 @@ def main(st,et):
     do_order(st,et)
     do_cost(st,et)
     dw_daily_channel()
-    dw_pitcher_daily()
     dw_pitcher_trend()
     dm_pitcher_daily_overview()
 
@@ -51,4 +50,4 @@ def daily():
 
 
 if __name__ == '__main__':
-   do_cost('2020-12-01','2021-01-28')
+    hourly()

app/etl/dm/dm_pitcher_daily_overview.py

@@ -80,7 +80,7 @@ left outer join (
 """
     data=ck.execute(sql)
 # print(data)
-    db.quchen_text.executeMany("replace into dm_pitcher_daily_overview values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",data)
+    db.dm.executeMany("replace into dm_pitcher_daily_overview values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",data)
 
 if __name__ == '__main__':
     dm_pitcher_daily_overview()

app/etl/dw/dw_channel_daily.py

@@ -64,10 +64,12 @@ select formatDateTime(reg_time,'%Y-%m-%d') reg_date,channel,count(1) user_order_
 
     data=ck.execute('select * from dw_daily_channel')
     insert_sql="insert into dw_channel_daily values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
-    st=time.time()
+    db.dm.execute("truncate table dw_channel_daily")
+    db.dm.executeMany(insert_sql,data)
+
     db.quchen_text.execute("truncate table dw_channel_daily")
     db.quchen_text.executeMany(insert_sql,data)
-    print(time.time()-st)
+
 
 
 if __name__ == '__main__':

app/etl/dw/dw_pitcher_daily.py

@@ -10,14 +10,6 @@ db= MysqlUtils()
 ck = CkUtils()
 
 
-def dw_pitcher_daily():
-    sql="""
-    select dt,pitcher,sum(order_amount),sum(cost),sum(reg_order_amount),sum(first_order_amount),0,0
-    from dw_daily_channel group by dt,pitcher
-    """
-    data = ck.execute(sql)
-    db.quchen_text.executeMany('replace into dw_pitcher_daily values (%s,%s,%s,%s,%s,%s,%s,%s)',data)
-
 def dw_pitcher_trend():
     sql="""select a.dt,pitcher,
        sum(cost),
@@ -68,12 +60,15 @@ from order group by channel,dt) b on a.dt=b.dt and a.channel=b.channel
 group by pitcher,a.dt"""
 
     data=ck.execute(sql)
-    db.quchen_text.execute("truncate table dw_pitcher_trend")
-    db.quchen_text.executeMany("replace into dw_pitcher_trend values (%s,%s,%s,%s,%s,%s,%s,"
+    db.dm.execute("truncate table dw_pitcher_trend")
+    db.dm.executeMany("replace into dw_pitcher_trend values (%s,%s,%s,%s,%s,%s,%s,"
                            "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,"
                            "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",data)
 
-
+    db.quchen_text.execute("truncate table dw_pitcher_trend")
+    db.quchen_text.executeMany("replace into dw_pitcher_trend values (%s,%s,%s,%s,%s,%s,%s,"
+                      "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,"
+                      "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", data)
 
 
 if __name__ == '__main__':

config/db_config.yaml

@@ -8,4 +8,10 @@ zx:
   host: rm-bp145mi6r24ik50z5xo.mysql.rds.aliyuncs.com
   user: readonly
   passwd: toDataOfReadOnly$2020
-  db: zx_prod
+  db: zx_prod
+
+dm:
+  host: rm-bp10mvfktc9o41ir91o.mysql.rds.aliyuncs.com
+  user: qucheng
+  passwd: qucheng_123
+  db: dm

model/DataBaseUtils.py

@@ -15,6 +15,7 @@ log = logger()
 class MysqlUtils:
     _quchen_text = None
     _zx=None
+    _dm=None
 
     def __init__(self):
         p_path =os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
@@ -45,6 +46,14 @@ class MysqlUtils:
                                   db=conf['db'])
         return self._zx
 
+    @property
+    def dm(self):
+        conf = self.config['dm']
+        self._dm = MysqlOperation(host=conf['host'],
+                                  user=conf['user'],
+                                  passwd=conf['passwd'],
+                                  db=conf['db'])
+        return self._dm
 
 
     def find_db(self, db):
@@ -65,7 +74,9 @@ class MysqlUtils:
             self._zx.cursor.close()
             self._zx.conn.close()
 
-
+        if self._dm:
+            self._dm.cursor.close()
+            self._dm.conn.close()
 
 class CkUtils: