众妙音信——BUG

zhongmiaoyinxin/zhongmiaoyinxin_main/run_zmyx_recommend.py

@@ -1,142 +1,25 @@
 # -*- coding: utf-8 -*-
-# @Author: luojunhui
-# @Time: 2023/10/23
+# @Author: 罗俊辉
+# @Time: 2023/4/17
 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 os
+import sys
 sys.path.append(os.getcwd())
-from common.public import task_fun_mq, get_consumer, ack_message
-from common.scheduling_db import MysqlHelper
-from common.aliyun_log import AliyunLogger
+from common.common import Common
 from zhongmiaoyinxin.zhongmiaoyinxin_recommend import ZMYXRecommend
 
 
-def main(my_platform, mode, topic_name, group_id, env):
-    consumer = get_consumer(topic_name, group_id)
-    # 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3秒，3秒内如果有消息可以消费则立即返回响应。
-    # 长轮询时间3秒（最多可设置为30秒）。
-    wait_seconds = 30
-    # 一次最多消费3条（最多可设置为16条）。
-    batch = 1
-    AliyunLogger.logging(
-        code="1000",
-        platform=my_platform,
-        mode=mode,
-        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:
-                AliyunLogger.logging(
-                    code="1000",
-                    platform=my_platform,
-                    mode=mode,
-                    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=mode,
-                    crawler=my_platform,
-                    recv_msgs=recv_msgs,
-                    consumer=consumer,
-                )
-                # 解析 task_dict
-                task_dict = task_fun_mq(msg.message_body)["task_dict"]
-                AliyunLogger.logging(
-                    "1000", my_platform, mode, env, f"调度任务:{task_dict}"
-                )
-                # 解析 rule_dict
-                rule_dict = task_fun_mq(msg.message_body)["rule_dict"]
-                AliyunLogger.logging(
-                    "1000", my_platform, mode, 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(
-                    mode, my_platform, 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(
-                    code="1003",
-                    platform=my_platform,
-                    mode=mode,
-                    env=env,
-                    message="成功获取信息，启动爬虫，开始一轮抓取",
-                )
-                ZMYX = ZMYXRecommend(
-                    crawler=my_platform,
-                    log_type=mode,
-                    env=env,
-                    rule_dict=rule_dict,
-                    our_uid=our_uid,
-                )
-                AliyunLogger.logging(
-                    code="1004",
-                    platform=my_platform,
-                    mode=mode,
-                    env=env,
-                    message="成功抓取完一轮",
-                )
-
-        except MQExceptionBase as err:
-            # Topic中没有消息可消费。
-            if err.type == "MessageNotExist":
-                AliyunLogger.logging(
-                    code="1000",
-                    platform=my_platform,
-                    mode=mode,
-                    env=env,
-                    message=f"No new message! RequestId:{err.req_id}\n",
-                )
-                continue
-            AliyunLogger.logging(
-                code="1000",
-                platform=my_platform,
-                mode=mode,
-                env=env,
-                message=f"Consume Message Fail! Exception:{err}\n",
-            )
-            time.sleep(2)
-            continue
+def main(log_type, crawler, env):
+    Common.logger(log_type, crawler).info('开始抓取:众妙音信小程序\n')
+    ZMYXRecommend(log_type, crawler, env, our_uid="64624913", rule_dict={})
+    Common.del_logs(log_type, crawler)
+    Common.logger(log_type, crawler).info('抓取完一轮\n')
 
 
 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")  ## 添加参数
+    parser.add_argument('--log_type', type=str)  ## 添加参数,注明参数类型
+    parser.add_argument('--crawler')  ## 添加参数
+    parser.add_argument('--env')  ## 添加参数
     args = parser.parse_args()  ### 参数赋值，也可以通过终端赋值
-    main(
-        my_platform=args.crawler,
-        mode=args.log_type,
-        topic_name=args.topic_name,
-        group_id=args.group_id,
-        env=args.env,
-    )
+    main(log_type=args.log_type, crawler=args.crawler, env=args.env)

zhongmiaoyinxin/zhongmiaoyinxin_main/run_zmyx_recommend_online.py

@@ -0,0 +1,156 @@
+# -*- coding: utf-8 -*-
+# @Author: luojunhui
+# @Time: 2023/10/23
+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.public import task_fun_mq, get_consumer, ack_message
+from common.scheduling_db import MysqlHelper
+from common.aliyun_log import AliyunLogger
+from common.common import Common
+from zhongmiaoyinxin.zhongmiaoyinxin_recommend import ZMYXRecommend
+
+
+def main(my_platform, mode, topic_name, group_id, env):
+    consumer = get_consumer(topic_name, group_id)
+    # 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3秒，3秒内如果有消息可以消费则立即返回响应。
+    # 长轮询时间3秒（最多可设置为30秒）。
+    wait_seconds = 30
+    # 一次最多消费3条（最多可设置为16条）。
+    batch = 1
+    AliyunLogger.logging(
+        code="1000",
+        platform=my_platform,
+        mode=mode,
+        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)
+            print("consume_success")
+            for msg in recv_msgs:
+                AliyunLogger.logging(
+                    code="1000",
+                    platform=my_platform,
+                    mode=mode,
+                    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=mode,
+                    crawler=my_platform,
+                    recv_msgs=recv_msgs,
+                    consumer=consumer,
+                )
+                print(msg.message_body)
+                # 解析 task_dict
+                Common.logger(log_type=mode, crawler=my_platform).info("开始解析")
+                task_dict = task_fun_mq(msg.message_body)["task_dict"]
+                Common.logger(log_type=mode, crawler=my_platform).info("解析完成")
+                AliyunLogger.logging(
+                    "1000", my_platform, mode, env, f"调度任务:{task_dict}"
+                )
+                # 解析 rule_dict
+                rule_dict = task_fun_mq(msg.message_body)["rule_dict"]
+                AliyunLogger.logging(
+                    "1000", my_platform, mode, 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(
+                    mode, my_platform, 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=mode, crawler=my_platform).info("id_list", our_uid_list)
+                AliyunLogger.logging(
+                    code="1003",
+                    platform=my_platform,
+                    mode=mode,
+                    env=env,
+                    message="成功获取信息，启动爬虫，开始一轮抓取",
+                )
+                ZMYX = ZMYXRecommend(
+                    crawler=my_platform,
+                    log_type=mode,
+                    env=env,
+                    rule_dict=rule_dict,
+                    our_uid=our_uid
+                )
+                AliyunLogger.logging(
+                    code="1004",
+                    platform=my_platform,
+                    mode=mode,
+                    env=env,
+                    message="成功抓取完一轮",
+                )
+
+        except MQExceptionBase as err:
+            # Topic中没有消息可消费。
+            if err.type == "MessageNotExist":
+                AliyunLogger.logging(
+                    code="1000",
+                    platform=my_platform,
+                    mode=mode,
+                    env=env,
+                    message=f"No new message! RequestId:{err.req_id}\n",
+                )
+                continue
+            AliyunLogger.logging(
+                code="1000",
+                platform=my_platform,
+                mode=mode,
+                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(
+        my_platform=args.crawler,
+        mode=args.log_type,
+        topic_name=args.topic_name,
+        group_id=args.group_id,
+        env=args.env,
+    )
+    # main(
+    #     my_platform="zhongmiaoyinxin",
+    #     mode="recommend",
+    #     topic_name="zmyx",
+    #     group_id="zmyx",
+    #     env="prod"
+    # )

zhongmiaoyinxin/zhongmiaoyinxin_recommend/zhongmiaoyinxin_recommend_new.py

@@ -11,7 +11,6 @@ from hashlib import md5
 from appium import webdriver
 from appium.webdriver.common.touch_action import TouchAction
 from appium.webdriver.extensions.android.nativekey import AndroidKey
-from appium.webdriver.webdriver import WebDriver
 from selenium.common.exceptions import NoSuchElementException
 from selenium.webdriver.common.by import By
 
@@ -22,11 +21,9 @@ from common.pipeline import PiaoQuanPipeline
 
 
 class ZMYXRecommend:
-    env = None
-    driver = None
-    log_type = None
 
     def __init__(self, log_type, crawler, env, rule_dict, our_uid):
+        self.mq = None
         self.platform = "zhongmiaoyinxin"
         self.download_cnt = 0
         self.element_list = []
@@ -37,11 +34,10 @@ class ZMYXRecommend:
         self.env = env
         self.rule_dict = rule_dict
         self.our_uid = our_uid
-        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
         if self.env == "dev":
             chromedriverExecutable = "/Users/luojunhui/Downloads/chromedriver_V111/chromedriver"
         else:
-            chromedriverExecutable = '/Users/piaoquan/Downloads/chromedriver'  # Mac 爬虫机器
+            chromedriverExecutable = '/Users/luojunhui/Downloads/chromedriver_V111/chromedriver'  # Mac 爬虫机器
         # 微信的配置文件
         caps = {
             "platformName": "Android",  # 手机操作系统 Android / iOS
@@ -250,6 +246,14 @@ class ZMYXRecommend:
                     return
                 # 获取 trace_id, 并且把该 id 当做视频生命周期唯一索引
                 trace_id = self.crawler + str(uuid.uuid1())
+                AliyunLogger.logging(
+                    code="1001",
+                    platform=self.platform,
+                    mode=self.log_type,
+                    env=self.env,
+                    trace_id=trace_id,
+                    message="扫描到一条视频",
+                )
                 self.download_cnt += 1
                 self.search_elements('//wx-view[@class="cover"]')
                 time.sleep(3)
@@ -323,16 +327,4 @@ def run():
 
 if __name__ == "__main__":
     run()
-    # process = multiprocessing.Process(
-    #     target=run
-    # )
-    # process.start()
-    # while True:
-    #     if not process.is_alive():
-    #         print("正在重启")
-    #         process.terminate()
-    #         time.sleep(60)
-    #         os.system("adb forward --remove-all")
-    #         process = multiprocessing.Process(target=run)
-    #         process.start()
-    #     time.sleep(60)
+