crawler_zhongmiaoyinxin/main/zmyx_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/11/3
import difflib
import os
import shutil
import sys
import time

import ffmpeg
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())
from main.common import Common
from main.feishu_lib import Feishu
from main.zmyx_publish import Publish


class ZMYXRecommend:
    i = 0

    @classmethod
    def get_video_info_from_local(cls, video_path):
        probe = ffmpeg.probe(video_path)
        video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None)
        if video_stream is None:
            print('No video stream found!')
            return
        width = int(video_stream['width'])
        height = int(video_stream['height'])
        duration = float(video_stream['duration'])
        return width, height, duration

    @classmethod
    def filter_words(cls, log_type):
        try:
            filter_words_sheet = Feishu.get_values_batch(log_type, 'zmyx', 'KS1Npq')
            filter_words_list = []
            for x in filter_words_sheet:
                for y in x:
                    if y is None:
                        pass
                    else:
                        filter_words_list.append(y)
            return filter_words_list
        except Exception as e:
            Common.logger(log_type).error('filter_words异常:{}', e)

    @classmethod
    def start_wechat(cls, log_type, env):
        try:
            Common.logger(log_type).info('启动微信')
            caps = {
                "platformName": "Android",  # 手机操作系统 Android / iOS
                "deviceName": "a0a65126",  # 连接的设备名（模拟器或真机），安卓可以随便写
                "platforVersion": "11",  # 手机对应的系统版本（Android 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': '/Users/wangkun/Downloads/chromedriver_v86/chromedriver',
                'chromedriverExecutable': '/Users/piaoquan/Downloads/chromedriver',
                "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
                # "chromeOptions": {"androidProcess": "com.tencent.mm:tools"},
                'browserName': ''
            }
            driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
            driver.implicitly_wait(20)
            # 向下滑动页面，展示出小程序选择面板
            for i in range(120):
                try:
                    # 发现微信消息 TAB，代表微信已启动成功
                    if driver.find_elements(By.ID, 'com.tencent.mm:id/f2s'):
                        break
                    # 发现并关闭系统菜单栏
                    elif driver.find_element(By.ID, 'com.android.systemui:id/dismiss_view'):
                        Common.logger(log_type).info('发现并关闭系统下拉菜单栏')
                        driver.find_element(By.ID, 'com.android.systemui:id/dismiss_view').click()
                    else:
                        pass
                except NoSuchElementException:
                    time.sleep(1)
            Common.logger(log_type).info('下滑，展示小程序选择面板')
            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).info('打开小程序"西瓜座谈"')
            driver.find_elements(By.XPATH, '//*[@text="西瓜座谈"]')[-1].click()

            cls.get_zmyx_recommend(log_type, driver, env)

            cls.quit(log_type, driver)
        except Exception as e:
            Common.logger(log_type).error('start_wechat异常:{}\n', e)

    @classmethod
    def quit(cls, log_type, driver: WebDriver):
        driver.close()
        driver.quit()
        Common.logger(log_type).info('退出 APP 成功\n')

    @classmethod
    def search_element(cls, log_type, driver: WebDriver, element):
        try:
            windowHandles = driver.window_handles
            # 遍历所有的handles，找到当前页面所在的handle：如果pageSource有包含你想要的元素，就是所要找的handle
            # 小程序的页面来回切换也需要：遍历所有的handles，切换到元素所在的handle
            for handle in windowHandles:
                driver.switch_to.window(handle)
                time.sleep(1)
                if len(driver.find_elements(By.XPATH, element)) != 0:
                    return driver.find_element(By.XPATH, element)
                else:
                    pass
        except Exception as e:
            Common.logger(log_type).warning('search_element异常:{}\n', e)

    @classmethod
    def find_ad(cls, log_type, driver: WebDriver):
        Common.logger(log_type).info('切回 NATIVE APP')
        # webview = driver.contexts
        # Common.logger(log_type).info('contexts:{}', webview)
        driver.switch_to.context('NATIVE_APP')
        time.sleep(1)
        try:
            # driver.find_element(By.ID, 'com.tencent.mm:id/hj')
            cls.search_element(log_type, driver, 'com.tencent.mm:id/hj')
            Common.logger(log_type).info('发现广告')
            return True
        except NoSuchElementException:
            Common.logger(log_type).info('未发现广告，切换到小程序')
            webview = driver.contexts
            driver.switch_to.context(webview[1])
            return False

    @classmethod
    def close_ad(cls, log_type, driver: WebDriver):
        if cls.find_ad(log_type, driver) is True:
            while True:
                try:
                    # driver.find_element(By.ID, 'com.tencent.mm:id/ho')
                    cls.search_element(log_type, driver, 'com.tencent.mm:id/ho')
                    Common.logger(log_type).info('广告播放完毕，关闭广告')
                    driver.press_keycode(AndroidKey.BACK)
                    Common.logger(log_type).info('切换到小程序')
                    webview = driver.contexts
                    driver.switch_to.context(webview[1])
                    return
                except NoSuchElementException:
                    time.sleep(1)
        else:
            pass

    @classmethod
    def get_video_url(cls, log_type, driver: WebDriver, play_btn):
        try:
            if play_btn is None or play_btn == 0:
                Common.logger(log_type).info('标题为空\n')
                return 0
            else:
                Common.logger(log_type).info('进入视频详情')
                play_btn.click()

                # time.sleep(5)
                # cls.find_ad(log_type, driver)
                # cls.close_ad(log_type, driver)

                time.sleep(5)
                video_url_element = cls.search_element(log_type, driver, '//wx-video[@class="videoh"]')
                if video_url_element is None:
                    return 0
                else:
                    return video_url_element.get_attribute('src')
        except Exception as e:
            Common.logger(log_type).error('get_video_url异常:{}\n', e)

    @classmethod
    def title_like(cls, log_type, title):
        sheet = Feishu.get_values_batch(log_type, 'zmyx', '19c772')
        for i in range(1, len(sheet)):
            video_title = sheet[i][7]
            if difflib.SequenceMatcher(None, title, video_title).quick_ratio() >= 0.8:
                return True
            else:
                pass

    @classmethod
    def get_zmyx_recommend(cls, log_type, driver: WebDriver, env):
        try:
            time.sleep(5)
            Common.logger(log_type).info('切换到小程序\n')
            webview = driver.contexts
            driver.switch_to.context(webview[1])

            time.sleep(1)
            while True:
                cls.i += 1
                Common.logger(log_type).info('进入"综合"列表')
                cls.search_element(log_type, driver, '//wx-view[@id="top_wait"]')
                # try:
                title_element = cls.search_element(
                    log_type, driver, '//wx-view[@class="uls"]/*['+str(cls.i)+']//span/*[1]')
                play_btn = cls.search_element(
                    log_type, driver, '//wx-view[@class="uls"]/*['+str(cls.i)+']//wx-image[@class="wx-image"]')
                driver.execute_script(
                    "arguments[0].scrollIntoView({block:'center',inline:'center'})", title_element)
                video_title = title_element.get_attribute('innerHTML').replace('\n', '').replace(' ', '')
                play_cnt = cls.search_element(
                    log_type, driver, '//wx-view[@class="uls"]/*['+str(cls.i)+']//span/*[2]').get_attribute('innerHTML')
                cover_url = cls.search_element(
                    log_type, driver, '//wx-view[@class="uls"]/*['+str(cls.i)+']//wx-image').get_attribute('src')

                Common.logger(log_type).info('第{}条视频:{}', cls.i, video_title)

                if video_title == 0 or play_btn == 0 or cover_url == 0:
                    Common.logger(log_type).info('无效视频\n')
                elif video_title in [x for y in Feishu.get_values_batch(log_type, 'zmyx', '19c772') for x in y]:
                    Common.logger(log_type).info('视频已下载\n')
                elif cls.title_like(log_type, video_title) is True:
                    Common.logger(log_type).info('标题相似度>=80%:{}\n', video_title)
                elif any(word if word in video_title else False for word in cls.filter_words(log_type)) is True:
                    Common.logger(log_type).info('视频已中过滤词\n')
                else:
                    video_url = cls.get_video_url(log_type, driver, play_btn)
                    if video_url == 0:
                        Common.logger(log_type).info('未找到视频播放地址\n')
                    else:
                        Common.logger(log_type).info('play_cnt:{}', play_cnt)
                        Common.logger(log_type).info('video_url:{}\n', video_url)

                        # 下载视频
                        Common.download_method(log_type, 'video', video_title, video_url)
                        # 获取视频时长
                        video_info = cls.get_video_info_from_local(
                            "./videos/" + video_title + "/video.mp4")
                        download_width = str(video_info[0])
                        download_height = str(video_info[1])
                        download_duration = video_info[2]
                        # 视频时长<40s，直接删除
                        if int(download_duration) < 40:
                            # 删除视频文件夹
                            shutil.rmtree("./videos/" + video_title + "/")
                            Common.logger(log_type).info("时长:{}<40秒，删除成功\n", int(download_duration))
                            driver.press_keycode(AndroidKey.BACK)
                        else:
                            # 下载封面
                            Common.download_method(log_type, 'cover', video_title, cover_url)
                            # 保存视频信息至 "./videos/{download_video_title}/info.txt"
                            with open("./videos/" + video_title
                                      + "/" + "info.txt", "a", encoding="UTF-8") as f_a:
                                f_a.write("zhongmiaoyinxin" + str(int(time.time())) + "\n" +
                                          str(video_title) + "\n" +
                                          str(int(download_duration)) + "\n" +
                                          str(int(float(
                                              play_cnt.split('万')[0]) * 10000)) + "\n" +
                                          '0' + "\n" +
                                          '0' + "\n" +
                                          '0' + "\n" +
                                          str(download_width) + '*' + str(download_height) + "\n" +
                                          str(int(time.time())) + "\n" +
                                          '众妙音信小程序' + "\n" +
                                          str(cover_url) + "\n" +
                                          str(video_url) + "\n" +
                                          str(cover_url) + "\n" +
                                          "zmyx" + str(int(time.time())))
                            Common.logger(log_type).info("==========视频信息已保存至info.txt==========")

                            # 上传视频
                            Common.logger(log_type).info("开始上传视频:{}".format(video_title))
                            if env == 'dev':
                                our_video_id = Publish.upload_and_publish(log_type, env, "width")
                                our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/" + str(
                                    our_video_id) + "/info"
                            else:
                                our_video_id = Publish.upload_and_publish(log_type, env, "width")
                                our_video_link = "https://admin.piaoquantv.com/cms/post-detail/" + str(
                                    our_video_id) + "/info"
                            Common.logger(log_type).info("视频上传完成:{}", video_title)

                            # 保存视频 ID 到已下载表
                            Common.logger(log_type).info("保存视频至已下载表:{}", video_title)
                            # 视频ID工作表，插入首行
                            Feishu.insert_columns(log_type, "zmyx", "19c772", "ROWS", 1, 2)
                            # 视频ID工作表，首行写入数据
                            upload_time = int(time.time())
                            values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)),
                                       "推荐榜",
                                       video_title,
                                       our_video_link,
                                       play_cnt,
                                       int(download_duration),
                                       str(download_width) + '*' + str(download_height),
                                       cover_url,
                                       video_url]]
                            time.sleep(1)
                            Feishu.update_values(log_type, "zmyx", "19c772", "F2:V2", values)
                            Common.logger(log_type).info("视频:{}，下载/上传成功\n", video_title)
                            driver.press_keycode(AndroidKey.BACK)

                if cls.i == 1000:
                    cls.i = 0
                    break

        except Exception as e:
            Common.logger(log_type).error('get_zmyx_recommend异常，重启APP:{}\n', e)
            cls.i = 0
            cls.quit(log_type, driver)
            cls.start_wechat(log_type, env)


if __name__ == '__main__':
    # ZMYXRecommend.start_wechat('recommend', 'dev')
    ZMYXRecommend.title_like('recommend', '今天是9月3日！又有大事发生！')

    pass