# -*- 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)