线下爬虫，视频刷刷——上线

application/config/__init__.py

@@ -1,2 +1,3 @@
 from .ipconfig import ip_config
-from .mysql_config import env_dict
+from .mysql_config import env_dict
+from .topic_group_queue import TopicGroup

application/config/topic_group_queue.py

@@ -0,0 +1,18 @@
+class TopicGroup(object):
+    def __init__(self):
+        self.spider_list = [
+            ("zwwfs", "recommend"),
+            ("zchqs", "recommend"),
+        ]
+
+    def produce(self):
+        result = [
+            {
+                "topic": "{}_{}_prod".format(i[0], i[1]),
+                "group": "{}_{}_prod".format(i[0], i[1])
+            } for i in self.spider_list
+        ]
+        return result
+
+
+# print(TopicGroup().produce())

application/spider/crawler_offline/__init__.py

@@ -1 +1,2 @@
-from .xiaoniangao_plus import XiaoNianGaoPlusRecommend
+from .xiaoniangao_plus import XiaoNianGaoPlusRecommend
+from .shipinshuashua import SPSSRecommend

application/spider/crawler_offline/shipinshuashua.py

@@ -0,0 +1,333 @@
+# -*- coding: utf-8 -*-
+# @Author: luojunhui
+# @Time: 2023/12/25
+import json
+import os
+import random
+import sys
+import time
+import uuid
+from hashlib import md5
+
+from appium import webdriver
+from appium.webdriver.extensions.android.nativekey import AndroidKey
+from bs4 import BeautifulSoup
+from selenium.common.exceptions import NoSuchElementException
+from selenium.webdriver.common.by import By
+
+sys.path.append(os.getcwd())
+
+from application.common.log import AliyunLogger, Local
+from application.common.messageQueue import MQ
+from application.functions import get_redirect_url
+from application.pipeline import PiaoQuanPipeline
+
+
+class SPSSRecommend:
+
+    def __init__(self, log_type, crawler, env, rule_dict, our_uid):
+        self.mq = MQ(topic_name="topic_crawler_etl_" + env)
+        self.platform = "shipinshuashua"
+        self.download_cnt = 0
+        self.element_list = []
+        self.count = 0
+        self.swipe_count = 0
+        self.log_type = log_type
+        self.crawler = crawler
+        self.env = env
+        self.rule_dict = rule_dict
+        self.our_uid_list = our_uid
+        chromedriverExecutable = "/Users/luojunhui/chromedriver/chromedriver_v111/chromedriver"
+        self.aliyun_log = AliyunLogger(platform=crawler, mode=log_type, env=env)
+        Local.logger(self.log_type, self.crawler).info("启动微信")
+        # 微信的配置文件
+        caps = {
+            "platformName": "Android",
+            "devicesName": "Android",
+            "appPackage": "com.tencent.mm",
+            "appActivity": ".ui.LauncherUI",
+            "autoGrantPermissions": "true",
+            "noReset": True,
+            "resetkeyboard": True,
+            "unicodekeyboard": True,
+            "showChromedriverLog": True,
+            "printPageSourceOnFailure": True,
+            "recreateChromeDriverSessions": True,
+            "enableWebviewDetailsCollection": True,
+            "setWebContentsDebuggingEnabled": True,
+            "newCommandTimeout": 6000,
+            "automationName": "UiAutomator2",
+            "chromedriverExecutable": chromedriverExecutable,
+            "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
+        }
+        try:
+            self.driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
+        except Exception as e:
+            print(e)
+            self.aliyun_log.logging(
+                code="3002",
+                message=f'appium 启动异常: {e}'
+            )
+            return
+        self.driver.implicitly_wait(30)
+
+        for i in range(120):
+            try:
+                if self.driver.find_elements(By.ID, "com.tencent.mm:id/f2s"):
+                    Local.logger(self.log_type, self.crawler).info("微信启动成功")
+                    # Common.logging(self.log_type, self.crawler, self.env, '微信启动成功')
+                    self.aliyun_log.logging(
+                        code="1000",
+                        message="启动微信成功"
+                    )
+                    break
+                elif self.driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"):
+                    Local.logger(self.log_type, self.crawler).info("发现并关闭系统下拉菜单")
+                    # Common.logging(self.log_type, self.crawler, self.env, '发现并关闭系统下拉菜单')
+                    self.aliyun_log.logging(
+                        code="1000",
+                        message="发现并关闭系统下拉菜单"
+                    )
+                    size = self.driver.get_window_size()
+                    self.driver.swipe(int(size['width'] * 0.5), int(size['height'] * 0.8),
+                                      int(size['width'] * 0.5), int(size['height'] * 0.2), 200)
+                else:
+                    pass
+            except NoSuchElementException:
+                self.aliyun_log.logging(
+                    code="3001",
+                    message="打开微信异常"
+                )
+                time.sleep(1)
+
+        Local.logger(self.log_type, self.crawler).info("下滑，展示小程序选择面板")
+        size = self.driver.get_window_size()
+        self.driver.swipe(int(size['width'] * 0.5), int(size['height'] * 0.2),
+                          int(size['width'] * 0.5), int(size['height'] * 0.8), 200)
+        time.sleep(1)
+        Local.logger(self.log_type, self.crawler).info('打开小程序"视频刷刷"')
+        self.driver.find_elements(By.XPATH, '//*[@text="视频刷刷"]')[-1].click()
+        self.aliyun_log.logging(
+            code="1000",
+            message="打开小程序 视频刷刷 成功"
+        )
+        time.sleep(5)
+        print("打开小程序")
+
+        self.get_videoList()
+        time.sleep(1)
+        self.driver.quit()
+
+    def search_elements(self, xpath):
+        time.sleep(1)
+        windowHandles = self.driver.window_handles
+        for handle in windowHandles:
+            self.driver.switch_to.window(handle)
+            time.sleep(1)
+            try:
+                elements = self.driver.find_elements(By.XPATH, xpath)
+                if elements:
+                    return elements
+            except NoSuchElementException:
+                pass
+
+    def check_to_applet(self, xpath):
+        time.sleep(1)
+        webViews = self.driver.contexts
+        self.driver.switch_to.context(webViews[-1])
+        windowHandles = self.driver.window_handles
+        for handle in windowHandles:
+            self.driver.switch_to.window(handle)
+            time.sleep(1)
+            try:
+                self.driver.find_element(By.XPATH, xpath)
+                Local.logger(self.log_type, self.crawler).info("切换到WebView成功\n")
+                # Common.logging(self.log_type, self.crawler, self.env, '切换到WebView成功\n')
+                self.aliyun_log.logging(
+                    code="1000",
+                    message="成功切换到 webview"
+                )
+                return
+            except NoSuchElementException:
+                time.sleep(1)
+
+    def swipe_up(self):
+        self.search_elements('//*[@class="dynamic--title-container"]')
+        size = self.driver.get_window_size()
+        self.driver.swipe(int(size["width"] * 0.5), int(size["height"] * 0.8),
+                          int(size["width"] * 0.5), int(size["height"] * 0.442), 200)
+        self.swipe_count += 1
+
+    def get_video_url(self, video_title_element):
+        for i in range(3):
+            self.search_elements('//*[@class="dynamic--title-container"]')
+            Local.logger(self.log_type, self.crawler).info(f"video_title_element:{video_title_element[0]}")
+            time.sleep(1)
+            Local.logger(self.log_type, self.crawler).info("滑动标题至可见状态")
+            self.driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'});",
+                                       video_title_element[0])
+            time.sleep(3)
+            Local.logger(self.log_type, self.crawler).info("点击标题")
+            video_title_element[0].click()
+            self.check_to_applet(xpath=r'//wx-video[@class="infos--title infos--ellipsis"]')
+            Local.logger(self.log_type, self.crawler).info("点击标题完成")
+            time.sleep(10)
+            video_url_elements = self.search_elements(
+                '//wx-video[@class="dynamic-index--video-item dynamic-index--video"]')
+            if video_url_elements:
+                return video_url_elements[0].get_attribute("src")
+
+    def parse_detail(self, index):
+        page_source = self.driver.page_source
+        soup = BeautifulSoup(page_source, 'html.parser')
+        soup.prettify()
+        video_list = soup.findAll(name="wx-view", attrs={"class": "expose--adapt-parent"})
+        element_list = [i for i in video_list][index:]
+        return element_list[0]
+
+    def get_video_info_2(self, video_element):
+        Local.logger(self.log_type, self.crawler).info(f"本轮已抓取{self.download_cnt}条视频\n")
+        # Common.logging(self.log_type, self.crawler, self.env, f"本轮已抓取{self.download_cnt}条视频\n")
+        if self.download_cnt >= int(self.rule_dict.get("videos_cnt", {}).get("min", 10)):
+            self.count = 0
+            self.download_cnt = 0
+            self.element_list = []
+            return
+        self.count += 1
+        Local.logger(self.log_type, self.crawler).info(f"第{self.count}条视频")
+        # 获取 trace_id, 并且把该 id 当做视频生命周期唯一索引
+        trace_id = self.crawler + str(uuid.uuid1())
+        self.aliyun_log.logging(
+            code="1001",
+            trace_id=trace_id,
+            message="扫描到一条视频",
+        )
+        # 标题
+        video_title = video_element.find("wx-view", class_="dynamic--title").text
+        # 播放量字符串
+        play_str = video_element.find("wx-view", class_="dynamic--views").text
+        # 视频时长
+        duration_str = video_element.find("wx-view", class_="dynamic--duration").text
+        user_name = video_element.find("wx-view", class_="dynamic--nick-top").text
+        # 头像 URL
+        avatar_url = video_element.find("wx-image", class_="avatar--avatar")["src"]
+        # 封面 URL
+        cover_url = video_element.find("wx-image", class_="dynamic--bg-image")["src"]
+        play_cnt = int(play_str.replace("+", "").replace("次播放", ""))
+        duration = int(duration_str.split(":")[0].strip()) * 60 + int(duration_str.split(":")[-1].strip())
+        out_video_id = md5(video_title.encode('utf8')).hexdigest()
+        out_user_id = md5(user_name.encode('utf8')).hexdigest()
+        video_dict = {
+            "video_title": video_title,
+            "video_id": out_video_id,
+            'out_video_id': out_video_id,
+            "duration_str": duration_str,
+            "duration": duration,
+            "play_str": play_str,
+            "play_cnt": play_cnt,
+            "like_str": "",
+            "like_cnt": 0,
+            "comment_cnt": 0,
+            "share_cnt": 0,
+            "user_name": user_name,
+            "user_id": out_user_id,
+            'publish_time_stamp': int(time.time()),
+            'publish_time_str': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()))),
+            'update_time_stamp': int(time.time()),
+            "avatar_url": avatar_url,
+            "cover_url": cover_url,
+            "session": f"piaopiiaoquan-{int(time.time())}"
+        }
+        print(video_dict)
+        pipeline = PiaoQuanPipeline(
+            platform=self.crawler,
+            mode=self.log_type,
+            item=video_dict,
+            rule_dict=self.rule_dict,
+            env=self.env,
+            trace_id=trace_id
+        )
+        flag = pipeline.process_item()
+        if flag:
+            video_title_element = self.search_elements(f'//*[contains(text(), "{video_title}")]')
+            if video_title_element is None:
+                return
+            Local.logger(self.log_type, self.crawler).info("点击标题，进入视频详情页")
+            self.aliyun_log.logging(
+                code="1000",
+                message="点击标题，进入视频详情页",
+            )
+            video_url = self.get_video_url(video_title_element)
+            video_url = get_redirect_url(video_url)
+            if video_url is None:
+                self.driver.press_keycode(AndroidKey.BACK)
+                time.sleep(5)
+                return
+            video_dict['video_url'] = video_url
+            video_dict["platform"] = self.crawler
+            video_dict["strategy"] = self.log_type
+            video_dict["out_video_id"] = video_dict["video_id"]
+            video_dict["crawler_rule"] = json.dumps(self.rule_dict)
+            video_dict["user_id"] = random.choice(self.our_uid_list)
+            video_dict["publish_time"] = video_dict["publish_time_str"]
+            self.mq.send_msg(video_dict)
+            self.aliyun_log.logging(
+                code="1002",
+                message="成功发送至ETL",
+                data=video_dict
+            )
+            self.download_cnt += 1
+            self.driver.press_keycode(AndroidKey.BACK)
+            time.sleep(5)
+
+    def get_video_info(self, video_element):
+        try:
+            self.get_video_info_2(video_element)
+        except Exception as e:
+            self.driver.press_keycode(AndroidKey.BACK)
+            Local.logger(self.log_type, self.crawler).error(f"抓取单条视频异常:{e}\n")
+            self.aliyun_log.logging(
+                code="3001",
+                message=f"抓取单条视频异常:{e}\n"
+            )
+
+    def get_videoList(self):
+        self.driver.implicitly_wait(20)
+        # 切换到 web_view
+        self.check_to_applet(xpath='//*[@class="expose--adapt-parent"]')
+        print("切换到 webview 成功")
+        time.sleep(1)
+        page = 0
+        if self.search_elements('//*[@class="expose--adapt-parent"]') is None:
+            Local.logger(self.log_type, self.crawler).info("窗口已销毁\n")
+            # Common.logging(self.log_type, self.crawler, self.env, '窗口已销毁\n')
+            self.aliyun_log.logging(
+                code="3000",
+                message="窗口已销毁"
+            )
+            self.count = 0
+            self.download_cnt = 0
+            self.element_list = []
+            return
+
+        print("开始获取视频信息")
+        for i in range(50):
+            print("下滑{}次".format(i))
+            element = self.parse_detail(i)
+            self.get_video_info(element)
+            self.swipe_up()
+            time.sleep(1)
+            if self.swipe_count > 100:
+                return
+
+        print("下滑完成")
+        # time.sleep(100)
+        Local.logger(self.log_type, self.crawler).info("已抓取完一组，休眠 5 秒\n")
+        # Common.logging(self.log_type, self.crawler, self.env, "已抓取完一组，休眠 5 秒\n")
+        self.aliyun_log.logging(
+            code="1000",
+            message="已抓取完一组，休眠 5 秒\n",
+        )
+        time.sleep(5)
+
+

main.py

@@ -1,48 +1,44 @@
-import os
-import sys
-import time
-import schedule
-import multiprocessing
+import asyncio
+
+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 scheduler import SpiderHome
-
-
-class SpiderScheduler(object):
-    SH = SpiderHome()
-
-
-    @classmethod
-    def protect_spider_timeout(cls, function, hour):
-        run_time_limit = hour * 3600
-        start_time = time.time()
-        process = multiprocessing.Process(target=function)
-        process.start()
-        while True:
-            if time.time() - start_time >= run_time_limit:
-                process.terminate()
-                break
-            if not process.is_alive():
-                print("正在重启")
-                process.terminate()
-                time.sleep(60)
-                os.system("adb forward --remove-all")
-                process = multiprocessing.Process(target=function)
-                process.start()
-            time.sleep(60)
-
-    @classmethod
-    def run_xng_plus(cls, hour):
-        cls.protect_spider_timeout(function=cls.SH.run_xng_plus, hour=hour)
-
-    @classmethod
-    def run_zhuFuQuanZi(cls):
-        print("hello")
-
-
-if __name__ == "__main__":
-    SC = SpiderScheduler()
-    schedule.every().day.at("20:06").do(SC.run_xng_plus, hour=1)
-    while True:
-        schedule.run_pending()
+from application.common.messageQueue import get_consumer, ack_message
+from application.common.log import AliyunLogger
+from application.common.mysql import MysqlHelper
+from application.config import TopicGroup
+
+
+async def run():
+    """
+    传入参数，然后根据参数执行爬虫代码
+    :return:
+    """
+    # 创建并等待一个子进程
+    process = await asyncio.create_subprocess_shell("python3 test2.py")
+    # 等待子进程完成
+    await process.wait()
+
+
+async def main():
+    spider_list = TopicGroup().produce()
+    async_tasks = []  # 异步任务池
+    while spider_list:
+        for spider in spider_list:
+            topic = spider['topic']
+            group = spider['group']
+            consumer = get_consumer(topic, group)
+            messages = consumer.consume_message(batch=1, wait_seconds=5)
+            if messages:
+                task = asyncio.create_task(run())
+                async_tasks.append(task)
+            else:
+                continue
+
+
+if __name__ == '__main__':
+    # 运行主事件循环
+    asyncio.run(main())

off_line_controler.py

@@ -0,0 +1,52 @@
+import os
+import sys
+import time
+import schedule
+import multiprocessing
+
+sys.path.append(os.getcwd())
+
+from scheduler import SpiderHome
+
+
+class SpiderScheduler(object):
+    SH = SpiderHome()
+
+    @classmethod
+    def protect_spider_timeout(cls, function, hour):
+        run_time_limit = hour * 3600
+        start_time = time.time()
+        process = multiprocessing.Process(target=function)
+        process.start()
+        while True:
+            if time.time() - start_time >= run_time_limit:
+                process.terminate()
+                break
+            if not process.is_alive():
+                print("正在重启")
+                process.terminate()
+                time.sleep(60)
+                os.system("adb forward --remove-all")
+                process = multiprocessing.Process(target=function)
+                process.start()
+            time.sleep(60)
+
+    @classmethod
+    def run_xng_plus(cls, hour):
+        cls.protect_spider_timeout(function=cls.SH.run_xng_plus, hour=hour)
+
+    @classmethod
+    def run_zhuFuQuanZi(cls):
+        print("hello")
+
+    @classmethod
+    def run_spss(cls, hour):
+        cls.protect_spider_timeout(function=cls.SH.run_spss, hour=hour)
+
+
+if __name__ == "__main__":
+    SC = SpiderScheduler()
+    # schedule.every().day.at("20:06").do(SC.run_xng_plus, hour=1)
+    schedule.every().day.at("20:30").do(SC.run_spss, hour=1)
+    while True:
+        schedule.run_pending()

scheduler/run_spider_online.py

@@ -0,0 +1,155 @@
+import argparse
+import time
+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 import AliyunLogger
+# from zhuwanwufusu.zhuwanwufusu_recommend import ZhuWanWuFuSuRecommend
+
+
+def main(platform, mode, 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=crawler,
+    #     mode=log_type,
+    #     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=crawler,
+                    mode=log_type,
+                    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_dict = task_fun_mq(msg.message_body)["task_dict"]
+                AliyunLogger.logging(
+                    code="1000",
+                    platform=crawler,
+                    mode=log_type,
+                    env=env,
+                    message="f调度任务:{task_dict}",
+                )
+                # 解析 rule_dict
+                rule_dict = task_fun_mq(msg.message_body)["rule_dict"]
+                AliyunLogger.logging(
+                    code="1000",
+                    platform=crawler,
+                    mode=log_type,
+                    env=env,
+                    message=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=""
+                )
+                AliyunLogger.logging(
+                    code="1003",
+                    platform=crawler,
+                    mode=log_type,
+                    env=env,
+                    message="开始抓取"
+                )
+                AliyunLogger.logging(
+                    code="1000",
+                    platform=crawler,
+                    mode=log_type,
+                    env=env,
+                    message="开始抓取祝万物复苏——推荐",
+                )
+                main_process = ZhuWanWuFuSuRecommend(
+                    platform=crawler,
+                    mode=log_type,
+                    rule_dict=rule_dict,
+                    user_list=user_list,
+                    env=env
+                )
+                main_process.schedule()
+                AliyunLogger.logging(
+                    code="1000",
+                    platform=crawler,
+                    mode=log_type,
+                    env=env,
+                    message="完成抓取——祝万物复苏",
+                )
+                AliyunLogger.logging(
+                    code="1004", platform=crawler, mode=log_type, env=env,message="结束一轮抓取"
+                )
+
+        except MQExceptionBase as err:
+            # Topic中没有消息可消费。
+            if err.type == "MessageNotExist":
+                AliyunLogger.logging(
+                    code="2000",
+                    platform=crawler,
+                    mode=log_type,
+                    env=env,
+                    message=f"No new message! RequestId:{err.req_id}\n",
+                )
+                continue
+            AliyunLogger.logging(
+                code="2000",
+                platform=crawler,
+                mode=log_type,
+                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,
+    )

scheduler/spider_scheduler.py

@@ -23,3 +23,17 @@ class SpiderHome(object):
             rule_dict1,
             [64120158, 64120157, 63676778],
         )
+
+    @classmethod
+    def run_spss(cls):
+        rule_dict = {
+            "videos_cnt": {"min": 600, "max": 600},
+            "play_cnt": {"min": 5000, "max": 5000}
+        }
+        SPSSRecommend(
+            log_type="recommend",
+            crawler="shipinshuashua",
+            env="prod",
+            rule_dict=rule_dict,
+            our_uid=[66433018, 66433020, 66433022, 66433023, 66433024]
+        )