piaoquan_crawler/xiaoniangaoplus/xiaoniangaoplus/xiaoniangao_plus_scheduling2.py

# -*- coding: utf-8 -*-
# @Author: luojunhui
# @Time: 2023/9/27
import json
import os
import sys
import time
from hashlib import md5

import requests
from appium import webdriver
from appium.webdriver.extensions.android.nativekey import AndroidKey
from appium.webdriver.webdriver import WebDriver
from bs4 import BeautifulSoup
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
import multiprocessing

sys.path.append(os.getcwd())
from common.common import Common
from common.mq import MQ
from common.public import download_rule, get_config_from_mysql
from common.scheduling_db import MysqlHelper


def get_redirect_url(url):
    res = requests.get(url, allow_redirects=False)
    if res.status_code == 302 or res.status_code == 301:
        return res.headers['Location']
    else:
        return url

class XiaoNianGaoPlusRecommend:
    env = None
    driver = None
    log_type = None

    def __init__(self, log_type, crawler, env, rule_dict, our_uid):
        self.mq = None
        self.platform = "小年糕"
        self.download_cnt = 0
        self.element_list = []
        self.count = 0
        self.swipe_count = 0
        self.log_type = log_type
        self.crawler = crawler
        self.env = env
        self.rule_dict = rule_dict
        self.our_uid = our_uid
        if self.env == "dev":
            chromedriverExecutable = "/Users/a123456/Downloads/chromedriver_V111/chromedriver"
        else:
            chromedriverExecutable = "/Users/a123456/Downloads/chromedriver_v111/chromedriver"

        Common.logger(self.log_type, self.crawler).info("启动微信")
        Common.logging(self.log_type, self.crawler, self.env, '启动微信')
        # 微信的配置文件
        caps = {
            "platformName": "Android",
            "devicesName": "Android",
            "platformVersion": "13",
            # "udid": "emulator-5554",
            "appPackage": "com.tencent.mm",
            "appActivity": ".ui.LauncherUI",
            "autoGrantPermissions": "true",
            "noReset": True,
            "resetkeyboard": True,
            "unicodekeyboard": True,
            "showChromedriverLog": True,
            "printPageSourceOnFailure": True,
            "recreateChromeDriverSessions": True,
            "enableWebviewDetailsCollection": True,
            "setWebContentsDebuggingEnabled": True,
            "newCommandTimeout": 6000,
            "automationName": "UiAutomator2",
            "chromedriverExecutable": chromedriverExecutable,
            "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
        }
        self.driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
        self.driver.implicitly_wait(30)

        for i in range(120):
            try:
                if self.driver.find_elements(By.ID, "com.tencent.mm:id/f2s"):
                    Common.logger(self.log_type, self.crawler).info("微信启动成功")
                    Common.logging(self.log_type, self.crawler, self.env, '微信启动成功')
                    break
                elif self.driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"):
                    Common.logger(self.log_type, self.crawler).info("发现并关闭系统下拉菜单")
                    Common.logging(self.log_type, self.crawler, self.env, '发现并关闭系统下拉菜单')
                    self.driver.find_element(By.ID, "com.android.system:id/dismiss_view").click()
                else:
                    pass
            except NoSuchElementException:
                time.sleep(1)

        Common.logger(self.log_type, self.crawler).info("下滑，展示小程序选择面板")
        # Common.logging(self.log_type, self.crawler, self.env, '下滑，展示小程序选择面板')
        size = self.driver.get_window_size()
        self.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(1)
        Common.logger(self.log_type, self.crawler).info('打开小程序"小年糕+"')
        # Common.logging(self.log_type, self.crawler, self.env, '打开小程序"小年糕+"')
        self.driver.find_elements(By.XPATH, '//*[@text="小年糕+"]')[-1].click()
        time.sleep(5)
        self.get_videoList()
        time.sleep(1)
        self.driver.quit()

    def search_elements(self, xpath):
        time.sleep(1)
        windowHandles = self.driver.window_handles
        for handle in windowHandles:
            self.driver.switch_to.window(handle)
            time.sleep(1)
            try:
                elements = self.driver.find_elements(By.XPATH, xpath)
                if elements:
                    return elements
            except NoSuchElementException:
                pass

    def check_to_applet(self, xpath):
        time.sleep(1)
        webViews = self.driver.contexts
        self.driver.switch_to.context(webViews[-1])
        windowHandles = self.driver.window_handles
        for handle in windowHandles:
            self.driver.switch_to.window(handle)
            time.sleep(1)
            try:
                self.driver.find_element(By.XPATH, xpath)
                Common.logger(self.log_type, self.crawler).info("切换到WebView成功\n")
                Common.logging(self.log_type, self.crawler, self.env, '切换到WebView成功\n')
                return
            except NoSuchElementException:
                time.sleep(1)

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

    def swipe_up(self):
        self.search_elements('//*[@class="list-list--list"]')
        size = self.driver.get_window_size()
        self.driver.swipe(int(size["width"] * 0.5), int(size["height"] * 0.8),
                          int(size["width"] * 0.5), int(size["height"] * 0.442), 200)
        self.swipe_count += 1

    def get_video_url(self, video_title_element):
        for i in range(3):
            self.search_elements('//*[@class="list-list--list"]')
            Common.logger(self.log_type, self.crawler).info(f"video_title_element:{video_title_element[0]}")
            time.sleep(1)
            Common.logger(self.log_type, self.crawler).info("滑动标题至可见状态")
            self.driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'});",
                                       video_title_element[0])
            time.sleep(3)
            Common.logger(self.log_type, self.crawler).info("点击标题")
            video_title_element[0].click()
            self.check_to_applet(xpath=r'//wx-video[@class="dynamic-index--video-item dynamic-index--video"]')
            Common.logger(self.log_type, self.crawler).info("点击标题完成")
            time.sleep(10)
            video_url_elements = self.search_elements(
                '//wx-video[@class="dynamic-index--video-item dynamic-index--video"]')
            if video_url_elements:
                return video_url_elements[0].get_attribute("src")

    def parse_detail(self, index):
        page_source = self.driver.page_source
        soup = BeautifulSoup(page_source, 'html.parser')
        soup.prettify()
        video_list = soup.findAll(name="wx-view", attrs={"class": "expose--adapt-parent"})
        element_list = [i for i in video_list][index:]
        return element_list[0]

    def get_video_info_2(self, video_element):
        Common.logger(self.log_type, self.crawler).info(f"本轮已抓取{self.download_cnt}条视频\n")
        # Common.logging(self.log_type, self.crawler, self.env, f"本轮已抓取{self.download_cnt}条视频\n")
        if self.download_cnt >= int(self.rule_dict.get("videos_cnt", {}).get("min", 10)):
            self.count = 0
            self.download_cnt = 0
            self.element_list = []
            return
        self.count += 1
        Common.logger(self.log_type, self.crawler).info(f"第{self.count}条视频")
        # Common.logging(self.log_type, self.crawler, self.env, f"第{self.count}条视频")
        # 标题
        video_title = video_element.find("wx-view", class_="dynamic--title").text
        # 播放量字符串
        play_str = video_element.find("wx-view", class_="dynamic--views").text
        info_list = video_element.findAll("wx-view", class_="dynamic--commerce-btn-text")
        # 点赞数量
        like_str = info_list[1].text
        # 评论数量
        comment_str = info_list[2].text
        # 视频时长
        duration_str = video_element.find("wx-view", class_="dynamic--duration").text
        user_name = video_element.find("wx-view", class_="dynamic--nick-top").text
        # 头像 URL
        avatar_url = video_element.find("wx-image", class_="avatar--avatar")["src"]
        # 封面 URL
        cover_url = video_element.find("wx-image", class_="dynamic--bg-image")["src"]
        play_cnt = int(play_str.replace("+", "").replace("次播放", ""))
        duration = int(duration_str.split(":")[0].strip()) * 60 + int(duration_str.split(":")[-1].strip())
        if "点赞" in like_str:
            like_cnt = 0
        elif "万" in like_str:
            like_cnt = int(like_str.split("万")[0]) * 10000
        else:
            like_cnt = int(like_str)
        if "评论" in comment_str:
            comment_cnt = 0
        elif "万" in comment_str:
            comment_cnt = int(comment_str.split("万")[0]) * 10000
        else:
            comment_cnt = int(comment_str)
        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,
            "duration_str": duration_str,
            "duration": duration,
            "play_str": play_str,
            "play_cnt": play_cnt,
            "like_str": like_str,
            "like_cnt": like_cnt,
            "comment_cnt": comment_cnt,
            "share_cnt": 0,
            "user_name": user_name,
            "user_id": out_user_id,
            'publish_time_stamp': int(time.time()),
            'publish_time_str': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()))),
            "avatar_url": avatar_url,
            "cover_url": cover_url,
            "session": f"xiaoniangao-{int(time.time())}"
        }
        for k, v in video_dict.items():
            Common.logger(self.log_type, self.crawler).info(f"{k}:{v}")
        Common.logging(self.log_type, self.crawler, self.env, f"video_dict:{video_dict}")
        # Common.logger(self.log_type, self.crawler).info(f"==========分割线==========\n")

        if video_title is None or cover_url is None:
            Common.logger(self.log_type, self.crawler).info("无效视频\n")
            Common.logging(self.log_type, self.crawler, self.env, '无效视频\n')
            # self.swipe_up()
            time.sleep(0.5)
        elif download_rule(log_type=self.log_type,
                           crawler=self.crawler,
                           video_dict=video_dict,
                           rule_dict=self.rule_dict) is False:
            Common.logger(self.log_type, self.crawler).info("不满足抓取规则\n")
            Common.logging(self.log_type, self.crawler, self.env, "不满足抓取规则\n")
            # self.swipe_up()
            time.sleep(0.5)
        elif any(str(word) if str(word) in video_dict["video_title"] else False
                 for word in get_config_from_mysql(log_type=self.log_type,
                                                   source=self.crawler,
                                                   env=self.env,
                                                   text="filter",
                                                   action="")) is True:
            Common.logger(self.log_type, self.crawler).info('已中过滤词\n')
            Common.logging(self.log_type, self.crawler, self.env, '已中过滤词\n')
            # self.swipe_up()
            time.sleep(0.5)
        elif self.repeat_video(out_video_id) != 0:
            Common.logger(self.log_type, self.crawler).info('视频已下载\n')
            Common.logging(self.log_type, self.crawler, self.env, '视频已下载\n')
            # self.swipe_up()
            time.sleep(5)
        else:
            video_title_element = self.search_elements(f'//*[contains(text(), "{video_title}")]')
            if video_title_element is None:
                Common.logger(self.log_type, self.crawler).warning(
                    f"未找到该视频标题的element:{video_title_element}")
                Common.logging(self.log_type, self.crawler, self.env,
                               f"未找到该视频标题的element:{video_title_element}")
                # continue
                return
            Common.logger(self.log_type, self.crawler).info("点击标题，进入视频详情页")
            Common.logging(self.log_type, self.crawler, self.env, "点击标题，进入视频详情页")
            video_url = self.get_video_url(video_title_element)
            video_url = get_redirect_url(video_url)
            if video_url is None:
                Common.logger(self.log_type, self.crawler).info("未获取到视频播放地址\n")
                self.driver.press_keycode(AndroidKey.BACK)
                time.sleep(5)
                return
            video_dict['video_url'] = video_url
            Common.logger(self.log_type, self.crawler).info(f"video_url:{video_url}")

            video_dict["platform"] = self.crawler
            video_dict["strategy"] = self.log_type
            video_dict["out_video_id"] = video_dict["video_id"]
            video_dict["crawler_rule"] = json.dumps(self.rule_dict)
            video_dict["user_id"] = self.our_uid
            video_dict["publish_time"] = video_dict["publish_time_str"]
            self.mq.send_msg(video_dict)
            # print(video_dict)
            self.download_cnt += 1
            self.driver.press_keycode(AndroidKey.BACK)
            time.sleep(5)

    def get_video_info(self, video_element):
        try:
            self.get_video_info_2(video_element)
        except Exception as e:
            Common.logger(self.log_type, self.crawler).error(f"抓取单条视频异常:{e}\n")
            # Common.logging(self.log_type, self.crawler, self.env, f"抓取单条视频异常:{e}\n")

    def get_videoList(self):
        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
        self.driver.implicitly_wait(20)
        # 切换到 web_view
        self.check_to_applet(xpath='//*[@class="tab-bar--tab tab-bar--tab-selected"]')
        print("切换到 webview 成功")
        time.sleep(1)
        page = 0
        if self.search_elements('//*[@class="list-list--list"]') is None:
            Common.logger(self.log_type, self.crawler).info("窗口已销毁\n")
            Common.logging(self.log_type, self.crawler, self.env, '窗口已销毁\n')
            self.count = 0
            self.download_cnt = 0
            self.element_list = []
            return

        print("开始获取视频信息")
        for i in range(50):
            print("下滑{}次".format(i))
            element = self.parse_detail(i)
            self.get_video_info(element)
            self.swipe_up()
            time.sleep(1)
            if self.swipe_count > 100:
                return

        print("下滑完成")
        # time.sleep(100)
        Common.logger(self.log_type, self.crawler).info("已抓取完一组，休眠 5 秒\n")
        Common.logging(self.log_type, self.crawler, self.env, "已抓取完一组，休眠 5 秒\n")
        time.sleep(5)


def run():
    rule_dict1 = {"period": {"min": 365, "max": 365},
                  "duration": {"min": 30, "max": 1800},
                  "favorite_cnt": {"min": 0, "max": 0},
                  "videos_cnt": {"min": 5000, "max": 0},
                  "share_cnt": {"min": 0, "max": 0}}
    XiaoNianGaoPlusRecommend("recommend", "xiaoniangao", "dev", rule_dict1, 6267141)


if __name__ == "__main__":
    process = multiprocessing.Process(
        target=run
    )
    process.start()
    while True:
        if not process.is_alive():
            print("正在重启")
            process.terminate()
            time.sleep(60)
            os.system("adb forward --remove-all")
            process = multiprocessing.Process(target=run)
            process.start()
        time.sleep(60)