piaoquan_crawler/piaopiaoquan/piaopiaoquan_main/run_ppqsift_recommend.py

# -*- coding: utf-8 -*-
# @Time: 2023/12/01
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
import multiprocessing


sys.path.append(os.getcwd())
from common.public import get_consumer, ack_message, task_fun_mq, get_rule_from_mysql
from common.common import Common
from common.scheduling_db import MysqlHelper
from common import AliyunLogger
from piaopiaoquan.piaopiaoquan.piaopiaoquan_sift import PPQSiftRecommend



def run(args1, args2, args3, args4, args5):
    PPQSiftRecommend(
        log_type=args1,
        crawler=args2,
        env=args3,
        rule_dict=args4,
        our_uid=args5
    )


def main(log_type, crawler, topic_name, group_id, env):
    topic_name = "ppqsift_recommend_prod"
    group_id = "ppqsift_recommend_prod"
    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}')
    AliyunLogger.logging(
        code="1000",
        platform=log_type,
        mode=crawler,
        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}")
                AliyunLogger.logging(
                    code="1000",
                    platform=log_type,
                    mode=crawler,
                    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)
                AliyunLogger.logging(
                    "1000", log_type, crawler, env, f"调度任务:{task_dict}"
                )
                # 解析 rule_dict
                rule_dict = task_fun_mq(msg.message_body)["rule_dict"]
                AliyunLogger.logging(
                    "1000", log_type, crawler, env, f"抓取规则:{rule_dict}\n"
                )
                Common.logger(log_type, crawler).info(f"用户列表:{user_list}\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

                process = multiprocessing.Process(
                    target=run,
                    args=(log_type, crawler, env, r_d, our_uid)
                )
                process.start()
                AliyunLogger.logging(
                    code="1003",
                    platform=log_type,
                    mode=crawler,
                    env=env,
                    message="成功获取信息，启动爬虫，开始一轮抓取",
                )
                print("进程开始")
                while True:
                    if not process.is_alive():
                        print("正在重启")
                        process.terminate()
                        os.system("adb forward --remove-all")
                        time.sleep(60)
                        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}")
                        process = multiprocessing.Process(target=run, args=(log_type, crawler, env, r_d, our_uid))
                        process.start()
                        AliyunLogger.logging(
                            code="1004",
                            platform=log_type,
                            mode=crawler,
                            env=env,
                            message="成功抓取完一轮",
                        )
                    time.sleep(60)
        except MQExceptionBase as err:
            # Topic中没有消息可消费。
            if err.type == "MessageNotExist":
                AliyunLogger.logging(
                    code="1000",
                    platform=log_type,
                    mode=crawler,
                    env=env,
                    message=f"No new message! RequestId:{err.req_id}\n",
                )
                Common.logging(
                    log_type=log_type,
                    crawler=crawler,
                    env=env,
                    message=f"No new message! RequestId:{err.req_id}\n",
                )
                continue
            AliyunLogger.logging(
                code="1000",
                platform=log_type,
                mode=crawler,
                env=env,
                message=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)