piaoquan_crawler/gongzhonghao/gongzhonghao_main/run_gzh_author.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/6/30
import argparse
from multiprocessing import Process
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 task_fun_mq, get_consumer, ack_message
from common.common import Common
from common.scheduling_db import MysqlHelper
from gongzhonghao.gongzhonghao_author.gongzhonghao_author import GongzhonghaoAuthor


def get_author_videos(log_type, crawler, token_index, task_dict, rule_dict, user_list, env):
    Common.logger(log_type, crawler).info(f'开始抓取:{task_dict["taskName"]}\n')
    Common.logging(log_type, crawler, env, f'开始抓取:{task_dict["taskName"]}\n')
    Common.logger(log_type, crawler).info(f"user_list:{user_list}")
    Common.logging(log_type, crawler, env, f"user_list:{user_list}")
    GongzhonghaoAuthor.get_all_videos(log_type=log_type,
                                      crawler=crawler,
                                      task_dict=task_dict,
                                      token_index=token_index,
                                      rule_dict=rule_dict,
                                      user_list=user_list,
                                      env=env)
    Common.del_logs(log_type, crawler)
    Common.logger(log_type, crawler).info('抓取一轮结束\n')
    Common.logging(log_type, crawler, env, '抓取一轮结束\n')


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:
                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}")
                # ack_mq_message
                ack_message(log_type=log_type, crawler=crawler, recv_msgs=recv_msgs, consumer=consumer)

                # 解析 task_dict
                task_dict = task_fun_mq(msg.message_body)['task_dict']
                Common.logger(log_type, crawler).info(f"调度任务:{task_dict}")
                Common.logging(log_type, crawler, env, f"调度任务:{task_dict}")

                # 解析 rule_dict
                rule_dict = task_fun_mq(msg.message_body)['rule_dict']
                Common.logger(log_type, crawler).info(f"抓取规则:{rule_dict}\n")
                Common.logging(log_type, crawler, env, f"抓取规则:{rule_dict}\n")

                # 解析 user_list
                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="")

                # 计算启动脚本数 crawler_num
                user_num = len(user_list)
                chunk_size = 100  # 每个进程处理的用户数量
                crawler_num = int(user_num // chunk_size)  # 向下取整
                if user_num % chunk_size != 0:
                    crawler_num += 1
                Common.logger(log_type, crawler).info(f"共{user_num}个公众号，需要启动{crawler_num}个脚本任务")
                Common.logging(log_type, crawler, env, f"共{user_num}个公众号，需要启动{crawler_num}个脚本任务")

                # 多进程并行抓取
                processes = []
                for i in range(crawler_num):
                    start = i * chunk_size
                    end = min((i + 1) * chunk_size, user_num + 1)
                    process = Process(target=get_author_videos, args=(
                    f"{log_type}{i + 1}", crawler, i + 1, task_dict, rule_dict, user_list[start:end], env))
                    process.start()
                    processes.append(process)

                for process in processes:
                    process.join()

        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)