piaoquan_crawler/shipinhao/shipinhao_search/shipinhao_search.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/4/25
import difflib
import os
import sys
import time
from datetime import date, timedelta
from hashlib import md5
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 common.feishu import Feishu
from common.publish import Publish
from common.common import Common
from common.public import get_config_from_mysql


class ShipinhaoSearch:
    i = 0

    @classmethod
    def start_wechat(cls, log_type, crawler, word, env):
        Common.logger(log_type, crawler).info('启动微信')
        if env == "dev":
            chromedriverExecutable = "/Users/wangkun/Downloads/chromedriver/chromedriver_v107/chromedriver"
        else:
            chromedriverExecutable = '/Users/piaoquan/Downloads/chromedriver'
        caps = {
            "platformName": "Android",  # 手机操作系统 Android / iOS
            "deviceName": "Android",  # 连接的设备名（模拟器或真机），安卓可以随便写
            "platforVersion": "13",  # 手机对应的系统版本（Android 13）
            "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
            "recreateChromeDriverSessions": True,  # 切换到非 chrome-Driver 会 kill 掉 session，就不需要手动 kill 了
            "printPageSourceOnFailure": True,  # 找不到元素时，appium log 会完整记录当前页面的 pagesource
            "newCommandTimeout": 6000,  # 初始等待时间
            "automationName": "UiAutomator2",  # 使用引擎，默认为 Appium，
            # 其中 Appium、UiAutomator2、Selendroid、Espresso 用于 Android，XCUITest 用于 iOS
            "showChromedriverLog": True,
            # "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
            "chromeOptions": {"androidProcess": "com.tencent.mm:tools"},
            'enableWebviewDetailsCollection': True,
            'setWebContentsDebuggingEnabled': True,
            'chromedriverExecutable': chromedriverExecutable,
        }
        driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
        driver.implicitly_wait(10)
        # # 向下滑动页面，展示出小程序选择面板
        # 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)
        if len(driver.find_elements(By.ID, 'android:id/text1')) != 0:
            driver.find_elements(By.ID, 'android:id/text1')[0].click()
        time.sleep(5)
        cls.search_video(log_type=log_type,
                         crawler=crawler,
                         word=word,
                         driver=driver,
                         env=env)
        cls.close_wechat(log_type=log_type,
                         crawler=crawler,
                         driver=driver)

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

    @classmethod
    def is_contain_chinese(cls, strword):
        for ch in strword:
            if u'\u4e00' <= ch <= u'\u9fff':
                return True
        return False

    # 查找元素
    @classmethod
    def search_elements(cls, driver: WebDriver, xpath):
        time.sleep(1)
        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 check_to_webview(cls, log_type, crawler, driver: WebDriver):
        # Common.logger(log_type, crawler).info('切换到webview')
        webviews = driver.contexts
        driver.switch_to.context(webviews[1])
        time.sleep(1)
        windowHandles = driver.window_handles
        for handle in windowHandles:
            driver.switch_to.window(handle)
            try:
                shipinhao_webview = driver.find_element(By.XPATH, '//div[@class="unit"]')
                if shipinhao_webview:
                    Common.logger(log_type, crawler).info('切换到视频号 webview 成功')
                    return "成功"
            except Exception as e:
                Common.logger(log_type, crawler).info(f"{e}\n")

    @classmethod
    def search_video(cls, log_type, crawler, word, driver: WebDriver, env):
        # 点击微信搜索框，并输入搜索词
        driver.implicitly_wait(10)
        driver.find_element(By.ID, 'com.tencent.mm:id/j5t').click()
        time.sleep(0.5)
        Common.logger(log_type, crawler).info(f'输入搜索词:{word}')
        driver.find_element(By.ID, 'com.tencent.mm:id/cd7').clear().send_keys(word)
        driver.press_keycode(AndroidKey.ENTER)
        # driver.find_elements(By.ID, 'com.tencent.mm:id/oi4')[0].click()
        driver.find_element(By.ID, 'com.tencent.mm:id/m94').click()
        time.sleep(5)

        # 切换到微信搜索结果页 webview
        check_to_webview = cls.check_to_webview(log_type, crawler, driver)
        if check_to_webview is None:
            Common.logger(log_type, crawler).info("切换到视频号 webview 失败\n")
            return
        time.sleep(1)

        # 切换到"视频号"分类
        shipinhao_tags = cls.search_elements(driver, '//div[@class="unit"]/*[2]')
        Common.logger(log_type, crawler).info('点击"视频号"分类')
        shipinhao_tags[0].click()
        time.sleep(5)

        index = 0
        while True:
            if cls.i >= 100:
                Common.logger(log_type, crawler).info(f'搜索词:"{word}"，已抓取视频数:{index}')
                cls.i = 0
                return
            # try:
            if cls.search_elements(driver, '//*[@class="double-rich double-rich_vertical"]') is None:
                Common.logger(log_type, crawler).info('窗口已销毁\n')
                return

            Common.logger(log_type, crawler).info('获取视频列表\n')
            video_elements = cls.search_elements(driver, '//div[@class="vc active__mask"]')
            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, '//div[@class="vc active__mask"]')

                Common.logger(log_type, crawler).info(f'拖动"视频"列表第{cls.i}个至屏幕中间')
                time.sleep(3)
                driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'})",
                                      video_element)
                if len(video_element.find_elements(By.XPATH, "//*[@text='没有更多的搜索结果']")) != 0:
                    Common.logger(log_type, crawler).info("没有更多的搜索结果\n")
                    return
                video_title = video_element.find_elements(By.XPATH, '//div[@class="title ellipsis_2"]/*[2]')[index + i].text
                video_url = video_element.find_elements(By.XPATH, '//div[@class="video-player"]')[index+i].get_attribute('src')
                cover_url = video_element.find_elements(By.XPATH, '//div[@class="video-player__bd"]')[index+i].get_attribute('style')
                duration = video_element.find_elements(By.XPATH, '//div[@class="play-mask__text"]/*[2]')[index+i].text
                duration = int(duration.split(':')[0]) * 60 + int(duration.split(':')[-1])
                user_name = video_element.find_elements(By.XPATH, '//p[@class="vc-source__text"]')[index+i].text
                avatar_url = video_element.find_elements(By.XPATH, '//div[@class="ui-image-image ui-image vc-source__thumb"]')[index+i].get_attribute('style')

                # Common.logger(log_type, crawler).info(f"video_title:{video_title}")
                # Common.logger(log_type, crawler).info(f"duration:{duration}")
                video_element.click()
                time.sleep(3)
                video_dict = cls.get_video_info(log_type=log_type,
                                   crawler=crawler,
                                   driver=driver)
                video_dict["video_title"] = video_title
                video_dict["duration"] = duration
                video_dict["video_url"] = video_url
                for k, v in video_dict.items():
                    Common.logger(log_type, crawler).info(f"{k}:{v}")
                if video_title in [x for y in Feishu.get_values_batch(log_type, crawler, "xYWCzf") for x in y]:
                    Common.logger(log_type, crawler).info("视频已存在\n")
                else:
                    cls.download_publish(log_type, crawler, word, video_dict)

            Common.logger(log_type, crawler).info('已抓取完一组视频，休眠1秒\n')
            time.sleep(1)
            index = index + len(video_element_temp)
            # except Exception as e:
            #     Common.logger(log_type, crawler).info(f"get_videoList:{e}\n")
            #     cls.i = 0

    @classmethod
    def download_publish(cls, log_type, crawler, word, video_dict):
        Feishu.insert_columns(log_type, crawler, "xYWCzf", "ROWS", 1, 2)
        time.sleep(0.5)
        values = [[time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()))),
                   "视频号搜索",
                   word,
                   video_dict["video_title"],
                   video_dict["duration"],
                   video_dict["like_cnt"],
                   video_dict["share_cnt"],
                   video_dict["favorite_cnt"],
                   video_dict["comment_cnt"],
                   video_dict["publish_time_str"],
                   "待获取",
                   "待获取",
                   "待获取",
                   video_dict["video_url"]]]
        Feishu.update_values(log_type, crawler, "xYWCzf", "F2:Z2", values)
        Common.logger(log_type, crawler).info("写入飞书成功\n")

    @classmethod
    def get_video_info(cls, log_type, crawler, driver: WebDriver):
        # Common.logger(log_type, crawler).info('切回NATIVE_APP')
        driver.switch_to.context('NATIVE_APP')

        # 点赞
        like_id = driver.find_element(By.ID, 'com.tencent.mm:id/k04')
        like_cnt = like_id.get_attribute('name')
        if like_cnt == "" or like_cnt == "喜欢" or like_cnt == "火" or cls.is_contain_chinese(like_cnt) is True:
            like_cnt = 0
        elif '万' in like_cnt:
            like_cnt = int(float(like_cnt.split('万')[0]) * 10000)
        elif '万+' in like_cnt:
            like_cnt = int(float(like_cnt.split('万+')[0]) * 10000)
        else:
            like_cnt = int(float(like_cnt))

        # 分享
        share_id = driver.find_element(By.ID, 'com.tencent.mm:id/jhv')
        share_cnt = share_id.get_attribute('name')
        if share_cnt == "" or share_cnt == "转发" or cls.is_contain_chinese(share_cnt) is True:
            share_cnt = 0
        elif '万' in share_cnt:
            share_cnt = int(float(share_cnt.split('万')[0]) * 10000)
        elif '万+' in share_cnt:
            share_cnt = int(float(share_cnt.split('万+')[0]) * 10000)
        else:
            share_cnt = int(float(share_cnt))

        # 收藏
        favorite_id = driver.find_element(By.ID, 'com.tencent.mm:id/fnp')
        favorite_cnt = favorite_id.get_attribute('name')
        if favorite_cnt == "" or favorite_cnt == "收藏" or favorite_cnt == "推荐" or favorite_cnt == "火" or cls.is_contain_chinese(favorite_cnt) is True:
            favorite_cnt = 0
        elif '万' in favorite_cnt:
            favorite_cnt = int(float(favorite_cnt.split('万')[0]) * 10000)
        elif '万+' in favorite_cnt:
            favorite_cnt = int(float(favorite_cnt.split('万+')[0]) * 10000)
        else:
            favorite_cnt = int(float(favorite_cnt))

        # 评论
        comment_id = driver.find_element(By.ID, 'com.tencent.mm:id/bje')
        comment_cnt = comment_id.get_attribute('name')
        if comment_cnt == "" or comment_cnt == "评论" or cls.is_contain_chinese(comment_cnt) is True:
            comment_cnt = 0
        elif '万' in comment_cnt:
            comment_cnt = int(float(comment_cnt.split('万')[0]) * 10000)
        elif '万+' in comment_cnt:
            comment_cnt = int(float(comment_cnt.split('万+')[0]) * 10000)
        else:
            comment_cnt = int(float(comment_cnt))

        # 发布时间
        comment_id.click()
        time.sleep(1)
        publish_time = driver.find_element(By.ID, "com.tencent.mm:id/bre").get_attribute("name")
        if "天前" in publish_time:
            days = int(publish_time.replace("天前", ""))
            publish_time_str = (date.today() + timedelta(days=-days)).strftime("%Y-%m-%d")
        elif "年" in publish_time:
            publish_time_str = publish_time.replace("年", "-").replace("月", "-").replace("日", "")
        else:
            publish_time_str = f'2023-{publish_time.replace("月", "-").replace("日", "")}'
        # publish_time_stamp = int(time.mktime(time.strptime(publish_time_str, "%Y-%m-%d")))

        # 收起评论
        # Common.logger(log_type, crawler).info("收起评论")
        driver.find_element(By.ID, "com.tencent.mm:id/be_").click()
        time.sleep(0.5)
        # 返回 webview
        # Common.logger(log_type, crawler).info(f"操作手机返回按键")
        driver.find_element(By.ID, "com.tencent.mm:id/a2z").click()
        time.sleep(0.5)
        # driver.press_keycode(AndroidKey.BACK)
        # cls.check_to_webview(log_type=log_type, crawler=crawler, driver=driver)
        webviews = driver.contexts
        driver.switch_to.context(webviews[1])

        video_dict = {
            "like_cnt": like_cnt,
            "share_cnt": share_cnt,
            "favorite_cnt": favorite_cnt,
            "comment_cnt": comment_cnt,
            "publish_time_str": publish_time_str,
            # "publish_time_stamp": publish_time_stamp,
        }
        return video_dict

    @classmethod
    def search_all_videos(cls, log_type, crawler, env):
        word_list = get_config_from_mysql(log_type, crawler, env, "search_word", action="")
        for word in word_list:
            cls.i = 0
            Common.logger(log_type, crawler).info(f"开始抓取搜索词:{word}")
            try:
                cls.start_wechat(log_type=log_type,
                                 crawler=crawler,
                                 word=word,
                                 env=env)
            except Exception as e:
                Common.logger(log_type, crawler).error(f"search_video:{e}\n")


if __name__ == '__main__':
    ShipinhaoSearch.search_all_videos(log_type="search",
                                      crawler="shipinhao",
                                      env="dev")
    pass