修改 祝福圈子、小年糕+日志

xiaoniangaoplus/xiaoniangaoplus_main/run_xngplus_recommend.py

@@ -8,12 +8,14 @@ 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 xiaoniangaoplus.xiaoniangaoplus.xiaoniangao_plus_scheduling import XiaoNianGaoPlusRecommend
 from xiaoniangaoplus.xiaoniangaoplus.xiaoniangao_plus_scheduling2 import XiaoNianGaoPlusRecommend
+from common import AliyunLogger
 
 
 def run(args1, args2, args3, args4, args5):
@@ -39,10 +41,16 @@ def main(log_type, crawler, topic_name, group_id, env):
                                           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=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:
             # 长轮询消费消息。
@@ -59,16 +67,22 @@ def main(log_type, crawler, topic_name, group_id, env):
                                                       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=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)
 
@@ -82,28 +96,34 @@ def main(log_type, crawler, topic_name, group_id, env):
                 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}")
+                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")
-                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)
-                # Common.logger(log_type, crawler).info(f'rule_dict:{new_r}\n')
                 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="1003",
+                    platform=log_type,
+                    mode=crawler,
+                    env=env,
+                    message="成功获取信息，启动爬虫，开始一轮抓取",
+                )
                 print("进程开始")
-
                 while True:
                     if not process.is_alive():
                         print("正在重启")
@@ -119,7 +139,15 @@ def main(log_type, crawler, topic_name, group_id, env):
                         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)
+
                 # XiaoNianGaoPlusRecommend.start_wechat(log_type=log_type,
                 #                                       crawler=crawler,
                 #                                       rule_dict=rule_dict,
@@ -135,12 +163,27 @@ def main(log_type, crawler, topic_name, group_id, env):
         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")
+                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
-
-            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")
+            AliyunLogger.logging(
+                code="1000",
+                platform=log_type,
+                mode=crawler,
+                env=env,
+                message=f"Consume Message Fail! Exception:{err}\n",
+            )
             time.sleep(2)
             continue
 

zhufuquanzi/zhufuquanzi_main/run_zfqz_recommend.py

@@ -10,11 +10,13 @@ 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
 from common.scheduling_db import MysqlHelper
 from zhufuquanzi.zhufuquanzi_recommend.zhufuquanzi_recommend_new import ZFQZRecommend
+from common import AliyunLogger
 
 
 def run(args1, args2, args3, args4, args5):
@@ -40,10 +42,16 @@ class ZFQZMain:
                                               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=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:
                 # 长轮询消费消息。
@@ -59,16 +67,22 @@ class ZFQZMain:
                                                           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=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)
 
@@ -83,19 +97,28 @@ class ZFQZMain:
                         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}")
+                    AliyunLogger.logging(
+                        "1000", 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}")
+                    AliyunLogger.logging(
+                        "1000", log_type, crawler, env, f"抓取规则:{rule_dict}\n"
+                    )
                     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')
 
                     process = multiprocessing.Process(
                         target=run,
                         args=(log_type, crawler, rule_dict, our_uid, env)
                     )
                     process.start()
+                    AliyunLogger.logging(
+                        code="1003",
+                        platform=log_type,
+                        mode=crawler,
+                        env=env,
+                        message="成功获取信息，启动爬虫，开始一轮抓取",
+                    )
                     print("进程开始")
 
                     for i in range(10):
@@ -110,6 +133,13 @@ class ZFQZMain:
                                 args=(log_type, crawler, rule_dict, our_uid, env)
                             )
                             process.start()
+                            AliyunLogger.logging(
+                                code="1004",
+                                platform=log_type,
+                                mode=crawler,
+                                env=env,
+                                message="成功抓取完一轮",
+                            )
                         time.sleep(60)
 
                     # # 抓取符合规则的视频列表