run_ylgxxsp_recommend.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. # -*- coding: utf-8 -*-
  2. # @Author: luojunhui
  3. # @Time: 2023/10/23
  4. import argparse
  5. import random
  6. from mq_http_sdk.mq_client import *
  7. from mq_http_sdk.mq_consumer import *
  8. from mq_http_sdk.mq_exception import MQExceptionBase
  9. sys.path.append(os.getcwd())
  10. from common.public import task_fun_mq, get_consumer, ack_message
  11. from common.scheduling_db import MysqlHelper
  12. from common.aliyun_log import AliyunLogger
  13. from youlegaoxiaoxiaoshipin.youlegaoxiaoxiaoshipin_recommend import YLGXXSPScheduling
  14. def main(my_platform, mode, topic_name, group_id, env):
  15. consumer = get_consumer(topic_name, group_id)
  16. # 长轮询表示如果Topic没有消息,则客户端请求会在服务端挂起3秒,3秒内如果有消息可以消费则立即返回响应。
  17. # 长轮询时间3秒(最多可设置为30秒)。
  18. wait_seconds = 30
  19. # 一次最多消费3条(最多可设置为16条)。
  20. batch = 1
  21. AliyunLogger.logging(
  22. code="1000",
  23. platform=my_platform,
  24. mode=mode,
  25. env=env,
  26. message=f'{10 * "="}Consume And Ack Message From Topic{10 * "="}\n'
  27. f"WaitSeconds:{wait_seconds}\n"
  28. f"TopicName:{topic_name}\n"
  29. f"MQConsumer:{group_id}",
  30. )
  31. while True:
  32. try:
  33. # 长轮询消费消息。
  34. recv_msgs = consumer.consume_message(batch, wait_seconds)
  35. for msg in recv_msgs:
  36. AliyunLogger.logging(
  37. code="1000",
  38. platform=my_platform,
  39. mode=mode,
  40. env=env,
  41. message=f"Receive\n"
  42. f"MessageId:{msg.message_id}\n"
  43. f"MessageBodyMD5:{msg.message_body_md5}\n"
  44. f"MessageTag:{msg.message_tag}\n"
  45. f"ConsumedTimes:{msg.consumed_times}\n"
  46. f"PublishTime:{msg.publish_time}\n"
  47. f"Body:{msg.message_body}\n"
  48. f"NextConsumeTime:{msg.next_consume_time}\n"
  49. f"ReceiptHandle:{msg.receipt_handle}\n"
  50. f"Properties:{msg.properties}",
  51. )
  52. # ack_mq_message
  53. ack_message(
  54. log_type=mode,
  55. crawler=my_platform,
  56. recv_msgs=recv_msgs,
  57. consumer=consumer,
  58. )
  59. # 解析 task_dict
  60. task_dict = task_fun_mq(msg.message_body)["task_dict"]
  61. AliyunLogger.logging(
  62. "1000", my_platform, mode, env, f"调度任务:{task_dict}"
  63. )
  64. # 解析 rule_dict
  65. rule_dict = task_fun_mq(msg.message_body)["rule_dict"]
  66. AliyunLogger.logging(
  67. "1000", my_platform, mode, env, f"抓取规则:{rule_dict}\n"
  68. )
  69. # 解析 user_list
  70. task_id = task_dict["id"]
  71. select_user_sql = (
  72. f"""select * from crawler_user_v3 where task_id={task_id}"""
  73. )
  74. user_list = MysqlHelper.get_values(
  75. mode, my_platform, select_user_sql, env, action=""
  76. )
  77. our_uid_list = []
  78. for user in user_list:
  79. our_uid_list.append(user["uid"])
  80. our_uid = random.choice(our_uid_list)
  81. YLGXXSP = YLGXXSPScheduling(
  82. platform=my_platform,
  83. mode=mode,
  84. env=env,
  85. rule_dict=rule_dict,
  86. our_uid=our_uid,
  87. )
  88. for i in range(10):
  89. YLGXXSP.get_videoList(page_id=i + 1)
  90. AliyunLogger.logging(
  91. code="1000",
  92. platform=my_platform,
  93. mode=mode,
  94. env=env,
  95. message="成功抓取完一轮",
  96. )
  97. except MQExceptionBase as err:
  98. # Topic中没有消息可消费。
  99. if err.type == "MessageNotExist":
  100. AliyunLogger.logging(
  101. code="1000",
  102. platform=my_platform,
  103. mode=mode,
  104. env=env,
  105. message=f"No new message! RequestId:{err.req_id}\n",
  106. )
  107. continue
  108. AliyunLogger.logging(
  109. code="1000",
  110. platform=my_platform,
  111. mode=mode,
  112. env=env,
  113. message=f"Consume Message Fail! Exception:{err}\n",
  114. )
  115. time.sleep(2)
  116. continue
  117. if __name__ == "__main__":
  118. parser = argparse.ArgumentParser() ## 新建参数解释器对象
  119. parser.add_argument("--log_type", type=str) ## 添加参数,注明参数类型
  120. parser.add_argument("--crawler") ## 添加参数
  121. parser.add_argument("--topic_name") ## 添加参数
  122. parser.add_argument("--group_id") ## 添加参数
  123. parser.add_argument("--env") ## 添加参数
  124. args = parser.parse_args() ### 参数赋值,也可以通过终端赋值
  125. main(
  126. my_platform=args.crawler,
  127. mode=args.log_type,
  128. topic_name=args.topic_name,
  129. group_id=args.group_id,
  130. env=args.env,
  131. )