crawler_zhiqingzongqun/main/zqttk_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/10/12
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.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
sys.path.append(os.getcwd())
from main.common import Common
from main.publish import Publish
from main.feishu_lib import Feishu


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'])
        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, 'zhiqingzongqun', 'i2kcSD')
            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, text):
        try:
            windowHandles = driver.window_handles
            Common.logger(log_type).info('windowHandles:{}', windowHandles)
            # 遍历所有的handles，找到当前页面所在的handle：如果pageSource有包含你想要的元素，就是所要找的handle
            # 小程序的页面来回切换也需要：遍历所有的handles，切换到元素所在的handle
            for handle in windowHandles:
                Common.logger(log_type).info('切换到对应的windowHandle', handle)
                driver.switch_to.window(handle)
                time.sleep(3)
                if driver.page_source.find(text) != -1:
                    break
        except Exception as e:
            Common.logger(log_type).warning('切换到小程序handle失败，重启APP:{}\n', e)
            cls.quit(log_type, driver)
            cls.start_wechat(log_type, env)

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

            # 鼠标左键点击， 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('webview:{}', webview)
                    Common.logger(log_type).info('切换到小程序\n')
                    # driver.switch_to.context('WEBVIEW_com.tencent.mm:appbrand0')
                    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)

            # Common.logger(log_type).info('点击"换"按钮')
            # try:
            #     driver.find_element(By.XPATH, '//wx-view[@class="refreshbtnbox"]').click()
            # except NoSuchElementException:
            #     Common.logger(log_type).warning('未点击到"换"按钮')

            time.sleep(5)
            Common.logger(log_type).info('获取推荐列表视频信息')
            while True:
                cls.i += 1
                recommend_handles = driver.window_handles
                for recommend_handle in recommend_handles:
                    try:
                        driver.switch_to.window(recommend_handle)

                        # ad
                        try:
                            ad = driver.find_element(
                                By.XPATH,
                                '//*[@class="videolistbox videolist--videolistbox"]'
                                '/*[' + str(cls.i) + ']//*[@class="ad-_banner-_-full"]'
                            )
                        except NoSuchElementException:
                            ad = 0

                        # video_title
                        try:
                            title = driver.find_element(
                                By.XPATH,
                                '//*[@class="videolistbox videolist--videolistbox"]'
                                '/*[' + str(cls.i) + ']//*[@class="video_title videolist--video_title"]')
                            # 向上滚动至-元素可见
                            # Common.logger(log_type).info('滑动视频标题至屏幕中间')
                            driver.execute_script(
                                "arguments[0].scrollIntoView({block:'center',inline:'center'})", title)
                            video_title = title.get_attribute('innerHTML')
                        except NoSuchElementException:
                            title = 0
                            video_title = 0

                        # play_cnt
                        try:
                            play_cnt = driver.find_element(
                                By.XPATH,
                                '//*[@class="videolistbox videolist--videolistbox"]'
                                '/*[' + str(cls.i) + ']//*[@class="clickbox videolist--clickbox"]')\
                                .get_attribute('innerHTML')
                        except NoSuchElementException:
                            play_cnt = 0

                        # cover_url
                        try:
                            cover_url = driver.find_element(
                                By.XPATH,
                                '//*[@class="videolistbox videolist--videolistbox"]'
                                '/*[' + str(cls.i) + ']//*[@class="itemimage videolist--itemimage"]')\
                                .get_attribute('src')
                        except Exception:
                            cover_url = 0

                        if ad != 0:
                            Common.logger(log_type).info('正在获取第{}条:广告\n', cls.i)
                            break
                        elif video_title == 0:
                            pass
                        else:
                            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)
                            # driver.press_keycode(4)
                        elif video_title in [x for y in Feishu.get_values_batch(
                                log_type, 'zhiqingzongqun', 'Z48hlq') for x in y]:
                            Common.logger(log_type).info('视频已存在\n')
                            # driver.press_keycode(4)
                        elif video_title in [x for y in Feishu.get_values_batch(
                                log_type, 'zhiqingzongqun', '1a88b3') for x in y]:
                            Common.logger(log_type).info('视频已下载\n')
                            # driver.press_keycode(4)
                        else:
                            # video_url
                            video_url = cls.get_url(log_type, driver, video_title, title)

                            if video_url == '':
                                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) < 60:
                                    # 删除视频文件夹
                                    shutil.rmtree("./videos/" + video_title + "/")
                                    Common.logger(log_type).info("时长:{}<60秒，删除成功\n", int(download_duration))
                                    return
                                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(' ')[-1].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" +
                                                  "zhiqingzongqun" + 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, "zhiqingzongqun", "1a88b3", "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, "zhiqingzongqun", "1a88b3", "F2:V2", values)
                                    Common.logger(log_type).info("视频:{}，下载/上传成功\n", video_title)
                                    driver.press_keycode(4)

                                break
                    except Exception:
                        pass

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

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

    @classmethod
    def get_url(cls, log_type, driver: WebDriver, video_title, title):
        try:
            Common.logger(log_type).info('进入视频详情:{}', video_title)
            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,
                        '//*[@class="wx-swiper-slide-frame"]'
                        '/*[2]//*[@class="video_item videoswiper--video_item"]').get_attribute('src')
                    return video_url
                except NoSuchElementException:
                    pass
        except Exception as e:
            Common.logger(log_type).error('get_url异常:{}\n', e)


if __name__ == '__main__':
    # Recommend.start_wechat('recommend', 'prod')
    print(Recommend.filter_words('recommend'))

    pass