KuaishouRuleOrderDataTherad.py 5.0 KB

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