# -*- coding: utf-8 -*- # @Author: luojunhui # @Time: 2023/10/19 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.common import Common from common.public import get_consumer, ack_message, task_fun_mq, get_rule_from_mysql from common.scheduling_db import MysqlHelper from haitunzhufu.haitunzhufu_recommend import HTZFScheduling from common import AliyunLogger class HTZFMain: @classmethod def main(cls, 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}') AliyunLogger.logging( code="1000", platform=crawler, mode=log_type, env=env, message=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: 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(log_type, 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}") AliyunLogger.logging( code="1000", platform=crawler, mode=log_type, env=env, message=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}") AliyunLogger.logging( "1000", crawler, log_type, env, f"调度任务:{task_dict}" ) Common.logger(log_type, crawler).info(f"用户列表:{user_list}\n") AliyunLogger.logging( "1000", crawler, log_type, env, f"用户列表:{user_list}\n" ) Common.logger(log_type, crawler).info(f'开始抓取:{task_dict["taskName"]}\n') AliyunLogger.logging( "1000", crawler, log_type, env, f'开始抓取:{task_dict["taskName"]}\n' ) new_r = get_rule_from_mysql(task_id=task_id, log_type=log_type, crawler=crawler, env=env) r_d = {} for item in new_r: for k, val in item.items(): r_d[k] = val Common.logger(log_type, crawler).info(f"抓取规则:{r_d}") AliyunLogger.logging( "1000", crawler, log_type, env, f"抓取规则:{r_d}" ) AliyunLogger.logging( code="1003", platform=crawler, mode=log_type, env=env, message="成功获取信息，启动爬虫，开始一轮抓取", ) # 初始化 HTZF = HTZFScheduling( log_type=log_type, crawler=crawler, env=env, rule_dict=r_d, our_uid=our_uid ) for i in range(100): if HTZF.download_count >= int(rule_dict.get("videos_cnt", {}).get("min", 10)): HTZF.download_count = 0 break else: HTZF.get_videoList(page_id=i + 1) time.sleep(60) Common.logger(log_type, crawler).info('抓取一轮结束\n') AliyunLogger.logging( code="1004", platform=crawler, mode=log_type, env=env, message="成功抓取完一轮", ) except MQExceptionBase as err: # Topic中没有消息可消费。 if err.type == "MessageNotExist": Common.logger(log_type, crawler).info(f"No new message! RequestId:{err.req_id}\n") continue Common.logger(log_type, crawler).info(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() ### 参数赋值，也可以通过终端赋值 HTZFMain.main(log_type=args.log_type, crawler=args.crawler, topic_name=args.topic_name, group_id=args.group_id, env=args.env)