piaoquan_crawler/zhiqingtiantiankan/zhiqingtiantiankan_recommend/zhiqingtiantiankan_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/4/18
import json
import os
import shutil
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 selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
sys.path.append(os.getcwd())
from common.common import Common
from common.feishu import Feishu
from common.publish import Publish
from common.scheduling_db import MysqlHelper


class ZhiqingtiantiankanRecommend:
    platform = "知青天天看"
    i = 0

    @classmethod
    def zhiqingtiantiankan_config(cls, log_type, crawler, text, env):
        select_sql = f"""select * from crawler_config where source="zhiqingtiantiankan" """
        contents = MysqlHelper.get_values(log_type, crawler, select_sql, env, action='')
        title_list = []
        filter_list = []
        for content in contents:
            config = content['config']
            config_dict = eval(config)
            for k, v in config_dict.items():
                if k == "title":
                    title_list_config = v.split(",")
                    for title in title_list_config:
                        title_list.append(title)
                if k == "filter":
                    filter_list_config = v.split(",")
                    for filter_word in filter_list_config:
                        filter_list.append(filter_word)
        if text == "title":
            return title_list
        elif text == "filter":
            return filter_list

    @classmethod
    def start_wechat(cls, log_type, crawler, env):
        try:
            Common.logger(log_type, crawler).info('启动微信')
            if env == "dev":
                chromedriverExecutable = '/Users/wangkun/Downloads/chromedriver/chromedriver_v111/chromedriver'
                # chromedriverExecutable = 'C:\\chromedriver\\chromedriver.exe'  # 阿里云 Windows
            else:
                chromedriverExecutable = '/Users/piaoquan/Downloads/chromedriver'  # Mac 爬虫机器
                # chromedriverExecutable = 'C:\\chromedriver\\chromedriver.exe'  # 阿里云 Windows
            caps = {
                "platformName": "Android",  # 手机操作系统 Android / iOS
                "deviceName": "Android",  # 连接的设备名（模拟器或真机），安卓可以随便写
                # "udid": "emulator-5554",  # 指定 adb devices 中的哪一台设备
                "platforVersion": "11",  # 手机对应的系统版本
                "appPackage": "com.tencent.mm",  # 被测APP的包名
                "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': ''
            }
            driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)  # 指定 Appium 端口号：4723
            driver.implicitly_wait(30)
            # 向下滑动页面，展示出小程序选择面板
            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, crawler).info('发现并关闭系统下拉菜单栏')
                        driver.find_element(By.ID, 'com.android.systemui:id/dismiss_view').click()
                    else:
                        pass
                except NoSuchElementException:
                    time.sleep(1)
            Common.logger(log_type, crawler).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, crawler).info('打开小程序"知青天天看"')
            driver.find_elements(By.XPATH, '//*[@text="知青天天看"]')[-1].click()

            # 获取视频信息
            # time.sleep(30)
            time.sleep(10)
            cls.get_videoList(log_type, crawler, driver, env)

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

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

    # 切换 Handle
    @classmethod
    def search_elements(cls, driver: WebDriver, xpath):
        windowHandles = driver.window_handles
        for handle in windowHandles:
            driver.switch_to.window(handle)
            time.sleep(1)
            try:
                elements = driver.find_elements(By.XPATH, xpath)
                if elements:
                    return elements
            except NoSuchElementException:
                pass

    @classmethod
    def repeat_out_video_id(cls, log_type, crawler, out_video_id, env):
        sql = f""" select * from crawler_video where platform in ("众妙音信", "刚刚都传", "吉祥幸福", "知青天天看", "zhufuquanzi", "祝福圈子", "haitunzhufu", "海豚祝福") and out_video_id="{out_video_id}"; """
        repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env)
        return len(repeat_video)

    @classmethod
    def repeat_video_url(cls, log_type, crawler, video_url, env):
        sql = f""" select * from crawler_video where platform="{cls.platform}" and video_url="{video_url}"; """
        repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env)
        return len(repeat_video)

    @classmethod
    def check_to_applet(cls, log_type, crawler, driver: WebDriver):
        while True:
            webview = driver.contexts
            Common.logger(log_type, crawler).info(f"webview:{webview}")
            driver.switch_to.context(webview[1])
            windowHandles = driver.window_handles
            for handle in windowHandles:
                driver.switch_to.window(handle)
                time.sleep(1)
                try:
                    applet = driver.find_element(By.XPATH, '//wx-view[@class="u-title navbar--u-title u-line-1 navbar--u-line-1 data-v-febd4d40 navbar--data-v-febd4d40"]')
                    if applet:
                        Common.logger(log_type, crawler).info('切换到小程序成功\n')
                        return
                except NoSuchElementException:
                    time.sleep(1)
            Common.logger(log_type, crawler).info("切换到小程序失败\n")
            break

    @classmethod
    def find_ad(cls, log_type, crawler, driver: WebDriver):
        windowHandles = driver.window_handles
        # Common.logger(log_type, crawler).info('windowHandles:{}', windowHandles)
        # 遍历所有的handles，找到当前页面所在的handle：如果pageSource有包含你想要的元素，就是所要找的handle
        # 小程序的页面来回切换也需要：遍历所有的handles，切换到元素所在的handle
        for handle in windowHandles:
            driver.switch_to.window(handle)
            time.sleep(1)
            try:
                Common.logger(log_type, crawler).info("寻找广告~~~~~~")
                ad_element = driver.find_element(By.XPATH, '//div[@class="ad-text"]')
                if ad_element:
                    Common.logger(log_type, crawler).info("发现广告")
                    for i in range(20):
                        if driver.find_element(By.XPATH, '//div[@id="count_down_container"]/*[1]').text == "已完成浏览":
                            Common.logger(log_type, crawler).info("广告播放完毕，点击返回")
                            driver.press_keycode(AndroidKey.BACK)
                            return
                        else:
                            Common.logger(log_type, crawler).info("广告未播放完毕，等待 1 秒")
                            time.sleep(1)
                else:
                    Common.logger(log_type, crawler).info("未发现广告， 退出")
                    return
            except NoSuchElementException:
                time.sleep(1)

    @classmethod
    def get_video_url(cls, log_type, crawler, driver: WebDriver, video_element):
        video_element.click()
        time.sleep(5)
        cls.close_native_ad(log_type, crawler, driver)
        windowHandles = driver.window_handles
        for handle in windowHandles:
            driver.switch_to.window(handle)
            time.sleep(1)
            try:
                video_url_element = driver.find_element(By.XPATH, '//*[@class="wx-swiper-slide-frame"]/*[2]//*[@class="video_item videoswiper--video_item"]')
                video_url = video_url_element.get_attribute("src")
                # cls.find_ad(log_type, crawler, driver)
                return video_url
            except NoSuchElementException:
                time.sleep(1)

    @classmethod
    def close_native_ad(cls, log_type, crawler, driver: WebDriver):
        Common.logger(log_type, crawler).info('关闭广告')
        size = driver.get_window_size()
        time.sleep(3)
        TouchAction(driver).tap(x=int(size['width'] * 0.4), y=int(size['height'] * 0.1)).perform()

    @classmethod
    def get_videoList(cls, log_type, crawler, driver: WebDriver, env):
        driver.implicitly_wait(20)
        # 关闭广告
        # cls.close_native_ad(log_type, crawler, driver)
        # 切换到小程序
        cls.check_to_applet(log_type, crawler, driver)

        time.sleep(5)
        index = 0
        while True:
            try:
                if cls.search_elements(driver, '//wx-view[@class="listbox"]') is None:
                    Common.logger(log_type, crawler).info('窗口已销毁\n')
                    return

                Common.logger(log_type, crawler).info('获取视频列表\n')
                video_elements = cls.search_elements(driver, '//wx-view[@class="videolist-box videolist--videolist-box"]')
                if video_elements is None:
                    Common.logger(log_type, crawler).warning(f'video_elements:{video_elements}')
                    return

                video_element_temp = video_elements[index:]
                if len(video_element_temp) == 0:
                    Common.logger(log_type, crawler).info('到底啦~~~~~~~~~~~~~\n')
                    return

                for i, video_element in enumerate(video_element_temp):
                    if video_element is None:
                        Common.logger(log_type, crawler).info('到底啦~\n')
                        return
                    cls.i += 1
                    cls.search_elements(driver, '//wx-view[@class="videolist-box videolist--videolist-box"]')

                    Common.logger(log_type, crawler).info(f'拖动"视频"列表第{cls.i}个至屏幕中间')
                    time.sleep(3)
                    driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'})", video_element)

                    # video_title = video_element.find_elements(By.XPATH, '//wx-view[@class="video_title videolist--video_title"]')[cls.i-1].text
                    # cover_url = video_element.find_elements(By.XPATH, '//wx-image[@class="itemimage videolist--itemimage"]')[cls.i-1].get_attribute('src')
                    # play_cnt = video_element.find_elements(By.XPATH, '//wx-view[@class="clickbox videolist--clickbox"]')[cls.i-1].text

                    video_title = video_element.find_elements(By.XPATH, '//wx-view[@class="video_title videolist--video_title"]')[index+i].text
                    cover_url = video_element.find_elements(By.XPATH, '//wx-image[@class="itemimage videolist--itemimage"]')[index+i].get_attribute('src')
                    play_cnt = video_element.find_elements(By.XPATH, '//wx-view[@class="clickbox videolist--clickbox"]')[index+i].text

                    play_cnt = int(float(play_cnt.replace("阅读数", "").strip().split("万")[0]) * 10000)
                    out_video_id = md5(video_title.encode('utf8')).hexdigest()
                    video_dict = {
                        'video_title': video_title,
                        '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()))),
                        'user_name': "zhiqingtiantiankan",
                        'user_id': "zhiqingtiantiankan",
                        'avatar_url': cover_url,
                        'cover_url': cover_url,
                        'session': f"zhiqingtiantiankan-{int(time.time())}"
                    }
                    for k, v in video_dict.items():
                        Common.logger(log_type, crawler).info(f"{k}:{v}")

                    if video_title is None or cover_url is None:
                        Common.logger(log_type, crawler).info("无效视频\n")
                    elif any(str(word) if str(word) in video_title else False for word in
                             cls.zhiqingtiantiankan_config(log_type, crawler, "filter", env)) is True:
                        Common.logger(log_type, crawler).info('已中过滤词\n')
                    elif cls.repeat_out_video_id(log_type, crawler, out_video_id, env) != 0:
                        Common.logger(log_type, crawler).info('视频已下载\n')
                    else:
                        video_url = cls.get_video_url(log_type, crawler, driver, video_element)

                        if video_url is None:
                            Common.logger(log_type, crawler).info("未获取到视频播放地址\n")
                            driver.press_keycode(AndroidKey.BACK)
                        elif cls.repeat_video_url(log_type, crawler, video_url, env) != 0:
                            Common.logger(log_type, crawler).info('视频已下载\n')
                            driver.press_keycode(AndroidKey.BACK)
                        else:
                            video_dict["video_url"] = video_url
                            Common.logger(log_type, crawler).info(f"video_url:{video_url}\n")
                            # driver.press_keycode(AndroidKey.BACK)
                            cls.download_publish(log_type, crawler, video_dict, env, driver)
                Common.logger(log_type, crawler).info('已抓取完一组视频，休眠10秒\n')
                time.sleep(10)
                index = index + len(video_element_temp)
            except Exception as e:
                Common.logger(log_type, crawler).info(f"get_videoList:{e}\n")
                cls.i = 0
                return

    @classmethod
    def download_publish(cls, log_type, crawler, video_dict, env, driver: WebDriver):
        # 下载视频
        Common.download_method(log_type=log_type, crawler=crawler, text='video', title=video_dict['video_title'],
                               url=video_dict['video_url'])
        ffmpeg_dict = Common.ffmpeg(log_type, crawler, f"./{crawler}/videos/{video_dict['video_title']}/video.mp4")
        if ffmpeg_dict is None:
            md_title = md5(video_dict['video_title'].encode('utf8')).hexdigest()
            shutil.rmtree(f"./{crawler}/videos/{md_title}/")
            Common.logger(log_type, crawler).info("视频size=0，删除成功\n")
            return
        video_dict["duration"] = ffmpeg_dict["duration"]
        video_dict["video_width"] = ffmpeg_dict["width"]
        video_dict["video_height"] = ffmpeg_dict["height"]
        # 下载封面
        Common.download_method(log_type=log_type, crawler=crawler, text='cover', title=video_dict['video_title'],
                               url=video_dict['cover_url'])
        # 保存视频信息至txt
        Common.save_video_info(log_type=log_type, crawler=crawler, video_dict=video_dict)

        # 上传视频
        Common.logger(log_type, crawler).info("开始上传视频...")
        our_video_id = Publish.upload_and_publish(log_type=log_type,
                                                  crawler=crawler,
                                                  strategy="推荐榜爬虫策略",
                                                  our_uid="recommend",
                                                  env=env,
                                                  oss_endpoint="out")
        if env == 'dev':
            our_video_link = f"https://testadmin.piaoquantv.com/cms/post-detail/{our_video_id}/info"
        else:
            our_video_link = f"https://admin.piaoquantv.com/cms/post-detail/{our_video_id}/info"
        Common.logger(log_type, crawler).info("视频上传完成")

        if our_video_id is None:
            # 删除视频文件夹
            shutil.rmtree(f"./{crawler}/videos/{video_dict['video_title']}")
            return

        # 视频信息保存至飞书
        Feishu.insert_columns(log_type, crawler, "1a88b3", "ROWS", 1, 2)
        # 视频ID工作表，首行写入数据
        upload_time = int(time.time())
        values = [[time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(upload_time)),
                   "推荐榜爬虫策略",
                   video_dict["video_title"],
                   video_dict["video_id"],
                   video_dict["play_cnt"],
                   video_dict["duration"],
                   f'{video_dict["video_width"]}*{video_dict["video_height"]}',
                   our_video_link,
                   video_dict["cover_url"],
                   video_dict["video_url"]]]
        time.sleep(1)
        Feishu.update_values(log_type, crawler, "1a88b3", "F2:V2", values)
        Common.logger(log_type, crawler).info(f"视频已保存至飞书文档\n")

        rule_dict = {}
        # 视频信息保存数据库
        insert_sql = f""" insert into crawler_video(video_id,
                                                            out_user_id,
                                                            platform,
                                                            strategy,
                                                            out_video_id,
                                                            video_title,
                                                            cover_url,
                                                            video_url,
                                                            duration,
                                                            publish_time,
                                                            play_cnt,
                                                            crawler_rule,
                                                            width,
                                                            height)
                                                            values({our_video_id},
                                                            "{video_dict['user_id']}",
                                                            "{cls.platform}",
                                                            "推荐榜爬虫策略",
                                                            "{video_dict['video_id']}",
                                                            "{video_dict['video_title']}",
                                                            "{video_dict['cover_url']}",
                                                            "{video_dict['video_url']}",
                                                            {int(video_dict['duration'])},
                                                            "{video_dict['publish_time_str']}",
                                                            {int(video_dict['play_cnt'])},
                                                            '{json.dumps(rule_dict)}',
                                                            {int(video_dict['video_width'])},
                                                            {int(video_dict['video_height'])}) """
        Common.logger(log_type, crawler).info(f"insert_sql:{insert_sql}")
        MysqlHelper.update_values(log_type, crawler, insert_sql, env, action='')
        Common.logger(log_type, crawler).info('视频信息插入数据库成功！\n')
        driver.press_keycode(AndroidKey.BACK)


if __name__ == '__main__':
    ZhiqingtiantiankanRecommend.start_wechat('recommend', 'zhiqingtiantiankan', 'prod')

    pass