123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128 |
- # Author renyupeng
- # coding=utf-8
- # @Time : 2024/7/8 15:54
- # @Site :
- # @File : CanalManagementThread.py
- # @Software: PyCharm
- # @contact: renyupeng@c-top.com.cn
- # @Tel 1501435553
- # 达人商品订单
- import datetime
- import threading
- from time import sleep
- from utils.mysql_utils_pro import MysqlProUtils
- class CanalManagementThread:
- def __init__(self):
- self.stat_date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
- self.date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
- def KuaishouRuleOrderDataHandler(self, database):
- sql = """
- select item_id from {database}.kuaishou_supply_chain where stat_date='{date}'
- group by item_id
- """.format(date=self.date, database=database)
- results = MysqlProUtils().QueryAll(sql=sql)
- items_per_thread = len(results)
- threads = []
- for i in range(10):
- start = i * items_per_thread
- end = start + items_per_thread
- thread = threading.Thread(target=self.Theard_hander, args=(results[start:end], database))
- threads.append(thread)
- thread.start()
- for thread in threads:
- thread.join()
- def Theard_hander(self, data, database):
- for result in data:
- item_id = result[0]
- handler_sql = """
- replace into {database}.canal_management
- select a.promoter_id,
- promoter_nick_name as promoter_name,
- promoter_url,
- a.item_id,
- b.item_title as item_name,
- item_img_url,
- item_create_id,
- item_create_name,
- collect_sample_id,
- collect_sample_name,
- order_count,
- valid_order_num,
- send_out_goods_num,
- no_send_out_goods_num,
- order_amount,
- valid_order_amount,
- regimental_promotion_amount,
- total_regimental_settle_amount,
- b.stat_date,
- sample_voucher_status
- from (select item_id,
- item_img_url,
- item_create_id,
- item_create_name,
- collect_sample_id,
- collect_sample_name,
- promoter_id,
- promoter_nick_name,
- promoter_url,
- create_time,
- sample_voucher_status
- from {database}.kuaishou_item_collect_samples
- where collect_sample_status != 2
- and item_id={item_id}
- ) a
- left join(
- select promoter_id,
- item_id,
- item_title,
- count(distinct oid) as order_count,
- sum(case when send_status = 1 then 1 else 0 end) as send_out_goods_num,
- sum(case when send_status = 0 then 1 else 0 end) as no_send_out_goods_num,
- sum(order_amount) as order_amount,
- sum(regimental_promotion_amount) as regimental_promotion_amount,
- sum(total_regimental_settle_amount) as total_regimental_settle_amount,
- date_format(stat_date, '%Y%m%d') as stat_date
- from {database}.kuaishou_supply_chain
- where stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d')
- and item_id={item_id}
- and stat_date is not null
- group by promoter_id, item_id, stat_date
- ) b
- on a.promoter_id = b.promoter_id and a.item_id = b.item_id
- left join(
- select promoter_id,
- item_id,
- count(distinct oid) as valid_order_num,
- sum(order_amount) as valid_order_amount,
- date_format(stat_date, '%Y%m%d') as stat_date
- from {database}.kuaishou_supply_chain
- where order_status != 80
- and item_id={item_id}
- and stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d')
- and stat_date is not null
- group by promoter_id, item_id, stat_date
- ) c
- on b.promoter_id = c.promoter_id and b.item_id = c.item_id and b.stat_date=c.stat_date
- where
- date_format(a.create_time,'%Y%m%d')<=b.stat_date
- and b.stat_date is not null
- """.format(item_id=item_id, database=database)
- try:
- print(handler_sql)
- MysqlProUtils().Operate(sql=handler_sql)
- sleep(10)
- except Exception as e:
- print(e)
- continue
- if __name__ == '__main__':
- databases = {'rocket', 'miaogousi', 'yufu', 'ruixuan','yuxiu'}
- for database in databases:
- CanalManagementThread().KuaishouRuleOrderDataHandler(database=database)
|