CanalManagement.py 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  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}.canal_management
  39. select a.promoter_id,
  40. promoter_nick_name as promoter_name,
  41. promoter_url,
  42. a.item_id,
  43. b.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. b.stat_date,
  58. sample_voucher_status
  59. from (select item_id,
  60. item_img_url,
  61. item_create_id,
  62. item_create_name,
  63. collect_sample_id,
  64. collect_sample_name,
  65. promoter_id,
  66. promoter_nick_name,
  67. promoter_url,
  68. create_time,
  69. sample_voucher_status
  70. from {database}.kuaishou_item_collect_samples
  71. where collect_sample_status != 2
  72. and item_id={item_id}
  73. ) a
  74. left join(
  75. select promoter_id,
  76. item_id,
  77. item_title,
  78. count(distinct oid) as order_count,
  79. sum(case when send_status = 1 then 1 else 0 end) as send_out_goods_num,
  80. sum(case when send_status = 0 then 1 else 0 end) as no_send_out_goods_num,
  81. sum(order_amount) as order_amount,
  82. sum(regimental_promotion_amount) as regimental_promotion_amount,
  83. sum(total_regimental_settle_amount) as total_regimental_settle_amount,
  84. date_format(stat_date, '%Y%m%d') as stat_date
  85. from {database}.kuaishou_supply_chain
  86. where stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d')
  87. and item_id={item_id}
  88. and stat_date is not null
  89. group by promoter_id, item_id, stat_date
  90. ) b
  91. on a.promoter_id = b.promoter_id and a.item_id = b.item_id
  92. left join(
  93. select promoter_id,
  94. item_id,
  95. count(distinct oid) as valid_order_num,
  96. sum(order_amount) as valid_order_amount,
  97. date_format(stat_date, '%Y%m%d') as stat_date
  98. from {database}.kuaishou_supply_chain
  99. where order_status != 80
  100. and item_id={item_id}
  101. and stat_date >= date_format(date_sub(now(), interval 61 day), '%Y%m%d')
  102. and stat_date is not null
  103. group by promoter_id, item_id, stat_date
  104. ) c
  105. on b.promoter_id = c.promoter_id and b.item_id = c.item_id and b.stat_date=c.stat_date
  106. where
  107. date_format(a.create_time,'%Y%m%d')<=b.stat_date
  108. and b.stat_date is not null
  109. """.format(item_id=item_id, database=database)
  110. try:
  111. print(handler_sql)
  112. MysqlProUtils().Operate(sql=handler_sql)
  113. sleep(10)
  114. except Exception as e:
  115. print(e)
  116. continue
  117. if __name__ == '__main__':
  118. databases = {'rocket', 'miaogousi', 'yufu', 'ruixuan','yuxiu'}
  119. for database in databases:
  120. CanalManagementThread().KuaishouRuleOrderDataHandler(database=database)