1

app/api_data/get_cost.py

@@ -53,7 +53,6 @@ def ad_cost_day(st,et):
 
 
 
-
 def adcreative(dt):
 
 	sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(adcreative_id)  from ad_info  a
@@ -94,7 +93,7 @@ def run(dt):
 		"""
 	try:
 
-		ad_cost_day(dt,dt)
+		ad_cost_day(dt, dt)
 		ad(dt)
 		adcreative(dt)
 		image(dt)

app/api_data/get_order.py

@@ -40,10 +40,10 @@ def huasheng(start=None, end=None):
         return
     else:
         print(f'花生有账号{len(accounts)}个')
-
+    # 花生有请求限制 不用多线程
     for account in accounts:
-        executor.submit(get_hs_order_task, start, end, account)
-    executor.shutdown()
+        get_hs_order_task(start, end, account)
+
 
 
 @DingTalkDecorators("七悦")
@@ -176,4 +176,5 @@ def daily():
 
 if __name__ == '__main__':
     # zhangzhongyun()
-    yangguang()
+    # yangguang()
+    huasheng("2021-04-12",'2021-05-10')

app/api_data/order_util.py

@@ -123,11 +123,12 @@ def get_hs_order_task(start, end, account):
 
     channel_data = response_result_json['data']
 
-    print(f"花生账号：{account[2]} 有channel{len(channel_data)}个:{str([i['merchant_name'] for i in channel_data])}")
+    # print(f"{account[2]} 有channel{len(channel_data)}个:{str([i['merchant_name'] for i in channel_data])}")
 
     li = []
     for merchant in channel_data:
-        get_huasheng_order(start, end, account, merchant, li)
+        orders = get_huasheng_order(start, end, account, merchant)
+        li.extend(orders)
 
     if len(li) > 0:
         print(f"花生账号：{account[2]} 有order{len(li)}个")
@@ -137,8 +138,8 @@ def get_hs_order_task(start, end, account):
 
 
 
-def get_huasheng_order(start,end, account, merchant,li):
-
+def get_huasheng_order(start,end, account, merchant):
+    li =[]
     apiKey = str(account[0])
     apiSecurity = account[1]
     stage = account[2]
@@ -169,6 +170,7 @@ def get_huasheng_order(start,end, account, merchant,li):
 
             response_result_json = r.json()
 
+
             if 'data' not in response_result_json.keys():
                 print('花生账号【{key}】, 查询时间【{date}】, 渠道【{merchant_id}:{merchant_name}】本次请求数据异常，响应报文【{result}】'
                       .format(key=apiKey, date=date, merchant_id=merchant_id, merchant_name=merchant_name,
@@ -203,7 +205,7 @@ def get_huasheng_order(start,end, account, merchant,li):
                 break
             else:
                 page = page + 1
-
+    return li
 
 
 def save_hs_data(data):

app/etl/data_stat_run.py

@@ -18,7 +18,6 @@ logging.getLogger().setLevel(logging.WARNING)
 def do_order(st,et):
     for i in du.getDateLists(st,et):
         print(i)
-        ods_order(i)
         order_sync_ck(i)
 def do_cost(st,et):
     for i in du.getDateLists(st,et):
@@ -57,7 +56,7 @@ def daily():
 
 
 if __name__ == '__main__':
-    hourly()
+    # hourly()
     # do_cost('2021-02-10','2021-02-10')
 
-    # do_order('2021-04-16','2021-04-17')
+    do_order('2021-05-07','2021-05-07')

config/db_config.yaml

@@ -14,4 +14,10 @@ dm:
   host: rm-bp10mvfktc9o41ir91o.mysql.rds.aliyuncs.com
   user: qucheng
   passwd: Qc_123456
-  db: dm
+  db: dm
+
+zx_test:
+  host: 118.178.187.109
+  user: root
+  passwd: zxdev3306
+  db: zx-ads-test

model/DataBaseOperation.py

@@ -423,6 +423,16 @@ class MysqlOperation:
             split=split
         )
 
+    def dfsave2mysql(self, df, table, key, tag):
+        self.insertorupdatemany_v2(
+            table=table,
+            keys=key,
+            tags=tag,
+            tag_values=df[tag],
+            key_values=df[key]
+        )
+
+
     def _get_s_format(self, data):
         """
         Args:
@@ -520,7 +530,7 @@ class MysqlOperation:
         t0 = time.time()
 
         self.cursor.execute(sql_insert,tuple(value_insert))
-        log.debug('insert %s rows, cost: %s' % (len(key_values), time.time() - t0))
+        log.debug('insert %s rows, cost: %s' % (len(key_values), round(time.time() - t0, 2)))
         self.conn.commit()
 
     def update_many(self, table, keys, tags, tag_values, key_values, flag=False, split=80):
@@ -612,7 +622,7 @@ class MysqlOperation:
         t0 = time.time()
         self.cursor.execute(sql, tuple(update_value))
         self.conn.commit()
-        log.debug('update %s rows, cost: %s' % (len(key_values), time.time() - t0))
+        log.debug('update %s rows, cost: %s' % (len(key_values), round(time.time() - t0, 2)))
 
 
     def getColumn(self,table,flag=0):

model/DataBaseUtils.py

@@ -16,6 +16,7 @@ class MysqlUtils:
     _quchen_text = None
     _zx=None
     _dm=None
+    _zx_test =None
 
     def __init__(self):
         p_path =os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
@@ -46,6 +47,16 @@ class MysqlUtils:
                                   db=conf['db'])
         return self._zx
 
+    @property
+    def zx_test(self):
+        conf = self.config['zx_test']
+        self._zx_test = MysqlOperation(host=conf['host'],
+                                  user=conf['user'],
+                                  passwd=conf['passwd'],
+                                  db=conf['db'])
+        return self._zx_test
+
+
     @property
     def dm(self):
         conf = self.config['dm']
@@ -78,6 +89,10 @@ class MysqlUtils:
             self._dm.cursor.close()
             self._dm.conn.close()
 
+        if self._zx_test:
+            self._zx_test.cursor.close()
+            self._zx_test.conn.close()
+
 class CkUtils:
 
     def __init__(self):