piaoquan_crawler/zhongmiaoyinxin/zhongmiaoyinxin_recommend/zhongmiaoyinxin_recommend_new.py

# 众妙音信接入后台
# @Author: luojunhui
# @Time: 2023/10/26
import json
import os
import sys
import time
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 bs4 import BeautifulSoup
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
import multiprocessing

sys.path.append(os.getcwd())
from common.mq import MQ
from common.aliyun_log import AliyunLogger
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 = "众妙音信"
        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 = our_uid
        if self.env == "dev":
            chromedriverExecutable = "/Users/luojunhui/Downloads/chromedriver_V111/chromedriver"
        else:
            chromedriverExecutable = '/Users/piaoquan/Downloads/chromedriver'  # Mac 爬虫机器
        # 微信的配置文件
        caps = {
            "platformName": "Android",  # 手机操作系统 Android / iOS
            "deviceName": "a0a65126",  # 连接的设备名（模拟器或真机），安卓可以随便写
            # "udid": "emulator-5554",  # 指定 adb devices 中的哪一台设备
            "platforVersion": "11",  # 手机对应的系统版本
            "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
            "printPageSourceOnFailure": True,  # 找不到元素时，appium log 会完整记录当前页面的 pagesource
            "newCommandTimeout": 6000,  # 初始等待时间
            "automationName": "UiAutomator2",  # 使用引擎，默认为 Appium，
            # 其中 Appium、UiAutomator2、Selendroid、Espresso 用于 Android，XCUITest 用于 iOS
            "showChromedriverLog": True,
            'enableWebviewDetailsCollection': True,
            'setWebContentsDebuggingEnabled': True,
            'recreateChromeDriverSessions': True,
            'chromedriverExecutable': chromedriverExecutable,
            "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
            # "chromeOptions": {"androidProcess": "com.tencent.mm:tools"},
            'browserName': ''
        }
        self.driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
        self.driver.implicitly_wait(30)

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

        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)
        self.driver.find_elements(By.XPATH, '//*[@text="西瓜悦"]')[-1].click()
        print("打开小程序成功")
        time.sleep(5)
        self.get_videoList()
        time.sleep(100)
        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):
        while True:
            webview = self.driver.contexts
            self.driver.switch_to.context(webview[1])
            windowHandles = self.driver.window_handles
            for handle in windowHandles:
                self.driver.switch_to.window(handle)
                time.sleep(1)
                try:
                    video_list = self.driver.find_element(By.XPATH, '//*[@class="index--navbar-list"]/*[1]')
                    video_list.click()
                    print("切换 webview 成功")
                    return
                except NoSuchElementException:
                    time.sleep(1)
            print("切换 webview 失败")
            break

    def swipe_up(self):
        self.search_elements('//*[@class="list-list--list"]')
        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 close_ad(self):
        window_size = self.driver.get_window_size()
        TouchAction(self.driver).tap(x=int(window_size['width'] * 0.5), y=int(window_size['height'] * 0.1)).perform()

    def get_videoList(self):
        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
        self.driver.implicitly_wait(20)
        print("关闭广告")
        self.close_ad()
        print("切换到 webview")
        self.check_to_applet()
        index = 0
        while True:
            if self.search_elements('//*[@id="scrollContainer"]') is None:
                print("窗口已销毁")
                return
            print("获取视频列表")
            video_elements = self.search_elements('//wx-view[@class="cover"]')
            if video_elements is None:
                print("视频列表为空列表")
                return
            video_element_temp = video_elements[index:]
            if len(video_element_temp) == 0:
                print("视频已经到底")
                return
            for i, video_element in enumerate(video_element_temp):
                if video_element is None:
                    return
                self.download_cnt += 1
                self.search_elements('//wx-view[@class="cover"]')
                time.sleep(3)
                self.driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'})",
                                           video_element)
                video_title = video_element.find_elements(By.XPATH, '//wx-view[@class="playImgs"]')[index + i].text
                cover_url = video_element.find_elements(By.XPATH, '//wx-image[@class="coverImg"]')[
                    index + i].get_attribute('src')
                play_cnt = video_element.find_elements(By.XPATH, '//wx-image[@class="coverImg"]/span/*[2]')[
                    index + i].text

                if "万" in play_cnt:
                    play_cnt = int(play_cnt.split("万")[0]) * 10000
                out_video_id = md5(video_title.encode('utf8')).hexdigest()
                video_dict = {
                    'video_title': video_title,
                    'video_id': out_video_id,
                    'out_video_id': out_video_id,
                    'play_cnt': play_cnt,
                    'comment_cnt': 0,
                    'like_cnt': 0,
                    'share_cnt': 0,
                    '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()),
                    'user_name': "zhongmiaoyinxin",
                    'user_id': "zhongmiaoyinxin",
                    'avatar_url': cover_url,
                    'cover_url': cover_url,
                    'session': f"zhongmiaoyinxin-{int(time.time())}"
                }
                pipeline = PiaoQuanPipeline(
                    platform=self.crawler,
                    mode=self.log_type,
                    item=video_dict,
                    rule_dict=self.rule_dict,
                    env=self.env
                )
                flag = pipeline.process_item()
                if flag:
                    print(video_dict)
                else:
                    print("被规则过滤")
                # self.mq.send_msg(video_dict)
                # print(video_dict)


def run():
    rule_dict1 = {}
    ZMYXRecommend("recommend", "zhongmiaoyinxin", "dev", rule_dict1, 6267141)


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)