words_func.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107
  1. import traceback
  2. from db_helper import MysqlHelper
  3. from log import Log
  4. mysql_helper = MysqlHelper()
  5. log_ = Log()
  6. def get_words(page_num, page_size):
  7. """
  8. 分页获取所有热点词
  9. :param page_num: 页码
  10. :param page_size: 每页请求条目数
  11. :return: words
  12. """
  13. try:
  14. sql = f"select id, word from word.hot_word order by id limit {(page_num-1)*page_size}, {page_size};"
  15. data = mysql_helper.get_data(sql=sql)
  16. if data is None:
  17. return None
  18. words = []
  19. for id_, word in data:
  20. words.append({'id': id_, 'word': word})
  21. return words
  22. except Exception as e:
  23. log_.error(traceback.format_exc())
  24. return None
  25. def update_wechat_score_data(data):
  26. """
  27. 根据爬取到的微信指数数据更新数据库
  28. :param data:
  29. :return:
  30. """
  31. if data is None or len(data) == 0:
  32. log_.info(f"无需要更新的数据!")
  33. return
  34. # 爬取数据解析
  35. log_.info(f"data count = {len(data)}")
  36. wechat_score_data_list = []
  37. for item in data:
  38. word_id = item.get('id')
  39. word = item.get('word')
  40. wechat_scores = item.get('wechatScores')
  41. if wechat_scores is None or len(wechat_scores) == 0:
  42. continue
  43. for score_data in wechat_scores:
  44. score = score_data.get('score')
  45. score_date = score_data.get('scoreDate')
  46. wechat_score_data_list.append({'word_id': word_id, 'word': word, 'score': score, 'score_date': score_date})
  47. log_.info(f"wechat_score_data_list count = {len(wechat_score_data_list)}")
  48. # update or insert 数据区分
  49. update_data = []
  50. insert_data = []
  51. for wechat_score_data in wechat_score_data_list:
  52. select_sql = f"SELECT id FROM word.word_wechat_score " \
  53. f"WHERE word_id = {wechat_score_data['word_id']} " \
  54. f"AND score_date = '{wechat_score_data['score_date']}';"
  55. res = mysql_helper.get_data(sql=select_sql)
  56. if res is None:
  57. continue
  58. if len(res) == 0:
  59. insert_data.append(wechat_score_data)
  60. else:
  61. id_ = res[0][0]
  62. wechat_score_data['id'] = id_
  63. update_data.append(wechat_score_data)
  64. log_.info(f"update_data = {len(update_data)}")
  65. log_.info(f"insert_data = {len(insert_data)}")
  66. # 批量插入
  67. if len(insert_data) > 0:
  68. for i in range(len(insert_data) // 100 + 1):
  69. log_.info(f"insert i = {i}")
  70. insert_temp_data = insert_data[i * 100:(i + 1) * 100]
  71. if len(insert_temp_data) > 0:
  72. insert_sql_values = ', '.join([f"({item['word_id']}, {item['score']}, '{item['score_date']}')"
  73. for item in insert_temp_data])
  74. insert_sql = f"insert into word.word_wechat_score (word_id, wechat_score, score_date) " \
  75. f"values {insert_sql_values};"
  76. mysql_helper.add_data(sql=insert_sql)
  77. log_.info(f"insert wechat score data finished! insert count = {len(insert_data)}")
  78. # 批量更新
  79. if len(update_data) > 0:
  80. for i in range(len(update_data) // 100 + 1):
  81. log_.info(f"update i = {i}")
  82. update_temp_data = update_data[i * 100:(i + 1) * 100]
  83. if len(update_temp_data) > 0:
  84. update_id = [item['id'] for item in update_temp_data]
  85. update_sql_values = ' '.join([f"when {item['id']} then {item['score']}" for item in update_temp_data])
  86. if len(update_id) > 1:
  87. update_sql = f"update word.word_wechat_score set wechat_score = " \
  88. f"case id {update_sql_values} end where id in {tuple(update_id)};"
  89. else:
  90. update_sql = f"update word.word_wechat_score set wechat_score = " \
  91. f"case id {update_sql_values} end where id in ({update_id[0]});"
  92. mysql_helper.add_data(sql=update_sql)
  93. log_.info(f"update wechat score data finished! update count = {len(update_data)}")
  94. if __name__ == '__main__':
  95. get_words(8, 100)
  96. # get_words(1, 20)
  97. # get_words(2, 10)