| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124 | # 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 datetimeimport threadingfrom time import sleepfrom utils.mysql_utils_pro import MysqlProUtilsclass 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_countselect 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_statusfrom        (    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_datewhere 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)                continueif __name__ == '__main__':    databases = {'rocket', 'miaogousi', 'yufu', 'ruixuan'}    for database in databases:        CanalManagementThread().KuaishouRuleOrderDataHandler(database=database)
 |