本山祝福老代码

benshanzhufu/benshanzhufu_main/run_bszf_recommend.py

@@ -1,116 +1,116 @@
-# -*- coding: utf-8 -*-
-# @Author: wangkun
-# @Time: 2023/6/7
-import argparse
-import random
-from mq_http_sdk.mq_client import *
-from mq_http_sdk.mq_consumer import *
-from mq_http_sdk.mq_exception import MQExceptionBase
-sys.path.append(os.getcwd())
-from common.public import get_consumer, ack_message, task_fun_mq
-from common.common import Common
-from common.scheduling_db import MysqlHelper
-from benshanzhufu.benshanzhufu_recommend.benshanzhufu_recommend_scheduling import BenshanzhufuRecommend
-
-
-def main(log_type, crawler, topic_name, group_id, env):
-    consumer = get_consumer(topic_name, group_id)
-    # 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3秒，3秒内如果有消息可以消费则立即返回响应。
-    # 长轮询时间3秒（最多可设置为30秒）。
-    wait_seconds = 30
-    # 一次最多消费3条（最多可设置为16条）。
-    batch = 1
-    Common.logger(log_type, crawler).info(f'{10 * "="}Consume And Ack Message From Topic{10 * "="}\n'
-                                          f'WaitSeconds:{wait_seconds}\n'
-                                          f'TopicName:{topic_name}\n'
-                                          f'MQConsumer:{group_id}')
-    Common.logging(log_type, crawler, env, f'{10 * "="}Consume And Ack Message From Topic{10 * "="}\n'
-                                            f'WaitSeconds:{wait_seconds}\n'
-                                            f'TopicName:{topic_name}\n'
-                                            f'MQConsumer:{group_id}')
-    while True:
-        try:
-            # 长轮询消费消息。
-            recv_msgs = consumer.consume_message(batch, wait_seconds)
-            for msg in recv_msgs:
-                benshanzhufu_start_time = int(time.time())
-                Common.logger(log_type, crawler).info(f"Receive\n"
-                                                      f"MessageId:{msg.message_id}\n"
-                                                      f"MessageBodyMD5:{msg.message_body_md5}\n"
-                                                      f"MessageTag:{msg.message_tag}\n"
-                                                      f"ConsumedTimes:{msg.consumed_times}\n"
-                                                      f"PublishTime:{msg.publish_time}\n"
-                                                      f"Body:{msg.message_body}\n"
-                                                      f"NextConsumeTime:{msg.next_consume_time}\n"
-                                                      f"ReceiptHandle:{msg.receipt_handle}\n"
-                                                      f"Properties:{msg.properties}")
-                Common.logging(logging, crawler, env, f"Receive\n"
-                                                      f"MessageId:{msg.message_id}\n"
-                                                      f"MessageBodyMD5:{msg.message_body_md5}\n"
-                                                      f"MessageTag:{msg.message_tag}\n"
-                                                      f"ConsumedTimes:{msg.consumed_times}\n"
-                                                      f"PublishTime:{msg.publish_time}\n"
-                                                      f"Body:{msg.message_body}\n"
-                                                      f"NextConsumeTime:{msg.next_consume_time}\n"
-                                                      f"ReceiptHandle:{msg.receipt_handle}\n"
-                                                      f"Properties:{msg.properties}")
-                # ack_mq_message
-                ack_message(log_type=log_type, crawler=crawler, recv_msgs=recv_msgs, consumer=consumer)
-
-                # 处理爬虫业务
-                task_dict = task_fun_mq(msg.message_body)['task_dict']
-                rule_dict = task_fun_mq(msg.message_body)['rule_dict']
-                task_id = task_dict['id']
-                select_user_sql = f"""select * from crawler_user_v3 where task_id={task_id}"""
-                user_list = MysqlHelper.get_values(log_type, crawler, select_user_sql, env, action="")
-                our_uid_list = []
-                for user in user_list:
-                    our_uid_list.append(user["uid"])
-                our_uid = random.choice(our_uid_list)
-                Common.logger(log_type, crawler).info(f"调度任务:{task_dict}")
-                Common.logging(log_type, crawler, env, f"调度任务:{task_dict}")
-                Common.logger(log_type, crawler).info(f"抓取规则:{rule_dict}")
-                Common.logging(log_type, crawler, env, f"抓取规则:{rule_dict}")
-                Common.logger(log_type, crawler).info(f"用户列表:{user_list}\n")
-                Common.logging(log_type, crawler, env, f"用户列表:{user_list}\n")
-                Common.logger(log_type, crawler).info(f'开始抓取:{task_dict["taskName"]}\n')
-                Common.logging(log_type, crawler, env, f'开始抓取:{task_dict["taskName"]}\n')
-                BenshanzhufuRecommend.get_videoList(log_type=log_type,
-                                                    crawler=crawler,
-                                                    our_uid=our_uid,
-                                                    rule_dict=rule_dict,
-                                                    env=env)
-                # Common.del_logs(log_type, crawler)
-                Common.logger(log_type, crawler).info('抓取一轮结束')
-                Common.logging(log_type, crawler, env, '抓取一轮结束')
-                benshanzhufu_end_time = int(time.time())
-                benshanzhufu_duration = benshanzhufu_end_time - benshanzhufu_start_time
-                Common.logger(log_type, crawler).info(f"duration {benshanzhufu_duration}\n")
-                Common.logging(log_type, crawler, env, f"duration {benshanzhufu_duration}\n")
-
-        except MQExceptionBase as err:
-            # Topic中没有消息可消费。
-            if err.type == "MessageNotExist":
-                Common.logger(log_type, crawler).info(f"No new message! RequestId:{err.req_id}\n")
-                Common.logging(log_type, crawler, env, f"No new message! RequestId:{err.req_id}\n")
-                continue
-
-            Common.logger(log_type, crawler).info(f"Consume Message Fail! Exception:{err}\n")
-            Common.logging(log_type, crawler, env, f"Consume Message Fail! Exception:{err}\n")
-            time.sleep(2)
-            continue
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser()  # 新建参数解释器对象
-    parser.add_argument('--log_type', type=str)  # 添加参数,注明参数类型
-    parser.add_argument('--crawler')  # 添加参数
-    parser.add_argument('--topic_name')  # 添加参数
-    parser.add_argument('--group_id')  # 添加参数
-    parser.add_argument('--env')  # 添加参数
-    args = parser.parse_args()  # 参数赋值，也可以通过终端赋值
-    main(log_type=args.log_type,
-         crawler=args.crawler,
-         topic_name=args.topic_name,
-         group_id=args.group_id,
-         env=args.env)
+# # -*- coding: utf-8 -*-
+# # @Author: wangkun
+# # @Time: 2023/6/7
+# import argparse
+# import random
+# from mq_http_sdk.mq_client import *
+# from mq_http_sdk.mq_consumer import *
+# from mq_http_sdk.mq_exception import MQExceptionBase
+# sys.path.append(os.getcwd())
+# from common.public import get_consumer, ack_message, task_fun_mq
+# from common.common import Common
+# from common.scheduling_db import MysqlHelper
+# from benshanzhufu.benshanzhufu_recommend.benshanzhufu_recommend_scheduling import BenshanzhufuRecommend
+#
+#
+# def main(log_type, crawler, topic_name, group_id, env):
+#     consumer = get_consumer(topic_name, group_id)
+#     # 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3秒，3秒内如果有消息可以消费则立即返回响应。
+#     # 长轮询时间3秒（最多可设置为30秒）。
+#     wait_seconds = 30
+#     # 一次最多消费3条（最多可设置为16条）。
+#     batch = 1
+#     Common.logger(log_type, crawler).info(f'{10 * "="}Consume And Ack Message From Topic{10 * "="}\n'
+#                                           f'WaitSeconds:{wait_seconds}\n'
+#                                           f'TopicName:{topic_name}\n'
+#                                           f'MQConsumer:{group_id}')
+#     Common.logging(log_type, crawler, env, f'{10 * "="}Consume And Ack Message From Topic{10 * "="}\n'
+#                                             f'WaitSeconds:{wait_seconds}\n'
+#                                             f'TopicName:{topic_name}\n'
+#                                             f'MQConsumer:{group_id}')
+#     while True:
+#         try:
+#             # 长轮询消费消息。
+#             recv_msgs = consumer.consume_message(batch, wait_seconds)
+#             for msg in recv_msgs:
+#                 benshanzhufu_start_time = int(time.time())
+#                 Common.logger(log_type, crawler).info(f"Receive\n"
+#                                                       f"MessageId:{msg.message_id}\n"
+#                                                       f"MessageBodyMD5:{msg.message_body_md5}\n"
+#                                                       f"MessageTag:{msg.message_tag}\n"
+#                                                       f"ConsumedTimes:{msg.consumed_times}\n"
+#                                                       f"PublishTime:{msg.publish_time}\n"
+#                                                       f"Body:{msg.message_body}\n"
+#                                                       f"NextConsumeTime:{msg.next_consume_time}\n"
+#                                                       f"ReceiptHandle:{msg.receipt_handle}\n"
+#                                                       f"Properties:{msg.properties}")
+#                 Common.logging(logging, crawler, env, f"Receive\n"
+#                                                       f"MessageId:{msg.message_id}\n"
+#                                                       f"MessageBodyMD5:{msg.message_body_md5}\n"
+#                                                       f"MessageTag:{msg.message_tag}\n"
+#                                                       f"ConsumedTimes:{msg.consumed_times}\n"
+#                                                       f"PublishTime:{msg.publish_time}\n"
+#                                                       f"Body:{msg.message_body}\n"
+#                                                       f"NextConsumeTime:{msg.next_consume_time}\n"
+#                                                       f"ReceiptHandle:{msg.receipt_handle}\n"
+#                                                       f"Properties:{msg.properties}")
+#                 # ack_mq_message
+#                 ack_message(log_type=log_type, crawler=crawler, recv_msgs=recv_msgs, consumer=consumer)
+#
+#                 # 处理爬虫业务
+#                 task_dict = task_fun_mq(msg.message_body)['task_dict']
+#                 rule_dict = task_fun_mq(msg.message_body)['rule_dict']
+#                 task_id = task_dict['id']
+#                 select_user_sql = f"""select * from crawler_user_v3 where task_id={task_id}"""
+#                 user_list = MysqlHelper.get_values(log_type, crawler, select_user_sql, env, action="")
+#                 our_uid_list = []
+#                 for user in user_list:
+#                     our_uid_list.append(user["uid"])
+#                 our_uid = random.choice(our_uid_list)
+#                 Common.logger(log_type, crawler).info(f"调度任务:{task_dict}")
+#                 Common.logging(log_type, crawler, env, f"调度任务:{task_dict}")
+#                 Common.logger(log_type, crawler).info(f"抓取规则:{rule_dict}")
+#                 Common.logging(log_type, crawler, env, f"抓取规则:{rule_dict}")
+#                 Common.logger(log_type, crawler).info(f"用户列表:{user_list}\n")
+#                 Common.logging(log_type, crawler, env, f"用户列表:{user_list}\n")
+#                 Common.logger(log_type, crawler).info(f'开始抓取:{task_dict["taskName"]}\n')
+#                 Common.logging(log_type, crawler, env, f'开始抓取:{task_dict["taskName"]}\n')
+#                 BenshanzhufuRecommend.get_videoList(log_type=log_type,
+#                                                     crawler=crawler,
+#                                                     our_uid=our_uid,
+#                                                     rule_dict=rule_dict,
+#                                                     env=env)
+#                 # Common.del_logs(log_type, crawler)
+#                 Common.logger(log_type, crawler).info('抓取一轮结束')
+#                 Common.logging(log_type, crawler, env, '抓取一轮结束')
+#                 benshanzhufu_end_time = int(time.time())
+#                 benshanzhufu_duration = benshanzhufu_end_time - benshanzhufu_start_time
+#                 Common.logger(log_type, crawler).info(f"duration {benshanzhufu_duration}\n")
+#                 Common.logging(log_type, crawler, env, f"duration {benshanzhufu_duration}\n")
+#
+#         except MQExceptionBase as err:
+#             # Topic中没有消息可消费。
+#             if err.type == "MessageNotExist":
+#                 Common.logger(log_type, crawler).info(f"No new message! RequestId:{err.req_id}\n")
+#                 Common.logging(log_type, crawler, env, f"No new message! RequestId:{err.req_id}\n")
+#                 continue
+#
+#             Common.logger(log_type, crawler).info(f"Consume Message Fail! Exception:{err}\n")
+#             Common.logging(log_type, crawler, env, f"Consume Message Fail! Exception:{err}\n")
+#             time.sleep(2)
+#             continue
+#
+#
+# if __name__ == "__main__":
+#     parser = argparse.ArgumentParser()  # 新建参数解释器对象
+#     parser.add_argument('--log_type', type=str)  # 添加参数,注明参数类型
+#     parser.add_argument('--crawler')  # 添加参数
+#     parser.add_argument('--topic_name')  # 添加参数
+#     parser.add_argument('--group_id')  # 添加参数
+#     parser.add_argument('--env')  # 添加参数
+#     args = parser.parse_args()  # 参数赋值，也可以通过终端赋值
+#     main(log_type=args.log_type,
+#          crawler=args.crawler,
+#          topic_name=args.topic_name,
+#          group_id=args.group_id,
+#          env=args.env)