piaoquan_crawler/shipinhao/shipinhao_main/test_sph_rec.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/7/26
import datetime
import json
import os
import sys
import time
from datetime import date, timedelta
from hashlib import md5
from appium import webdriver
from appium.webdriver.extensions.android.nativekey import AndroidKey
from appium.webdriver.webdriver import WebDriver
from selenium.common import NoSuchElementException
from selenium.webdriver.common.by import By

sys.path.append(os.getcwd())


def start_wechat():
    chromedriverExecutable = r'/opt/homebrew/bin/chromedriver'
    caps = {
        "platformName": "Android",  # 手机操作系统 Android / iOS
        "deviceName": "Android",  # 连接的设备名（模拟器或真机），安卓可以随便写
        "platforVersion": "13",  # 手机对应的系统版本（Android 13）
        "appPackage": "com.tencent.mm",  # 被测APP的包名，乐活圈 Android
        "appActivity": ".ui.LauncherUI",  # 启动的Activity名
        "autoGrantPermissions": True,  # 让 appium 自动授权 base 权限，
        # 如果 noReset 为 True，则该条不生效（该参数为 Android 独有），对应的值为 True 或 False
        "unicodekeyboard": True,  # 使用自带输入法，输入中文时填True
        "resetkeyboard": True,  # 执行完程序恢复原来输入法
        "noReset": True,  # 不重置APP
        "recreateChromeDriverSessions": True,  # 切换到非 chrome-Driver 会 kill 掉 session，就不需要手动 kill 了
        "printPageSourceOnFailure": True,  # 找不到元素时，appium log 会完整记录当前页面的 pagesource
        "newCommandTimeout": 6000,  # 初始等待时间
        "automationName": "UiAutomator2",  # 使用引擎，默认为 Appium，
        # 其中 Appium、UiAutomator2、Selendroid、Espresso 用于 Android，XCUITest 用于 iOS
        "showChromedriverLog": True,
        # "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
        "chromeOptions": {"androidProcess": "com.tencent.mm:tools"},
        # "chromeOptions": {"androidProcess": "com.tencent.mm:toolsmp"},
        # "chromeOptions": {"androidProcess": "com.tencent.mm"},
        "enableWebviewDetailsCollection": True,
        "setWebContentsDebuggingEnabled": True,
        "chromedriverExecutable": chromedriverExecutable,
    }
    driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
    driver.implicitly_wait(10)
    tabs = driver.find_elements(By.ID, "com.tencent.mm:id/f2s")
    for tab in tabs:
        if tab.text == "发现":
            tab.click()
            time.sleep(0.5)
            break
    tabs = driver.find_elements(By.ID, "com.tencent.mm:id/f2s")
    for tab in tabs:
        if tab.text == "发现":
            tab.click()
            time.sleep(0.5)
            break

    textviews = driver.find_elements(By.ID, "android:id/title")
    for textview in textviews:
        if textview.text == "视频号":
            textview.click()
            time.sleep(0.5)
            break
    try:
        driver.find_element(By.ID, "com.tencent.mm:id/lqz").click()
    except NoSuchElementException:
        pass

    for i in range(100):
        try:
            if len(driver.find_elements(By.ID, "com.tencent.mm:id/dkf")) != 0:
                driver.swipe(10, 1600, 10, 300, 200)
                continue
        #     video_dict = cls.get_video_info(driver)
        #     for k, v in video_dict.items():
        #         Common.logger(log_type, crawler).info(f"{k}:{v}")
        #     Common.logging(log_type, crawler, env, f"video_dict:{video_dict}")
        #
        #     if video_dict["video_title"] is None:
        #         Common.logger(log_type, crawler).info("无效视频")
        #         Common.logging(log_type, crawler, env, "无效视频")
        #     elif (
        #             download_rule(
        #                 log_type=log_type,
        #                 crawler=crawler,
        #                 video_dict=video_dict,
        #                 rule_dict=rule_dict,
        #             )
        #             is False
        #     ):
        #         Common.logger(log_type, crawler).info("不满足抓取规则")
        #         Common.logging(log_type, crawler, env, "不满足抓取规则\n")
        #     elif (
        #             cls.repeat_out_video_id(
        #                 log_type, crawler, video_dict["video_id"], env
        #             )
        #             != 0
        #     ):
        #         Common.logger(log_type, crawler).info("视频已下载")
        #         Common.logging(log_type, crawler, env, "视频已下载\n")
        #     else:
        #         cls.download_video_list.append(video_dict)
        #     if i + 1 == scan_count:
        #         Common.logger(log_type, crawler).info("扫描一轮结束\n")
        #         Common.logging(log_type, crawler, env, "扫描一轮结束\n")
        #         return
        #     Common.logger(log_type, crawler).info(
        #         f"已抓取符合规则视频{len(cls.download_video_list)}条，滑动至下一个视频\n"
        #     )
        #     Common.logging(
        #         log_type,
        #         crawler,
        #         env,
        #         f"已抓取符合规则视频{len(cls.download_video_list)}条，滑动至下一个视频\n",
        #     )
        #     driver.swipe(10, 1600, 10, 300, 200)
        except Exception as e:
            print(123)
        #     Common.logger(log_type, crawler).info(f"扫描单条视频时异常:{e}\n")
        #     Common.logging(log_type, crawler, env, f"扫描单条视频时异常:{e}\n")
    time.sleep(50)
    driver.quit()
    # cls.search_video(
    #     log_type=log_type,
    #     crawler=crawler,
    #     rule_dict=rule_dict,
    #     user_dict=user_dict,
    #     driver=driver,
    #     env=env,
    # )
    # cls.close_wechat(log_type=log_type, crawler=crawler, env=env, driver=driver)


def get_recommend_list(cls, log_type, crawler, rule_dict, scan_count, env):
    try:
        driver = cls.start_wechat(log_type, crawler, env)
        cls.get_videoList(
            log_type=log_type,
            crawler=crawler,
            rule_dict=rule_dict,
            env=env,
            scan_count=scan_count,
            driver=driver,
        )
        driver.quit()
        print("退出微信成功")
    except Exception as e:
        print("扫描异常", e)


def main():
    while True:
        scan_count = 20
        rule_dict = {
            "period": {"min": 365, "max": 365},
            "duration": {"min": 10, "max": 1800},
            "favorite_cnt": {"min": 5000, "max": 0},
            "share_cnt": {"min": 1000, "max": 0},
        }


if __name__ == "__main__":
    can_count = 20
    rule_dict = {
        "period": {"min": 365, "max": 365},
        "duration": {"min": 10, "max": 1800},
        "favorite_cnt": {"min": 5000, "max": 0},
        "share_cnt": {"min": 1000, "max": 0},
    }
    start_wechat()