PromoterItemOrderCountThread.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124
  1. # Author renyupeng
  2. # coding=utf-8
  3. # @Time : 2024/7/8 15:54
  4. # @Site :
  5. # @File : CanalManagementThread.py
  6. # @Software: PyCharm
  7. # @contact: renyupeng@c-top.com.cn
  8. # @Tel 1501435553
  9. # 达人商品订单
  10. import datetime
  11. import threading
  12. from time import sleep
  13. from utils.mysql_utils_pro import MysqlProUtils
  14. class CanalManagementThread:
  15. def __init__(self):
  16. self.stat_date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
  17. self.date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
  18. def KuaishouRuleOrderDataHandler(self,database):
  19. sql = """
  20. select item_id from {database}.kuaishou_supply_chain where stat_date='{date}'
  21. group by item_id
  22. """.format(date=self.date,database=database)
  23. results = MysqlProUtils().QueryAll(sql=sql)
  24. items_per_thread = len(results)
  25. threads = []
  26. for i in range(10):
  27. start = i * items_per_thread
  28. end = start + items_per_thread
  29. thread = threading.Thread(target=self.Theard_hander, args=(results[start:end],database))
  30. threads.append(thread)
  31. thread.start()
  32. for thread in threads:
  33. thread.join()
  34. def Theard_hander(self, data,database):
  35. for result in data:
  36. item_id = result[0]
  37. handler_sql = """
  38. replace into {database}.promoter_item_order_count
  39. select a.promoter_id,
  40. promoter_nick_name as promoter_name,
  41. promoter_url,
  42. a.item_id,
  43. a.item_title as item_name,
  44. item_img_url,
  45. item_create_id,
  46. item_create_name,
  47. collect_sample_id,
  48. collect_sample_name,
  49. order_count,
  50. valid_order_num,
  51. send_out_goods_num,
  52. no_send_out_goods_num,
  53. order_amount,
  54. valid_order_amount,
  55. regimental_promotion_amount,
  56. total_regimental_settle_amount,
  57. a.stat_date,
  58. sample_voucher_status
  59. from
  60. (
  61. select promoter_id,
  62. item_id,
  63. item_title,
  64. count(distinct oid) as order_count,
  65. sum(case when send_status = 1 then 1 else 0 end) as send_out_goods_num,
  66. sum(case when send_status = 0 then 1 else 0 end) as no_send_out_goods_num,
  67. sum(order_amount) as order_amount,
  68. sum(case when order_status != 80 then regimental_promotion_amount else 0 end) as regimental_promotion_amount,
  69. sum(total_regimental_settle_amount) as total_regimental_settle_amount,
  70. stat_date as stat_date
  71. from {database}.kuaishou_supply_chain
  72. where stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d')
  73. and stat_date is not null and item_id={item_id}
  74. group by promoter_id, item_id, stat_date
  75. ) a left join
  76. (select item_id,
  77. item_img_url,
  78. item_create_id,
  79. item_create_name,
  80. collect_sample_id,
  81. collect_sample_name,
  82. promoter_id,
  83. promoter_nick_name,
  84. promoter_url,
  85. create_time,
  86. sample_voucher_status
  87. from {database}.kuaishou_item_collect_samples
  88. where collect_sample_status != 2 and item_id={item_id}) b
  89. on a.promoter_id = b.promoter_id and a.item_id = b.item_id
  90. left join(
  91. select promoter_id,
  92. item_id,
  93. count(distinct oid) as valid_order_num,
  94. sum(order_amount) as valid_order_amount,
  95. stat_date as stat_date
  96. from {database}.kuaishou_supply_chain
  97. where order_status != 80
  98. and stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d')
  99. and item_id={item_id}
  100. and stat_date is not null
  101. group by promoter_id, item_id, stat_date
  102. ) c
  103. on a.promoter_id = c.promoter_id and a.item_id = c.item_id and a.stat_date=c.stat_date
  104. where
  105. a.stat_date is not null
  106. """.format(item_id=item_id,database=database)
  107. try:
  108. print(handler_sql)
  109. MysqlProUtils().Operate(sql=handler_sql)
  110. except Exception as e:
  111. print(e)
  112. continue
  113. if __name__ == '__main__':
  114. databases = {'rocket', 'miaogousi', 'yufu', 'ruixuan'}
  115. for database in databases:
  116. CanalManagementThread().KuaishouRuleOrderDataHandler(database=database)