# 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}.promoter_item_order_count select a.promoter_id, promoter_nick_name as promoter_name, promoter_url, a.item_id, a.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, a.stat_date, sample_voucher_status from ( 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(case when order_status != 80 then regimental_promotion_amount else 0 end) as regimental_promotion_amount, sum(total_regimental_settle_amount) as total_regimental_settle_amount, stat_date as stat_date from {database}.kuaishou_supply_chain where stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d') and stat_date is not null and item_id={item_id} group by promoter_id, item_id, stat_date ) a left join (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}) 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, stat_date as stat_date from {database}.kuaishou_supply_chain where order_status != 80 and 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 ) c on a.promoter_id = c.promoter_id and a.item_id = c.item_id and a.stat_date=c.stat_date where a.stat_date is not null """.format(item_id=item_id,database=database) try: print(handler_sql) MysqlProUtils().Operate(sql=handler_sql) except Exception as e: print(e) continue if __name__ == '__main__': databases = {'rocket', 'miaogousi', 'yufu', 'ruixuan'} for database in databases: CanalManagementThread().KuaishouRuleOrderDataHandler(database=database)