piaoquan_crawler/shipinhao/shipinhao_recommend/shipinhao_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/7/25
import datetime
import os
import sys
import time
from datetime import date, timedelta
from hashlib import md5
from appium import webdriver
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.common import Common
from common.public import download_rule
from common.scheduling_db import MysqlHelper


class ShipinhaoRecommend:
    platform = "视频号"
    download_video_list = []
    scan_count = 20

    @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 start_wechat(cls, log_type, crawler, 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"},
            # "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)
        time.sleep(5)
        return driver

    @classmethod
    def get_videoList(cls, log_type, crawler, rule_dict, env, driver: WebDriver):
        Common.logger(log_type, crawler).info("进入发现页")
        tabs = driver.find_elements(By.ID, "com.tencent.mm:id/f2s")
        for tab in tabs:
            if tab.text == "发现":
                tab.click()
                time.sleep(0.5)
                break

        Common.logger(log_type, crawler).info('点击"视频号"')
        textviews = driver.find_elements(By.ID, "android:id/title")
        for textview in textviews:
            if textview.text == "视频号":
                textview.click()
                time.sleep(0.5)
                break

        # 关闭青少年模式弹框
        Common.logger(log_type, crawler).info("尝试关闭青少年模式弹框\n")
        try:
            driver.find_element(By.ID, "com.tencent.mm:id/lqz").click()
        except NoSuchElementException:
            pass

        for i in range(cls.scan_count):
            Common.logger(log_type, crawler).info(f"第{i + 1}条视频")
            if len(driver.find_elements(By.ID, "com.tencent.mm:id/dkf")) != 0:
                Common.logger(log_type, crawler).info("这是一个直播间，滑动至下一个视频\n")
                driver.swipe(10, 1600, 10, 300, 200)
                continue
            video_dict = cls.get_video_info(driver)
            for k, v in video_dict.items():
                Common.logger(log_type, crawler).info(f"{k}:{v}")

            if video_dict["video_title"] is None:
                Common.logger(log_type, crawler).info("无效视频")
            elif download_rule(log_type=log_type, crawler=crawler, video_dict=video_dict, rule_dict=rule_dict) is False:
                Common.logger(log_type, crawler).info("不满足抓取规则")
                # Common.logging(log_type, crawler, env, "不满足抓取规则\n")
            elif cls.repeat_out_video_id(log_type, crawler, video_dict["video_id"], env) != 0:
                Common.logger(log_type, crawler).info('视频已下载')
                # Common.logging(log_type, crawler, env, '视频已下载\n')
            else:
                cls.download_video_list.append(video_dict)
            if i+1 == cls.scan_count:
                Common.logger(log_type, crawler).info("扫描一轮结束\n")
                driver.quit()
                return
            Common.logger(log_type, crawler).info(f"已抓取符合规则视频{len(cls.download_video_list)}条，滑动至下一个视频\n")
            driver.swipe(10, 1600, 10, 300, 200)

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

    @classmethod
    def get_video_info(cls, driver: WebDriver):

        # 点击暂停
        global duration
        for i in range(3):
            try:
                driver.find_element(By.ID, "com.tencent.mm:id/gpx").click()
                duration_str = driver.find_element(By.ID, "com.tencent.mm:id/l7i").text
                duration = int(duration_str.split(":")[0]) * 60 + int(duration_str.split(":")[-1])
                break
            except NoSuchElementException:
                duration = 0

        # user_name
        user_name = driver.find_element(By.ID, "com.tencent.mm:id/hft").text

        # 点赞
        like_cnt = driver.find_element(By.ID, 'com.tencent.mm:id/k04').text
        if like_cnt == "" or like_cnt == "喜欢" or like_cnt == "火":
            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_cnt = driver.find_element(By.ID, 'com.tencent.mm:id/jhv').text
        if share_cnt == "" or share_cnt == "转发":
            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_cnt = driver.find_element(By.ID, 'com.tencent.mm:id/fnp').text
        if favorite_cnt == "" or favorite_cnt == "收藏" or favorite_cnt == "推荐" or favorite_cnt == "火":
            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.text
        if comment_cnt == "" or comment_cnt == "评论":
            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)

        # title
        title = driver.find_elements(By.ID, "com.tencent.mm:id/bga")[0].text.replace("\n", " ")[:40]

        # 发布时间
        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:
            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_stamp = int(time.mktime(time.strptime(publish_time_str, "%Y-%m-%d")))

        # 收起评论
        driver.find_element(By.ID, "com.tencent.mm:id/be_").click()
        time.sleep(0.5)

        video_id = md5(title.encode('utf8')).hexdigest()
        video_dict = {
            "video_title": title,
            "video_id": video_id,
            "duration": duration,
            "user_name": user_name,
            "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_recommend_list(cls, log_type, crawler, rule_dict, env):
        driver = cls.start_wechat(log_type, crawler, env)
        cls.get_videoList(log_type=log_type,
                          crawler=crawler,
                          rule_dict=rule_dict,
                          env=env,
                          driver=driver)
        driver.quit()
        Common.logger(log_type, crawler).info(f"微信退出成功\n")


if __name__ == "__main__":
    rule_dict1 = {"period": {"min": 365, "max": 365},
                 "duration": {"min": 10, "max": 1800},
                 "favorite_cnt": {"min": 50000, "max": 0},
                 "share_cnt": {"min": 10000, "max": 0}}
    ShipinhaoRecommend.get_recommend_list("recommend", "shipinhao", rule_dict1, "dev")
    print(ShipinhaoRecommend.download_video_list)
    pass