MOD:代码格式修改

app/api_data/platform_order/get_order.py

@@ -116,7 +116,8 @@ def zhangzhongyun(start=None, end=None, new=None):
     else:
         logging.info(f'掌中云有账号{len(accounts)}个')
     for account in accounts:
-        executor.submit(get_zzy_order_task, start, end, account)
+        executor.submit(
+            get_zzy_order_task, start, end, account)
     executor.shutdown()
     logging.info('掌中云订单数据拉取,结束')
 

app/api_data/platform_order/order_util.py

@@ -60,7 +60,6 @@ def get_huasheng_order(start, end, account, merchant):
     limit = 500
 
     for date in du.getDateLists(start, end):
-
         page = 1
         while True:
             sign = md5(apiKey + date + str(merchant_id) + timestamp + apiSecurity).upper()
@@ -92,7 +91,6 @@ def get_huasheng_order(start, end, account, merchant):
             order_item_list = response_result_json['data']
             if len(order_item_list) == 0:
                 break
-
             for i in order_item_list:
                 li.append(
                     (i['request_at'][:10],

app/api_data/platform_order/yangguang.py

@@ -262,7 +262,7 @@ if __name__ == '__main__':
 
     # get_channel_info()
     # exit(0)
-    yangguang()
+    yangguang(start=du.get_n_days(-1),end=du.get_n_days(0))
     # # exit(0)
     # yangguang('2021-05-28','2021-05-28')
     # daily_yg()

app/api_data/platform_order/youshuge.py

@@ -3,12 +3,12 @@ from model.DateUtils import DateUtils
 from model.ComUtils import md5
 import requests
 from app.api_data.platform_order.order_util import save_order
+
 ut = DateUtils()
 
 
 def get_youshuge_order_task(st, et, account):
-
-    et = ut.add_days(et,1)
+    et = ut.add_days(et, 1)
     url = 'https://novel.youshuge.com/v2/open/orders'
     host_name = account[0]
     channel_id = int(account[1])
@@ -49,11 +49,11 @@ def get_youshuge_order_task(st, et, account):
                        i['openid'],
                        i["create_time"],
                        i['reg_time'],
-                       int(i['price'])/100,
+                       int(i['price']) / 100,
                        i['book_name'],
                        i['order_num'],
                        2 if i['pay_status'] == '1' else 1
-            ))
+                       ))
 
         page += 1
 
@@ -62,9 +62,7 @@ def get_youshuge_order_task(st, et, account):
         save_order(li)
 
 
-
 if __name__ == '__main__':
     a = "趣程15期,10696,8OC7SNCL46ZEI7JBACXFDM8CP5JM1FSL,盛德文苑,趣程15期"
-    get_youshuge_order_task('2021-01-06', '2021-05-06', a.split(','))
-
-
+    a = '趣程清勇6,10675,B6FK1HFW1V9ZPD6WX0FGBX3WP73E6M5V,璃月文楼,清勇7月'
+    get_youshuge_order_task('2021-01-06', '2021-08-06', a.split(','))

app/api_data/platform_order/yuewen.py

@@ -58,18 +58,14 @@ def get_yuewen_order_task(st, et, account):
             str_params = ''
             for k, v in sorted_data:
                 str_params = str_params + str(k) + str(v)
-
             sign = md5(appsecert + str_params).upper()
 
             # 放入签名
             params['sign'] = sign
             # print(params)
-
             response_result_json = requests.get(url=url, params=params).json()
-            # print(response_result_json)
 
             # print(len(response_result_json["data"]["list"]))
-
             if not response_result_json.get('data'):
                 print(response_result_json)
                 DingTalkUtils().send('阅文订单拉取失败')
@@ -118,4 +114,6 @@ def get_yuewen_order_task(st, et, account):
 if __name__ == '__main__':
     a = "guangzhouliuqi2@sina.com,10ce1dd6ccb330a82b73701d1e78f518"
     b = "mqud82950@163.com,74ca754515fa253c8ab790603cebc2ee"
-    get_yuewen_order_task('2021-05-28', '2021-05-28', a.split(','))
+    {"version": "1", "app_secret": "74ca754515fa253c8ab790603cebc2ee", "email": "mqud82950@163.com"}
+    a = 'mqud82950@163.com,74ca754515fa253c8ab790603cebc2ee'
+    get_yuewen_order_task('2021-08-01', '2021-08-02', a.split(','))

app/api_data/tx_ad_cost/cost_util.py

@@ -11,6 +11,9 @@ from model.DateUtils import DateUtils
 from PIL import Image
 from io import BytesIO
 import cv2
+import oss2
+import os
+import ffmpeg
 
 du = DateUtils()
 db = MysqlUtils()
@@ -291,6 +294,7 @@ def images_info_get(account_id, access_token, image_ids):  # 获取图片信息
 def video_info_get(account_id, access_token, image_ids):  # 获取视频信息
     # 接口 https://developers.e.qq.com/docs/api/business_assets/video/videos_get?version=1.3
 
+
     def get_video_info(video_url, err_num=0):
         try:
             if video_url:

app/api_data/tx_ad_cost/get_cost.py

@@ -1,4 +1,3 @@
-
 from app.api_data.tx_ad_cost.cost_util import *
 from model.DateUtils import DateUtils
 from model.DataBaseUtils import MysqlUtils
@@ -9,22 +8,25 @@ import logging
 db = MysqlUtils()
 du = DateUtils()
 
-max_workers =10
+max_workers = 10
 
-def get_accounts(filter=None):
-	if filter:
-		if filter=='MP':
-			return db.quchen_text.getData("select account_id,access_token,name channel from advertiser_vx where (name !='' or name is not null)")
-		else:
-			return db.quchen_text.getData("select account_id,access_token,name channel from advertiser_qq where (name !='' or name is not null)")
 
-	return db.quchen_text.getData("select account_id,access_token,name channel,'MP' flag from advertiser_vx where (name !='' or name is not null) union "
-								   "select account_id,access_token,name channel,'GDT' flag from advertiser_qq where (name !='' or name is not null)")
+def get_accounts(filter=None):
+    if filter:
+        if filter == 'MP':
+            return db.quchen_text.getData(
+                "select account_id,access_token,name channel from advertiser_vx where (name !='' or name is not null)")
+        else:
+            return db.quchen_text.getData(
+                "select account_id,access_token,name channel from advertiser_qq where (name !='' or name is not null)")
 
+    return db.quchen_text.getData(
+        "select account_id,access_token,name channel,'MP' flag from advertiser_vx where (name !='' or name is not null) union "
+        "select account_id,access_token,name channel,'GDT' flag from advertiser_qq where (name !='' or name is not null)")
 
 
 def ad(dt):
-	sql =f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(ad_id)  from ad_cost_day  a
+    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(ad_id)  from ad_cost_day  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)
@@ -32,32 +34,26 @@ select account_id,access_token,'GDT' type  from advertiser_qq where (name !='' o
 where a.dt='{dt}'
 GROUP BY b.account_id,b.access_token,b.type"""
 
-	accounts = db.quchen_text.getData(sql)
-	executor = ThreadPoolExecutor(max_workers=max_workers)
-
-	for account in accounts:
-		executor.submit(get_ad_info, account[0], account[1], account[2],account[3],dt)
-	executor.shutdown()
-
-
+    accounts = db.quchen_text.getData(sql)
+    executor = ThreadPoolExecutor(max_workers=max_workers)
 
+    for account in accounts:
+        executor.submit(get_ad_info, account[0], account[1], account[2], account[3], dt)
+    executor.shutdown()
 
 
 """广告日消耗"""
-def ad_cost_day(st,et):
-
-	executor = ThreadPoolExecutor(max_workers=max_workers)
-	for account in get_accounts():
-		executor.submit(get_ad_cost_day, account[0], account[1],account[3],st,et)
-	executor.shutdown()
-
 
 
+def ad_cost_day(st, et):
+    executor = ThreadPoolExecutor(max_workers=max_workers)
+    for account in get_accounts():
+        executor.submit(get_ad_cost_day, account[0], account[1], account[3], st, et)
+    executor.shutdown()
 
 
 def adcreative(dt):
-
-	sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(adcreative_id)  from ad_info  a
+    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(adcreative_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)
@@ -65,44 +61,46 @@ select account_id,access_token,'GDT' type from advertiser_qq where (name !='' or
 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)
-	# print(accounts)
-	executor = ThreadPoolExecutor(max_workers=max_workers)
-	for account in accounts:
-		executor.submit(get_adcreatives, account[0], account[1],account[2],account[3],dt)
-	executor.shutdown()
+    accounts = db.quchen_text.getData(sql)
+    # print(accounts)
+    executor = ThreadPoolExecutor(max_workers=max_workers)
+    for account in accounts:
+        executor.submit(get_adcreatives, account[0], account[1], account[2], account[3], dt)
+    executor.shutdown()
+
 
 def image(dt):
-	sql=f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(image_id)  from adcreative_info  a
+    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(image_id)  from adcreative_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}' and a.is_video=0
 GROUP BY b.account_id,b.access_token,b.type"""
-	accounts = db.quchen_text.getData(sql)
-	executor = ThreadPoolExecutor(max_workers=max_workers)
-	for account in accounts:
-		executor.submit(images_info_get, account[0], account[1], account[3])
-	executor.shutdown()
+    accounts = db.quchen_text.getData(sql)
+    executor = ThreadPoolExecutor(max_workers=max_workers)
+    for account in accounts:
+        executor.submit(images_info_get, account[0], account[1], account[3])
+    executor.shutdown()
 
 
 def video(dt):
-	sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(image_id)  from adcreative_info  a
+    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(image_id)  from adcreative_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}' and a.is_video=1
 	GROUP BY b.account_id,b.access_token,b.type"""
-	accounts = db.quchen_text.getData(sql)
-	executor = ThreadPoolExecutor(max_workers=max_workers)
-	for account in accounts:
-		executor.submit(video_info_get, account[0], account[1], account[3])
-	executor.shutdown()
+    accounts = db.quchen_text.getData(sql)
+    executor = ThreadPoolExecutor(max_workers=max_workers)
+    for account in accounts:
+        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
+    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)
@@ -110,73 +108,88 @@ def campaign(dt):
 	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()
+    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.拉取有消耗的广告
-		2.用有消耗的广告id 去拉取广告基础信息
-		3.用第2步获取的创意id 去拉取广告创意基础信息
-		4.用创意信息中的图片id 去获取图片的基础信息
-		"""
-	try:
-		logging.info('获取广告id,开始')
-		ad_cost_day(dt, dt)
-		logging.info('获取广告id,结束')
-		logging.info('获取广告基础信息,开始')
-		ad(dt)
-		logging.info('获取广告基础信息,结束')
-		logging.info('获取广告创意素材,开始')
-		adcreative(dt)
-		logging.info('获取广告创意素材,结束')
-		logging.info('获取图片信息,开始')
-		image(dt)
-		logging.info('获取图片信息,结束')
-		logging.info('获取视频信息,开始')
-		video(dt)
-		logging.info('获取视频信息,结束')
-
-	except:
-		DingTalkUtils().send("拉取广告数据出错")
+    """
+        1.拉取有消耗的广告
+        2.用有消耗的广告id 去拉取广告基础信息
+        3.用第2步获取的创意id 去拉取广告创意基础信息
+        4.用创意信息中的图片id 去获取图片的基础信息
+        """
+    try:
+        logging.info('获取广告id,开始')
+        ad_cost_day(dt, dt)
+        logging.info('获取广告id,结束')
+        logging.info('获取广告基础信息,开始')
+        ad(dt)
+        logging.info('获取广告基础信息,结束')
+        logging.info('获取广告创意素材,开始')
+        adcreative(dt)
+        logging.info('获取广告创意素材,结束')
+        logging.info('获取图片信息,开始')
+        image(dt)
+        logging.info('获取图片信息,结束')
+        logging.info('获取视频信息,开始')
+        video(dt)
+        logging.info('获取视频信息,结束')
+
+    except:
+        DingTalkUtils().send("拉取广告数据出错")
 
 
 def day():
-	ad_cost_day(du.get_n_days(-10), du.get_n_days(0))
+    ad_cost_day(du.get_n_days(-10), du.get_n_days(0))
 
 
 def hourly():
-	dt = du.getNow()
-	run(dt)
+    dt = du.getNow()
+    run(dt)
 
-if __name__ == '__main__':
-	# MP
-	# account_id = 18516323
-	# access_token = '262deda76aec00c2e144e83bd3c0b2a2'
-	#
-	# account_id2= 14709511
-	# access_token2 = 'e87f7b6f860eaeef086ddcc9c3614678'
-	# run()
-	# ad_cost_day('2020-04-08','2021-04-07')
-	
-	# day()
-	#
-
-	# day()
-	# run('2021-05-10')
-	# adcreative('2021-05-11')
-	# video('2021-05-14')
-	# campaign('2021-05-14')
-	for dt in list(reversed(du.getDateLists('2020-05-18','2021-05-17'))):
-		print(dt)
-		# ad(dt)
-		adcreative(dt)
-		# image(dt)
-		# video(dt)
 
+if __name__ == '__main__':
+    from logging import handlers
+
+    logging.basicConfig(
+        handlers=[
+            logging.handlers.RotatingFileHandler('./dw_image_cost.log',
+                                                 maxBytes=10 * 1024 * 1024,
+                                                 backupCount=5,
+                                                 encoding='utf-8')
+            , logging.StreamHandler()  # 供输出使用
+        ],
+        level=logging.INFO,
+        format="%(asctime)s - %(levelname)s %(filename)s %(funcName)s %(lineno)s - %(message)s"
+    )
+    # MP
+    # account_id = 18516323
+    # access_token = '262deda76aec00c2e144e83bd3c0b2a2'
+    #
+    # account_id2= 14709511
+    # access_token2 = 'e87f7b6f860eaeef086ddcc9c3614678'
+    # run()
+    # ad_cost_day('2020-04-08','2021-04-07')
+
+    # day()
+    #
+
+    # day()
+    # run('2021-05-10')
+    # adcreative('2021-05-11')
+    # video('2021-05-14')
+    # campaign('2021-05-14')
+    ad_cost_day(du.get_n_days(-365), du.get_n_days(0))
+
+    for dt in list(reversed(du.getDateLists(du.get_n_days(-500), du.get_n_days(0)))):
+        print(dt)
+        # run(dt)
+        # ad(dt)
+        # adcreative(dt)
+    # image(dt)
+    # video(dt)

app/crontab_task/task.py

@@ -106,5 +106,5 @@ def yueweng_order_repair():
 
 
 if __name__ == '__main__':
-    # hourly()
-    cost_yestoday_repair()
+    hourly()
+    # cost_yestoday_repair()

config/db_config.yaml

@@ -29,6 +29,13 @@ zx_test:
   passwd: zxdev3306
   db: zx-operation-test
 
+#zx_ads:
+#  host: 118.178.187.109
+#  user: root
+#  passwd: zxdev3306
+#  db: zx-ads-test
+
+
 clickhouse:
   host: cc-bp1h3yc7o3g3o7k64o.ads.aliyuncs.com
   user: qucheng_ck

example/update_cost_data.py

@@ -103,6 +103,6 @@ if __name__ == "__main__":
     )
     st = du.get_n_days(-20)
     et = du.get_n_days(0)
-    get_data_vx(channel='安宜文海', st=st, et=et)
+    get_data_vx(channel='语慕书海', st=st, et=et)
 
     # get_data(st,et)

example/update_media_info.py

@@ -4,6 +4,7 @@ import pandas
 import requests
 import time
 
+
 def update_media_info():
     # 1.按用户维度获取图片
     sql = '''
@@ -30,8 +31,7 @@ def update_media_info():
             where 
         '''
 
-        #3.获取图片信息(腾讯),图片信息(大小,格式)
-
+        # 3.获取图片信息(腾讯),图片信息(大小,格式)
 
 
 from PIL import Image
@@ -39,28 +39,30 @@ from io import BytesIO
 
 
 def update_image_info():
+    db_qc = DB(config=quchen_text)
+    db_dm = DB(config=dm)
     while True:
-        user_sql='''
+        user_sql = '''
          select * from image_info ii 
         where preview_url is not null 
-        and size is null
+        and (size is null or type is null)
         limit 100
         '''
-        df=pandas.read_sql(user_sql,con=db_qc.engine)
-        if len(df)==0:
+        df = pandas.read_sql(user_sql, con=db_qc.engine)
+        if len(df) == 0:
             break
-        for index,row in df.iterrows():
+        for index, row in df.iterrows():
             try:
                 print(row['preview_url'])
-                rsp=requests.get(row['preview_url'])
-                #1.图片写入内存
+                rsp = requests.get(row['preview_url'])
+                # 1.图片写入内存
                 im = Image.open(BytesIO(rsp.content))
-                #2.获取图片属性
+                # 2.获取图片属性
                 image_format = im.format
                 image_size = len(rsp.content)
-                #3.数据进行存储
-                sql=''' update image_info set size={},type='{}'
-                            where image_id = '{}'  '''.format(image_size,image_format,row['image_id'])
+                # 3.数据进行存储
+                sql = ''' update image_info set size={},type='{}'
+                            where image_id = '{}'  '''.format(image_size, image_format, row['image_id'])
                 db_qc.session.execute(sql)
                 db_qc.session.commit()
             except Exception as e:
@@ -69,7 +71,12 @@ def update_image_info():
 
 
 if __name__ == '__main__':
-    db_qc = DB(config=quchen_text)
-    db_dm = DB(config=dm)
+
     # update_media_info()
-    update_image_info()
+    import threading
+
+    for i in range(5):
+        thread_update = threading.Thread(target=update_image_info, )
+        thread_update.start()
+
+

example/update_order_data.py

@@ -9,18 +9,17 @@ from app.etl.dw.dw_book_trend import book_trend
 import threading
 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.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
 from app.api_data.tx_ad_cost import get_cost_older
 from app.etl.data_stat_run import do_cost
 from logging import handlers
 
-
 du = DateUtils()
 
-def order_date_get(st,et):
 
+def order_date_get(st, et):
     t1 = threading.Thread(target=huasheng, args=(st, et))
     t2 = threading.Thread(target=qiyue, args=(st, et))
     t3 = threading.Thread(target=qiyueyousheng, args=(st, et))
@@ -53,13 +52,14 @@ def order_date_get(st,et):
     t10.join()
     yangguang.get_channel_info()
 
+
 def update_order():
     # 1.获取数据
 
     st = du.get_n_days(-30)
     et = du.get_n_days(0)
     # print(st, et)
-    order_date_get(st,et)
+    order_date_get(st, et)
     # do_order(st, et)
     # src_book_info()  # 书籍卡点信息
     # # book_annual_expect_profit.run() # 年预期收益
@@ -68,11 +68,21 @@ def update_order():
     # book_trend()
     # dm_pitcher_daily_overview()
 
+
 def update_order_zhangzhognyun():
-    zhangzhongyun(start=du.get_n_days(-15),end=du.get_n_days(0))
+    zhangzhongyun(start=du.get_n_days(-15), end=du.get_n_days(0))
+
+
+def update_order_zhangzhongyun_single():
+    i = 'vip-qucheng8qi,17838a8f9ea64f809a88a5cd872c7375,'
+    new_data = i.replace('\n', '').split(",")
+    print(new_data)
+    get_zzy_order_task(start=du.get_n_days(-1), end=du.get_n_days(0), account=new_data)
+
 
 def update_order_guofeng():
-    guofeng(start=du.get_n_days(-15),end=du.get_n_days(0))
+    guofeng(start=du.get_n_days(-15), end=du.get_n_days(0))
+
 
 def update_order_do():
     st = du.get_n_days(-30)
@@ -80,7 +90,8 @@ def update_order_do():
     print(st, et)
     do_order(st, et)
 
-if __name__=='__main__':
+
+if __name__ == '__main__':
     logging.basicConfig(
         handlers=[
             logging.handlers.RotatingFileHandler('./update_order_data.log',
@@ -96,6 +107,5 @@ if __name__=='__main__':
     # update_order()
     # update_order()
     # update_order_zhangzhognyun()
-    update_order_guofeng()
-
-
+    update_order_zhangzhognyun()
+    # update_order_zhangzhongyun_single()

example/update_video.py

@@ -47,7 +47,7 @@ def update_video_info():
 def update_byte_rate():
     sql = '''
                 select * from video_info vi 
-                    where length (preview_url )>0;
+                    where length (preview_url )>0 and not type
                 '''
     df = pandas.read_sql(sql=sql, con=db_qc.engine)
     print(df)
@@ -78,4 +78,4 @@ if __name__ == '__main__':
     db_qc = DB(config=quchen_text)
     db_dm = DB(config=dm)
     # update_video_info()
-    update_byte_rate()
+    # update_byte_rate()

model/DateUtils.py

@@ -62,6 +62,7 @@ class DateUtils:
         返回一个时间列表
         """
         interval = self.getInterval(begin, end)
+        print(interval)
         return [self.getLastDays(begin, -x) for x in range(interval + 1)]
 
     def getTodayOrYestoday(self):