crawler_jixiangxingfu/main/jxxf_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/10/22
import os
import shutil
import sys
import time

import ffmpeg
from appium import webdriver
from appium.webdriver.common.touch_action import TouchAction
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.publish import Publish


class Recommend:
    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'])
        # num_frames = int(video_stream['nb_frames'])
        # fps = int(video_stream['r_frame_rate'].split('/')[0]) / int(video_stream['r_frame_rate'].split('/')[1])
        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, 'jxxf', '9Ws66h')
            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": "Android",  # 连接的设备名（模拟器或真机），安卓可以随便写
                "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',
                '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)
            # 向下滑动页面，展示出小程序选择面板
            time.sleep(20)
            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()

            # 获取视频信息
            time.sleep(1)
            cls.get_recommend(log_type, driver, env)

            # 退出微信
            time.sleep(3)
            Common.logger(log_type).info('退出微信')
            cls.quit(log_type, driver)

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

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

    # 切换 Handle
    @classmethod
    def switch_to_handle(cls, log_type, driver, env):
        try:
            windowHandles = driver.window_handles
            # Common.logger(log_type).info('windowHandles:{}', windowHandles)
            # 遍历所有的handles，找到当前页面所在的handle：如果pageSource有包含你想要的元素，就是所要找的handle
            # 小程序的页面来回切换也需要：遍历所有的handles，切换到元素所在的handle
            for handle in windowHandles:
                driver.switch_to.window(handle)
                time.sleep(1)
                try:
                    video_list = driver.find_element(By.XPATH, '//wx-view[@class="nav-tab-container"]/*[2]')
                    video_list.click()
                    Common.logger(log_type).info('切换到视频列表成功\n')
                    return
                except NoSuchElementException:
                    pass
        except Exception as e:
            Common.logger(log_type).warning('切换到小程序Handle失败，重启APP:{}\n', e)
            cls.quit(log_type, driver)
            cls.i = 0
            cls.start_wechat(log_type, env)

    # 切换视频列表 Handle
    @classmethod
    def switch_to_video_list_handle(cls, log_type, driver, env):
        try:
            recommend_handles = driver.window_handles
            for recommend_handle in recommend_handles:
                # Common.logger(log_type).info("recommend_handle:{}", recommend_handle)
                try:
                    driver.switch_to.window(recommend_handle)
                    driver.find_element(By.XPATH, '//wx-view[@class="video-list-container"]')
                    return
                except NoSuchElementException:
                    pass
        except Exception as e:
            Common.logger(log_type).error('switch_to_video_list_handle异常:{}\n', e)
            cls.quit(log_type, driver)
            cls.i = 0
            cls.start_wechat(log_type, env)

    @classmethod
    def get_recommend(cls, log_type, driver, env):
        try:
            driver.implicitly_wait(10)

            # 鼠标左键点击， 1为x坐标， 2为y坐标
            time.sleep(10)
            Common.logger(log_type).info('关闭广告')
            size = driver.get_window_size()
            TouchAction(driver).tap(x=int(size['width'] * 0.5), y=int(size['height'] * 0.1)).perform()

            while True:
                try:
                    # 切换到 webview
                    webview = driver.contexts
                    Common.logger(log_type).info('切换到小程序')
                    driver.switch_to.context(webview[1])
                    time.sleep(5)
                    cls.switch_to_handle(log_type, driver, env)
                    break
                except Exception as e:
                    Common.logger(log_type).warning('切换到小程序失败，重启APP:{}\n', e)
                    cls.quit(log_type, driver)
                    cls.i = 0
                    cls.start_wechat(log_type, env)

            time.sleep(5)
            while True:
                cls.i += 1
                Common.logger(log_type).info('i:{}\n', cls.i)
                cls.switch_to_video_list_handle(log_type, driver, env)
                try:
                    title = driver.find_element(
                        By.XPATH,
                        '//wx-view[@class="video-list-container"]/*[' + str(cls.i) + ']//*[@class="video-title"]')
                    # 向上滚动至-元素可见
                    # Common.logger(log_type).info('滑动视频标题至屏幕中间')
                    driver.execute_script(
                        "arguments[0].scrollIntoView({block:'center',inline:'center'})", title)
                    video_title = title.get_attribute('innerHTML')

                    play_cnt = driver.find_element(
                        By.XPATH,
                        '//wx-view[@class="video-list-container"]'
                        '/*[' + str(cls.i) + ']//*[@class="video-play-num"]').get_attribute('innerHTML')

                    cover_url = driver.find_element(
                        By.XPATH,
                        '//wx-view[@class="video-list-container"]'
                        '/*[' + str(cls.i) + ']//*[@class="video-cover-img"]').get_attribute('src')
                except NoSuchElementException:
                    title = 0
                    video_title = 0
                    play_cnt = 0
                    cover_url = 0

                Common.logger(log_type).info('正在获取第{}条:{}', cls.i, video_title)

                if video_title == 0 or cover_url == 0:
                    Common.logger(log_type).info('无效视频\n')
                elif '精美图文' in video_title:
                    Common.logger(log_type).info('精美图文\n')
                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', video_title)
                elif video_title in [x for y in Feishu.get_values_batch(log_type, 'jxxf', 'd9e9b1') for x in y]:
                    Common.logger(log_type).info('视频已下载\n')
                else:

                    # video_url
                    video_url = cls.get_url(log_type, driver, title)

                    if video_url == '' or video_url is None:
                        Common.logger(log_type).info('无法播放的视频\n')
                        driver.press_keycode(4)
                    else:
                        Common.logger(log_type).info('play_cnt:{}', play_cnt)
                        Common.logger(log_type).info('video_url:{}', 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]
                        # 视频时长<60s，直接删除
                        if int(download_duration) < 40:
                            # 删除视频文件夹
                            shutil.rmtree("./videos/" + video_title + "/")
                            Common.logger(log_type).info("时长:{}<40秒，删除成功\n", int(download_duration))
                            driver.press_keycode(4)
                        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(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" +
                                          "jxxf" + str(int(time.time())))
                            Common.logger(log_type).info("==========视频信息已保存至info.txt==========")

                            # 上传视频
                            Common.logger(log_type).info("开始上传视频:{}".format(video_title))
                            if env == 'dev' and int(download_width) >= int(download_height):
                                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"
                            elif env == 'dev' and int(download_width) < int(download_height):
                                our_video_id = Publish.upload_and_publish(log_type, env, "height")
                                our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/" + str(
                                    our_video_id) + "/info"
                            elif env == 'prod' and int(download_width) >= int(download_height):
                                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"
                            elif env == 'prod' and int(download_width) < int(download_height):
                                our_video_id = Publish.upload_and_publish(log_type, env, "height")
                                our_video_link = "https://admin.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, "jxxf", "d9e9b1", "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, "jxxf", "d9e9b1", "F2:V2", values)
                            Common.logger(log_type).info("视频:{}，下载/上传成功\n", video_title)
                            driver.press_keycode(4)

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

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

    @classmethod
    def get_url(cls, log_type, driver: WebDriver, title):
        try:
            Common.logger(log_type).info('进入视频详情')
            title.click()
            time.sleep(5)
            Common.logger(log_type).info('关闭广告')
            size = driver.get_window_size()
            TouchAction(driver).tap(x=int(size['width'] * 0.5), y=int(size['height'] * 0.1)).perform()
            time.sleep(10)
            info_handles = driver.window_handles
            for info_handle in info_handles:
                try:
                    driver.switch_to.window(info_handle)
                    video_url = driver.find_element(
                        By.XPATH,
                        '//wx-view[@class="one-screen-2"]/*[2]').get_attribute('src')
                    return video_url
                except NoSuchElementException:
                    pass
        except Exception as e:
            Common.logger(log_type).error('get_url异常:{}\n', e)
            driver.press_keycode(4)


if __name__ == "__main__":

    # print(Recommend.filter_words('recommend'))
    Recommend.start_wechat('recommend', 'dev')

    pass