piaoquan_crawler/zhufuquanzi/zhufuquanzi_recommend/zhufuquanzi_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/9/6
import os
import sys
import time

from appium import webdriver
from appium.webdriver.webdriver import WebDriver
from selenium.common import NoSuchElementException
from selenium.webdriver.common.by import By

sys.path.append(os.getcwd())
from common.common import Common


class ZhufuquanziRecommend:
    platform = "祝福圈子"
    download_cnt = 0
    i = 0

    @classmethod
    def start_wechat(cls, log_type, crawler, env, rule_dict):
        if env == "dev":
            chromedriverExecutable = "/Users/wangkun/Downloads/chromedriver/chromedriver_v111/chromedriver"
        else:
            chromedriverExecutable = "/Users/piaoquan/Downloads/chromedriver"

        Common.logger(log_type, crawler).info("启动微信")
        Common.logging(log_type, crawler, env, '启动微信')
        caps = {
            "platformName": "Android",
            "devicesName": "Android",
            "platformVersion": "12",
            # "udid": "emulator-5554",
            "appPackage": "com.tencent.mm",
            "appActivity": ".ui.LauncherUI",
            "autoGrantPermissions": "true",
            "unicodekeyboard": True,
            "resetkeyboard": True,
            "noReset": True,
            "printPageSourceOnFailure": True,
            "newCommandTimeout": 6000,
            "aotomationName": "UiAutomator2",
            "showChromedriverLog": True,
            "enableWebviewDetailsCollection": True,
            "setWebContentsDebuggingEnabled": True,
            "recreateChromeDriverSessions": True,
            "chromedriverExecutable": chromedriverExecutable,
            "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
        }
        driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
        driver.implicitly_wait(30)

        for i in range(120):
            try:
                if driver.find_elements(By.ID, "com.tencent.mm:id/f2s"):
                    Common.logger(log_type, crawler).info("微信启动成功")
                    Common.logging(log_type, crawler, env, '微信启动成功')
                    break
                elif driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"):
                    Common.logger(log_type, crawler).info("发现并关闭系统下拉菜单")
                    Common.logging(log_type, crawler, env, '发现并关闭系统下拉菜单')
                    driver.find_element(By.ID, "com.android.system:id/dismiss_view").click()
                else:
                    pass
            except NoSuchElementException:
                time.sleep(1)

        Common.logger(log_type, crawler).info("下滑，展示小程序选择面板")
        Common.logging(log_type, crawler, env, '下滑，展示小程序选择面板')
        size = driver.get_window_size()
        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(5)
        Common.logger(log_type, crawler).info('打开小程序"祝福圈子"')
        Common.logging(log_type, crawler, env, '打开小程序"祝福圈子"')
        driver.find_elements(By.XPATH, '//*[@text="祝福圈子"]')[-1].click()
        time.sleep(10)

        cls.get_videoList(log_type, crawler, driver, env, rule_dict)

        time.sleep(3)
        driver.quit()

    @classmethod
    def search_elements(cls, driver: WebDriver, xpath):
        time.sleep(1)
        windowHandles = driver.window_handles
        for handle in windowHandles:
            driver.switch_to.window(handle)
            time.sleep(1)
            try:
                elements = driver.find_elements(By.XPATH, xpath)
                if elements:
                    return elements
            except NoSuchElementException:
                pass

    @classmethod
    def check_to_appplet(cls, log_type, crawler, env, driver: WebDriver, xpath):
        time.sleep(1)
        webviews = driver.contexts
        Common.logger(log_type, crawler).info(f"webviews:{webviews}")
        Common.logging(log_type, crawler, env, f"webviews:{webviews}")
        driver.switch_to.context(webviews[1])
        windowHandles = driver.window_handles
        for handle in windowHandles:
            driver.switch_to.window(handle)
            time.sleep(1)
            try:
                driver.find_element(By.XPATH, xpath)
                Common.logger(log_type, crawler).info("切换到小程序成功\n")
                Common.logging(log_type, crawler, env, '切换到小程序成功\n')
                return
            except NoSuchElementException:
                time.sleep(1)

    @classmethod
    def get_videoList(cls, log_type, crawler, driver: WebDriver, env, rule_dict):
        driver.implicitly_wait(20)
        cls.check_to_appplet(log_type=log_type,
                             crawler=crawler,
                             env=env,
                             driver=driver,
                             xpath='//*[@class="tags--tag tags--tag-0 tags--checked"]')

        time.sleep(3)

        index = 0
        while True:
            if cls.search_elements(driver, '//*[@class="bless--list"]') is None:
                Common.logger(log_type, crawler).info("窗口已销毁\n")
                Common.logging(log_type, crawler, env, '窗口已销毁\n')
                return

            video_list_elements = cls.search_elements(driver,
                                                      '//*[@is="pages/discover/components/bless/dynamic/dynamic"]')
            if video_list_elements is None:
                Common.logger(log_type, crawler).warning(f"当前视频列表为空:{video_list_elements}")
                Common.logging(log_type, crawler, env, f"当前视频列表为空:{video_list_elements}")
                return

            video_list = video_list_elements[index:]
            if len(video_list) == 0 or video_list is None:
                Common.logger(log_type, crawler).info("到底啦~~~~~~~~~~\n")
                Common.logging(log_type, crawler, env, "到底啦~~~~~~~~~~\n")
                return

            for i, video_element in enumerate(video_list):
                if cls.download_cnt >= int(rule_dict.get("videos_cnt", {}).get("min", 20)):
                    Common.logger(log_type, crawler).info(f"本轮已抓取视频数:{cls.download_cnt}")
                    Common.logging(log_type, crawler, env, f"本轮已抓取视频数:{cls.download_cnt}")
                    return

                if video_element is None:
                    Common.logger(log_type, crawler).info("没有更多数据啦\n")
                    Common.logging(log_type, crawler, env, "没有更多数据啦\n")
                    return

                cls.i += 1
                Common.logger(log_type, crawler).info(f"拖动第{cls.i}条视频至屏幕中间")
                Common.logging(log_type, crawler, env, f"拖动第{cls.i}条视频至屏幕中间")
                time.sleep(3)
                driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'})",
                                      video_element)
                cls.download_cnt += 1

            Common.logger(log_type, crawler).info("已抓取完一组，休眠 10 秒\n")
            Common.logging(log_type, crawler, env, "已抓取完一组，休眠 10 秒\n")
            time.sleep(10)
            index = index + len(video_list)


if __name__ == "__main__":
    rule_dict1 = {"period": {"min": 365, "max": 365},
                 "duration": {"min": 30, "max": 1800},
                 "favorite_cnt": {"min": 5000, "max": 0},
                 "videos_cnt": {"min": 10, "max": 20},
                 "share_cnt": {"min": 1000, "max": 0}}
    ZhufuquanziRecommend.start_wechat("recommend", "zhufuquanzi", "dev", rule_dict1)