piaoquan_crawler/shipinhao/shipinhao_search/shipinhao_search_scheduling.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/4/25
import datetime
import json
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.public import download_rule
from common.mq import MQ
from common.common import Common
from common.scheduling_db import MysqlHelper


class ShipinhaoSearchScheduling:
    platform = "视频号"
    i = 0
    download_cnt = 0

    @classmethod
    def start_wechat(cls, log_type, crawler, rule_dict, user_dict, env):
        Common.logger(log_type, crawler).info('启动微信')
        Common.logging(log_type, crawler, env, '启动微信')
        if env == "dev":
            chromedriverExecutable = "/Users/wangkun/Downloads/chromedriver/chromedriver_v107/chromedriver"
        else:
            chromedriverExecutable = "/Users/piaoquan/Downloads/chromedriver/chromedriver_v111/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"},
            # "chromeOptions": {"androidProcess": "com.tencent.mm:toolsmp"},
            # "chromeOptions": {"androidProcess": "com.tencent.mm"},
            'enableWebviewDetailsCollection': True,
            'setWebContentsDebuggingEnabled': True,
            'chromedriverExecutable': chromedriverExecutable,
        }
        driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
        driver.implicitly_wait(10)
        # Common.logger(log_type, crawler).info("点击微信")
        # if len(driver.find_elements(By.ID, 'android:id/text1')) != 0:
        #     driver.find_elements(By.ID, 'android:id/text1')[0].click()
        # Common.logger(log_type, crawler).info("等待 5s")
        time.sleep(5)
        cls.search_video(log_type=log_type,
                         crawler=crawler,
                         rule_dict=rule_dict,
                         user_dict=user_dict,
                         driver=driver,
                         env=env)
        cls.close_wechat(log_type=log_type,
                         crawler=crawler,
                         env=env,
                         driver=driver)

    @classmethod
    def close_wechat(cls, log_type, crawler, env, driver: WebDriver):
        driver.quit()
        Common.logger(log_type, crawler).info(f"微信退出成功\n")
        Common.logging(log_type, crawler, env, 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):
        webviews = driver.contexts
        Common.logger(log_type, crawler).info(f"webviews:{webviews}")
        driver.switch_to.context(webviews[1])
        Common.logger(log_type, crawler).info(driver.current_context)
        time.sleep(1)
        windowHandles = driver.window_handles
        for handle in windowHandles:
            try:
                driver.switch_to.window(handle)
                time.sleep(1)
                driver.find_element(By.XPATH, '//div[@class="unit"]')
                Common.logger(log_type, crawler).info('切换 webview 成功')
                return "成功"
            except Exception:
                Common.logger(log_type, crawler).info("切换 webview 失败")

    @classmethod
    def repeat_out_video_id(cls, log_type, crawler, out_video_id, env):
        sql = f""" select * from crawler_video where platform in ("{crawler}","{cls.platform}") 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 in ("{crawler}","{cls.platform}") and video_url="{video_url}"; """
        repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env)
        return len(repeat_video)

    @classmethod
    def search_video(cls, log_type, crawler, rule_dict, driver: WebDriver, user_dict, env):
        mq = MQ(topic_name="topic_crawler_etl_" + env)
        # 点击微信搜索框，并输入搜索词
        driver.implicitly_wait(10)
        Common.logger(log_type, crawler).info("点击搜索框")
        Common.logging(log_type, crawler, env, "点击搜索框")
        driver.find_element(By.ID, 'com.tencent.mm:id/j5t').click()  # 微信8.0.30版本
        # driver.find_element(By.ID, 'com.tencent.mm:id/he6').click()  # 微信8.0.16版本
        time.sleep(0.5)
        driver.find_element(By.ID, 'com.tencent.mm:id/cd7').clear().send_keys(user_dict["link"])  # 微信8.0.30版本
        # driver.find_element(By.ID, 'com.tencent.mm:id/bxz').clear().send_keys(word)  # 微信8.0.16版本
        driver.press_keycode(AndroidKey.ENTER)
        Common.logger(log_type, crawler).info("进入搜索词页面")
        Common.logging(log_type, crawler, env, "进入搜索词页面")
        driver.find_elements(By.ID, 'com.tencent.mm:id/br8')[0].click()  # 微信8.0.30版本
        # driver.find_elements(By.ID, 'com.tencent.mm:id/jkg')[0].click()  # 微信8.0.16版本
        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")
            Common.logging(log_type, crawler, env, "切换到视频号 webview 失败\n")
            return
        time.sleep(1)

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

        videos_cnt = rule_dict.get('videos_cnt', {}).get('min', 30)
        index = 0
        while True:
            if cls.search_elements(driver, '//*[@class="mixed-box__bd"]') is None:
                Common.logger(log_type, crawler).info('窗口已销毁\n')
                Common.logging(log_type, crawler, env, '窗口已销毁\n')
                return

            Common.logger(log_type, crawler).info('获取视频列表\n')
            Common.logging(log_type, crawler, env, '获取视频列表\n')
            video_elements = cls.search_elements(driver, '//div[@class="rich-media active__absolute"]')
            if video_elements is None:
                Common.logger(log_type, crawler).warning(f'video_elements:{video_elements}')
                Common.logging(log_type, crawler, env, 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')
                Common.logging(log_type, crawler, env, '到底啦~~~~~~~~~~~~~\n')
                return

            for i, video_element in enumerate(video_element_temp):
                try:
                    Common.logger(log_type, crawler).info(f"download_cnt:{cls.download_cnt}")
                    Common.logging(log_type, crawler, env, f"download_cnt:{cls.download_cnt}")
                    if cls.download_cnt >= int(videos_cnt):
                        Common.logger(log_type, crawler).info(f'搜索词:{user_dict["link"]}，已抓取视频数:{cls.download_cnt}')
                        Common.logging(log_type, crawler, env, f'搜索词:{user_dict["link"]}，已抓取视频数:{cls.download_cnt}')
                        cls.download_cnt = 0
                        return

                    if video_element is None:
                        Common.logger(log_type, crawler).info('到底啦~\n')
                        Common.logging(log_type, crawler, env, '到底啦~\n')
                        return

                    cls.i += 1
                    cls.search_elements(driver, '//*[@class="rich-media active__absolute"]')

                    Common.logger(log_type, crawler).info(f'拖动"视频"列表第{cls.i}个至屏幕中间')
                    Common.logging(log_type, crawler, env, 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")
                        Common.logging(log_type, crawler, env, "没有更多的搜索结果\n")
                        return
                    video_title = \
                    video_element.find_elements(By.XPATH, '//div[@class="rich-media__title ellipsis_2"]/span')[
                        index + i].text[:40]
                    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')
                    cover_url = cover_url.split('url("')[-1].split('")')[0]
                    duration = video_element.find_elements(By.XPATH, '//div[@class="video-player-mask__text"]')[
                        index + i].text
                    duration = int(duration.split(':')[0]) * 60 + int(duration.split(':')[-1])
                    user_name = video_element.find_elements(By.XPATH, '//div[@class="rich-media__source__title"]')[
                        index + i].text
                    avatar_url = video_element.find_elements(By.XPATH,
                                                             '//div[@class="ui-image-image ui-image rich-media__source__thumb"]')[
                        index + i].get_attribute('style')
                    avatar_url = avatar_url.split('url("')[-1].split('")')[0]
                    out_video_id = md5(video_title.encode('utf8')).hexdigest()
                    out_user_id = md5(user_name.encode('utf8')).hexdigest()

                    video_dict = {
                        "video_title": video_title,
                        "video_id": out_video_id,
                        "play_cnt": 0,
                        "duration": duration,
                        # "duration": 60,
                        "user_name": user_name,
                        "user_id": out_user_id,
                        "avatar_url": avatar_url,
                        "cover_url": cover_url,
                        "video_url": video_url,
                        "session": f"shipinhao-search-{int(time.time())}"
                    }
                    for k, v in video_dict.items():
                        Common.logger(log_type, crawler).info(f"{k}:{v}")
                    Common.logging(log_type, crawler, env, f"{video_dict}")
                    if video_title is None or video_url is None:
                        Common.logger(log_type, crawler).info("无效视频\n")
                        Common.logging(log_type, crawler, env, "无效视频\n")
                    elif cls.repeat_out_video_id(log_type, crawler, out_video_id, env) != 0:
                        Common.logger(log_type, crawler).info('视频已下载\n')
                        Common.logging(log_type, crawler, env, '视频已下载\n')
                    elif cls.repeat_video_url(log_type, crawler, video_url, env) != 0:
                        Common.logger(log_type, crawler).info('视频已下载\n')
                        Common.logging(log_type, crawler, env, '视频已下载\n')
                    else:
                        video_element.click()
                        time.sleep(3)
                        video_info_dict = cls.get_video_info(driver)
                        video_dict["like_cnt"] = video_info_dict["like_cnt"]
                        video_dict["share_cnt"] = video_info_dict["share_cnt"]
                        video_dict["favorite_cnt"] = video_info_dict["favorite_cnt"]
                        video_dict["comment_cnt"] = video_info_dict["comment_cnt"]
                        video_dict["publish_time_str"] = video_info_dict["publish_time_str"] + " 00:00:00"
                        video_dict["publish_time_stamp"] = video_info_dict["publish_time_stamp"]
                        Common.logger(log_type, crawler).info(f'publish_time:{video_dict["publish_time_str"]}')
                        Common.logging(log_type, crawler, env, f'publish_time:{video_dict["publish_time_str"]}')
                        if download_rule(log_type=log_type, crawler=crawler, video_dict=video_dict, rule_dict=rule_dict) is False:
                            Common.logger(log_type, crawler).info("不满足抓取规则\n")
                            Common.logging(log_type, crawler, env, "不满足抓取规则\n")
                        else:
                            video_dict["out_user_id"] = video_dict["user_id"]
                            video_dict["platform"] = crawler
                            video_dict["strategy"] = log_type
                            video_dict["out_video_id"] = video_dict["video_id"]
                            video_dict["width"] = 0
                            video_dict["height"] = 0
                            video_dict["crawler_rule"] = json.dumps(rule_dict)
                            video_dict["user_id"] = user_dict["uid"]
                            video_dict["publish_time"] = video_dict["publish_time_str"]
                            mq.send_msg(video_dict)
                            cls.download_cnt += 1
                except Exception as e:
                    Common.logger(log_type, crawler).error(f"抓取单条视频异常:{e}\n")
                    Common.logging(log_type, crawler, env, f"抓取单条视频异常:{e}\n")

            Common.logger(log_type, crawler).info('已抓取完一组视频，休眠1秒\n')
            Common.logging(log_type, crawler, env, '已抓取完一组视频，休眠1秒\n')
            time.sleep(1)
            index = index + len(video_element_temp)

    @classmethod
    def get_video_info(cls, 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')  # 微信版本 8.0.30
        like_cnt = like_id.get_attribute('name')
        if '万' 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)
        elif like_cnt == "" or like_cnt == "喜欢" or like_cnt == "火" or cls.is_contain_chinese(like_cnt) is True:
            like_cnt = 0
        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 '万' 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)
        elif share_cnt == "" or share_cnt == "转发" or cls.is_contain_chinese(share_cnt) is True:
            share_cnt = 0
        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 '万' 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)
        elif favorite_cnt == "" or favorite_cnt == "收藏" or favorite_cnt == "推荐" or favorite_cnt == "火" or cls.is_contain_chinese(
                favorite_cnt) is True:
            favorite_cnt = 0
        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 '万' 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)
        elif comment_cnt == "" or comment_cnt == "评论" or cls.is_contain_chinese(comment_cnt) is True:
            comment_cnt = 0
        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 or "分钟" in publish_time or "小时" in publish_time:
            publish_time_str = (date.today() + timedelta(days=0)).strftime("%Y-%m-%d")
        elif "天前" 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("日", "")
            year_str = publish_time.split("年")[0]
            month_str = publish_time.split("年")[-1].split("月")[0]
            day_str = publish_time.split("月")[-1].split("日")[0]
            if int(month_str) < 10:
                month_str = f"0{month_str}"
            if int(day_str) < 10:
                day_str = f"0{day_str}"
            publish_time_str = f"{year_str}-{month_str}-{day_str}"
        else:
            year_str = str(datetime.datetime.now().year)
            month_str = publish_time.split("月")[0]
            day_str = publish_time.split("月")[-1].split("日")[0]
            if int(month_str) < 10:
                month_str = f"0{month_str}"
            if int(day_str) < 10:
                day_str = f"0{day_str}"
            publish_time_str = f"{year_str}-{month_str}-{day_str}"
            # 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 get_search_videos(cls, log_type, crawler, rule_dict, user_list, env):
        Common.logger(log_type, crawler).info(f"搜索词总数:{len(user_list)}\n")
        Common.logging(log_type, crawler, env, f"搜索词总数:{len(user_list)}\n")
        if len(user_list) == 0:
            return
        for user_dict in user_list:
            try:
                cls.i = 0
                cls.download_cnt = 0
                Common.logger(log_type, crawler).info(f"开始抓取 {user_dict['link']}\n")
                Common.logging(log_type, crawler, env, f"开始抓取 {user_dict['link']}\n")
                cls.start_wechat(log_type=log_type,
                                 crawler=crawler,
                                 rule_dict=rule_dict,
                                 user_dict=user_dict,
                                 env=env)
            except Exception as e:
                Common.logger(log_type, crawler).error(f"抓取 {user_dict['link']} 时异常:{e}\n")
                Common.logging(log_type, crawler, env, f"抓取 {user_dict['link']} 时异常:{e}\n")


if __name__ == '__main__':
    print(ShipinhaoSearchScheduling.repeat_out_video_id(log_type="search",
                                                        crawler="shipinhao",
                                                        out_video_id="123",
                                                        env="dev"))
    pass