renyupeng 9 kuukautta sitten
vanhempi
commit
a4cd0d9683

+ 128 - 0
ETL/CanalManagement.py

@@ -0,0 +1,128 @@
+# 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'}
+    for database in databases:
+        CanalManagementThread().KuaishouRuleOrderDataHandler(database=database)

+ 71 - 0
ETL/ItermOrderAmountThread.py

@@ -0,0 +1,71 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2024/7/8 15:54
+# @Site    : 
+# @File    : ItermOrderAmountThread.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 ItermOrderAmountThread:
+    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}.item_oerder_amount
+select
+a.item_id, valid_order_num, valid_amount,b.30day_order_num,b.30day_gmv,item_first_order_time
+from
+(select item_id, count(distinct oid) as valid_order_num, sum(order_amount) as valid_amount,min(stat_date) as item_first_order_time
+from {database}.kuaishou_supply_chain
+where order_status != 80 and item_id={item_id}
+group by item_id)a
+left join (
+  select item_id, count(distinct oid) 30day_order_num, sum(order_amount) as 30day_gmv
+from {database}.kuaishou_supply_chain
+where order_status != 80 and stat_date>date_format(date_sub({stat_date},interval 30 day),'%Y%m%d')
+and item_id={item_id}
+group by item_id
+    )b
+on a.item_id=b.item_id
+                        """.format(item_id=item_id, stat_date=self.stat_date, 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:
+        ItermOrderAmountThread().KuaishouRuleOrderDataHandler(database=database)

+ 103 - 0
ETL/KuaishouRuleOrderDataTherad.py

@@ -0,0 +1,103 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2024/7/8 15:54
+# @Site    : 
+# @File    : KuaishouRuleOrderData.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 KuaishouRuleOrderData:
+    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}.kuaishou_rule_order_data(relate_id, item_id, item_title, promoter_id, promoter_name,
+                                                          item_create_id, item_create_name, promoter_create_id,
+                                                          promoter_create_name, order_num, valid_order_num, yesterday_order_num,
+                                                          yesterday_valid_order_num, stat_date)
+            select relate_id,
+                   a.item_id,
+                   ifnull(b.item_title, d.item_title),
+                   a.promoter_id,
+                   promoter_nick_name,
+                   ifnull(item_create_id, user_id)     as item_create_id,
+                   ifnull(item_create_name, user_name) as item_create_name,
+                   collect_sample_id                   as promoter_create_id,
+                   collect_sample_name                 as promoter_create_name,
+                   order_count                         as order_num,
+                   a.valid_order_num,
+                   e.order_num                         as yesterday_order_num,
+                   e.valid_order_num                   as yesterday_order_num,
+                   '{stat_date}'                      as stat_date
+    
+            from (select promoter_id,
+                         item_id,
+                         count(distinct oid) as order_count,
+                       count(distinct case when order_status!=80 then oid  end) as valid_order_num
+                  from {database}.kuaishou_supply_chain
+                  where stat_date <= {stat_date}
+                    and item_id = {item_id}
+                  group by promoter_id, item_id) a
+                     left join
+                 (select item_id,
+                         item_title,
+                         promoter_id,
+                         promoter_nick_name,
+                         item_create_id,
+                         item_create_name,
+                         collect_sample_id,
+                         collect_sample_name,
+                         create_time
+                  from {database}.kuaishou_item_collect_samples
+                  where item_id = {item_id}) b
+                 on a.promoter_id = b.promoter_id and a.item_id = b.item_id
+                     left join (select item_id, relate_id, user_id, user_name, item_title
+                                from {database}.kuaishou_item_list
+                                ) d
+                               on a.item_id = d.item_id
+                     left join (select promoter_id, item_id, order_num, valid_order_num
+                                from {database}.kuaishou_rule_order_data
+                                where stat_date = date_format(date_sub('{stat_date}', interval 1 day), '%Y%m%d')
+                                  and item_id = {item_id}) e
+                               on a.promoter_id = e.promoter_id and a.item_id = e.item_id
+                        """.format(stat_date=self.stat_date, 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:
+        KuaishouRuleOrderData().KuaishouRuleOrderDataHandler(database=database)

+ 124 - 0
ETL/PromoterItemOrderCountThread.py

@@ -0,0 +1,124 @@
+# 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)

+ 71 - 0
ETL/PromoterItermOrderAmountThread.py

@@ -0,0 +1,71 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2024/7/8 15:54
+# @Site    : 
+# @File    : PromoterItermOrderAmountThread.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 PromoterItermOrderAmountThread:
+    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_iterm_oerder_amount
+select
+a.promoter_id,a.item_id, valid_order_num, valid_amount,b.30day_order_num,b.30day_gmv
+from
+(select promoter_id,item_id, count(distinct oid) as valid_order_num, sum(order_amount) as valid_amount
+from {database}.kuaishou_supply_chain
+where order_status != 80 and item_id={item_id}
+group by promoter_id,item_id)a
+left join (
+  select promoter_id,item_id, count(distinct oid) 30day_order_num, sum(order_amount) as 30day_gmv
+from {database}.kuaishou_supply_chain
+where order_status != 80 and stat_date>date_format(date_sub(${stat_date},interval 30 day),'%Y%m%d')
+and item_id={item_id}
+group by promoter_id,item_id
+    )b
+on a.item_id=b.item_id and a.promoter_id=b.promoter_id
+                        """.format(stat_date=self.stat_date, 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:
+        PromoterItermOrderAmountThread().KuaishouRuleOrderDataHandler(database=database)

+ 71 - 0
ETL/PromoterOrderAmountThread.py

@@ -0,0 +1,71 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2024/7/8 15:54
+# @Site    :
+# @File    : PromoterOrderAmountThread.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 PromoterOrderAmountThread:
+    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 promoter_id from {database}.kuaishou_supply_chain where stat_date='{date}'
+group by promoter_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:
+            promoter_id = result[0].decode('utf-8')
+            handler_sql = """
+replace into {database}.promoter_oerder_amount
+select
+a.promoter_id, valid_order_num, valid_amount,b.30day_order_num,b.30day_gmv
+from
+(select promoter_id, count(distinct oid) as valid_order_num, sum(order_amount) as valid_amount
+from {database}.kuaishou_supply_chain
+where order_status != 80 and promoter_id={promoter_id}
+group by promoter_id)a
+left join (
+  select promoter_id, count(distinct oid) 30day_order_num, sum(order_amount) as 30day_gmv
+from {database}.kuaishou_supply_chain
+where order_status != 80 and stat_date>date_format(date_sub('{stat_date}',interval 30 day),'%Y%m%d')
+ and promoter_id={promoter_id}
+group by promoter_id
+    )b
+on a.promoter_id=b.promoter_id;
+                        """.format(promoter_id=promoter_id, stat_date=self.stat_date, 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:
+        PromoterOrderAmountThread().KuaishouRuleOrderDataHandler(database=database)