piaoquan_crawler/ganggangdouchuan/ganggangdouchuan_main/run_ggdc_recommend.py

# -*- coding: utf-8 -*-
# @Author: luojunhui
# @Time: 2023/10/20
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 ganggangdouchuan.ganggangdouchuan_recommend import GGDCScheduling


class GGDCMain:
    @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}",
        )
        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_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'
                    )
                    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}")
                    Common.logging(log_type, crawler, env, f"抓取规则:{r_d}")
                    # 初始化
                    GGDC = GGDCScheduling(
                        log_type=log_type,
                        crawler=crawler,
                        env=env,
                        rule_dict=r_d,
                        our_uid=our_uid,
                    )
                    for i in range(20):
                        if GGDC.download_count >= int(
                            rule_dict.get("videos_cnt", {}).get("min", 10)
                        ):
                            GGDC.download_count = 0
                            break
                        else:
                            GGDC.get_videoList(page_id=i + 1)
                            time.sleep(60)
                    Common.logger(log_type, crawler).info("抓取一轮结束\n")
                    Common.logging(log_type, crawler, env, "抓取一轮结束\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()  ### 参数赋值，也可以通过终端赋值
    GGDCMain.main(
        log_type=args.log_type,
        crawler=args.crawler,
        topic_name=args.topic_name,
        group_id=args.group_id,
        env=args.env,
    )