automatic_crawler/spider/crawler_offline/xiaoniangao_zhanghao.py

# -*- coding: utf-8 -*-
import json
import os
import random
import subprocess
import sys
import time
import uuid
import requests
from datetime import datetime, timedelta
from appium import webdriver
from appium.webdriver.extensions.android.nativekey import AndroidKey
from appium.webdriver.common.touch_action import TouchAction
from bs4 import BeautifulSoup
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By


sys.path.append(os.getcwd())

from application.common import MysqlHelper, Feishu


class XiaoNianGaoZH(object):
    """
    小年糕+线下爬虫
    """

    def __init__(self):
        self.count = 0
        self.swipe_count = 0
        chromedriverExecutable = "/Users/a123456/Downloads/chromedriver-mac-x64/chromedriver"
        print("启动微信")
        # 微信的配置文件
        caps = {
            "platformName": "Android",
            "devicesName": "Android",
            "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"},
        }
        try:
            self.driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
        except Exception as e:
            print(e)
            return
        self.driver.implicitly_wait(30)

        for i in range(10):
            try:
                if self.driver.find_elements(By.ID, "com.tencent.mm:id/f2s"):
                    print("启动微信成功")
                    break
                elif self.driver.find_element(
                        By.ID, "com.android.systemui:id/dismiss_view"
                ):
                    print("发现并关闭系统下拉菜单")
                    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.2),
                        200,
                    )
                else:
                    pass
            except Exception as e:
                print(f"打开微信异常：{e}")
                time.sleep(1)

        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)
        command = 'adb shell service call statusbar 2'
        process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE)
        process.communicate()
        self.driver.find_elements(By.XPATH, '//*[@text="小年糕+"]')[-1].click()
        print("打开小程序小年糕+成功")
        time.sleep(5)
        self.get_videoList()
        time.sleep(1)
        self.driver.quit()


    def save_pq_uid(self, uid):
        current_time = datetime.now()
        time_after_10_minutes = current_time + timedelta(minutes=5)
        # 获取时间戳
        timestamp_seconds = time_after_10_minutes.timestamp()
        timestamp_milliseconds = int(timestamp_seconds * 1000)
        file_path = '/Users/tzld/Desktop/automatic_crawler/spider/crawler_offline/xng_zh.txt'
        with open(file_path, 'r') as file:
            content = file.read()
        if content and content[-1] != ',':
            uid = ',' + str(uid)
        with open(file_path, 'a') as file:
            file.write(uid)
        with open(file_path, 'r') as file:
            content = file.read()

        url = "https://admin.piaoquantv.com/manager/crawler/v3/task/save"
        payload = {
            "taskName": "小年糕账号",
            "source": "xiaoniangao",
            "mode": "author",
            "modeValue": "0",
            "modeBoard": "0",
            "spiderName": "run_xng_author",
            "startTime": timestamp_milliseconds,
            "interval": 4800,
            "uid": str(content),
            "machine": "aliyun",
            "rule": [{"period": {"min": 15, "max": 3}},
                     {"duration": {"min": 50, "max": 0}},
                     {"share_cnt": {"min": 2, "max": 0}},
                     {"videos_cnt": {"min": 300, "max": 0}}],
            "id": 21
        }
        headers = {
            'accept': 'application/json',
            'content-type': 'application/json;',
            'cookie': 'SESSION=ZmYwMzBmOWItM2M5YS00ZGMyLTk3MjctMzE0YzE4MmUxNThh',
            'origin': 'https://admin.piaoquantv.com',
            'pragma': 'no-cache',
            'priority': 'u=1, i',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36'
        }

        requests.request("POST", url, headers=headers, json=payload)

    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)
                print("切换到WebView成功\n")
                return
            except NoSuchElementException:
                time.sleep(1)

    def swipe_up(self):
        self.search_elements('//*[@class="list-list--list"]')
        size = self.driver.get_window_size()
        action = TouchAction(self.driver)
        action.press(x=int(size["width"] * 0.5), y=int(size["height"] * 0.85))
        action.wait(ms=1300)  # 可以调整等待时间
        action.move_to(x=int(size["width"] * 0.5), y=int(size["height"] * 0.2))
        action.release()
        action.perform()
        self.swipe_count += 1

    def get_video_url(self, video_title_element):
        for i in range(3):
            self.search_elements('//*[@class="list-list--list"]')
            time.sleep(1)
            self.driver.execute_script(
                "arguments[0].scrollIntoView({block:'center',inline:'center'});",
                video_title_element[0],
            )
            time.sleep(3)
            video_title_element[0].click()
            self.check_to_applet(
                xpath=r'//wx-video[@class="dynamic-index--video-item dynamic-index--video"]'
            )
            time.sleep(10)
            video_url_elements = self.search_elements(
                '//wx-video[@class="dynamic-index--video-item dynamic-index--video"]'
            )
            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"}
        )
        index = index + 1
        element_list = [i for i in video_list][index:]
        return element_list[0]

    def get_video_info_2(self, video_element):
        self.count += 1
        video_title = video_element.find("wx-view", class_="dynamic--title").text

        # 头像 URL
        avatar_url = video_element.find("wx-image", class_="avatar--avatar")["src"]
        # 用户名称
        user_name = video_element.find("wx-view", class_="dynamic--nick-top").text
        name_url = self.select_name_url(avatar_url, user_name)
        if name_url:
            video_title_element = self.search_elements(f'//*[contains(text(), "{video_title}")]')
            if video_title_element is None:
                return
            self.get_video_url(video_title_element)
            video_mid_elements = self.search_elements("//wx-view[@class='bar--navBar-content-capsule-wrap']")
            mid = int(video_mid_elements[0].get_attribute("data-mid"))
            self.driver.press_keycode(AndroidKey.BACK)
            time.sleep(5)
            uid = self.select_id(mid)
            if uid:
                self.update_name_url(mid, avatar_url, user_name)
            else:
                time.sleep(1)
                link = self.select_id_status(mid)
                if link:
                    current_time = datetime.now()
                    formatted_time = current_time.strftime("%Y%m%d")
                    date_int = int(formatted_time)
                    # 获取时间标签
                    tag_id = self.get_tag_id(date_int)
                    time.sleep(5)
                    print(tag_id)
                    # 新增账号
                    pq_uid = self.insert_number(mid, tag_id)
                    time.sleep(5)
                    if pq_uid:
                        self.insert_name_url(mid, avatar_url, user_name)
                        time.sleep(2)
                        self.save_pq_uid(pq_uid)
                        # 获取当前时间
                        current_time = datetime.now()
                        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
                        values = [[
                            str(mid),
                            user_name,
                            avatar_url,
                            str(pq_uid),
                            formatted_time,

                        ]]
                        Feishu.insert_columns('xiaoniangao', 'xiaoniangao', "8zlceR", "ROWS", 1, 2)
                        time.sleep(0.5)
                        Feishu.update_values('xiaoniangao', 'xiaoniangao', "8zlceR", "A2:Z2", values)
                        print("写入飞书表格成功")


    def get_video_info(self, video_element):
        try:
            self.get_video_info_2(video_element)
        except Exception as e:
            print(f"抓取单条视频异常:{e}\n")

    def get_videoList(self):
        """
        获取视频列表
        :return:
        """
        # while True:
        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)
        if self.search_elements('//*[@class="list-list--list"]') is None:
            print("窗口已销毁")
            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(random.randint(1, 5))


    def insert_number(self, mid, tag_id):
        for i in range(3):
            url = "https://admin.piaoquantv.com/manager/crawler/v3/user/save"
            payload = {
                "source": "xiaoniangao",
                "mode": "author",
                "modeValue": "",
                "modeBoard": "",
                "recomStatus": -6,
                "appRecomStatus": -6,
                "autoAuditStatus": 0,
                "tag": f"459,454,106,8240,{int(tag_id)}",
                "contentCategory": 0,
                "link": str(mid)
            }

            headers = {
                'content-length': '0',
                'cookie': 'SESSION=MWM4YzVlMTctNzdkNC00NjE3LWIxZTctOGQwYzgzYmVmN2Qw',
                'origin': 'https://admin.piaoquantv.com',
                'priority': 'u=1, i',
                'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
                'sec-ch-ua-mobile': '?0',
                'sec-ch-ua-platform': '"macOS"'
            }

            response = requests.request("POST", url, headers=headers, json=payload)
            response = response.json()
            code = response["code"]
            if code == 0:
                print("添加账号成功")
                time.sleep(1)
                url = "https://admin.piaoquantv.com/manager/crawler/v3/user/list"
                payload = {
                    "pageNum": 1,
                    "pageSize": 20
                }
                response = requests.request("POST", url, headers=headers, json=payload)
                response = response.json()
                list = response["content"]['list']
                link = list[0]["link"]
                if link == str(mid):
                    print("获取站内账号ID成功")
                    return list[0]["uid"]



    """
    查询用户名+头像是否存在
    """
    def select_name_url(self, avatar_url, user_name):
        sql = f""" select uid from xng_uid where avatar_url = "{avatar_url}" and user_name="{user_name}"; """
        db = MysqlHelper()
        repeat_video = db.select(sql=sql)
        if repeat_video:
            return False
        return True



    def get_tag_id(self, date_int):
        for i in range(3):
            url = f"https://admin.piaoquantv.com/manager/user/up/searchUserTypeTag?keyword={date_int}&muid=7"

            payload = {}
            headers = {
                'content-length': '0',
                'cookie': 'SESSION=MWM4YzVlMTctNzdkNC00NjE3LWIxZTctOGQwYzgzYmVmN2Qw',
                'origin': 'https://admin.piaoquantv.com',
                'priority': 'u=1, i',
                'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
                'sec-ch-ua-mobile': '?0',
                'sec-ch-ua-platform': '"macOS"'
            }

            response = requests.request("POST", url, headers=headers, data=payload)
            response = response.json()
            content = response["content"]
            if content:
                tagId = content[0]['tagId']
                return tagId
            else:
                url = f"https://admin.piaoquantv.com/manager/user/up/createUserTypeTag?tagName={date_int}&muid=7"
                response = requests.request("POST", url, headers=headers, data=payload)
                response = response.json()
                content = response["content"]
                if content:
                    tagId = content['tagId']
                    return tagId

    """
    修改用户名+头像
    """
    def update_name_url(self, mid, avatar_url, user_name):
        sql = f""" update xng_uid set avatar_url = "{avatar_url}", user_name="{user_name}" where uid = "{mid}"; """
        db = MysqlHelper()
        repeat_video = db.update(sql=sql)
        if repeat_video:
            return True
        return False

    """
    插入 用户名 头像 用户id
    """
    def insert_name_url(self, uid, avatar_url, user_name):
        current_time = datetime.now()
        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
        insert_sql = f"""INSERT INTO xng_uid (uid, avatar_url, user_name, data_time) values ('{uid}' ,'{avatar_url}','{user_name}', '{formatted_time}')"""
        db = MysqlHelper()
        repeat_video = db.update(sql=insert_sql)
        if repeat_video:
            return True
        return False


    """
    查询用户id是否存在
    """
    def select_id(self, uid):
        sql = f""" select uid from xng_uid where uid = "{uid}"; """
        db = MysqlHelper()
        repeat_video = db.select(sql=sql)
        if repeat_video:
            return True
        return False

    """
    查询用户id是否之前已添加过
    """
    def select_id_status(self, uid):
        sql = f""" select uid from crawler_user_v3 where link = "{uid}"; """
        db = MysqlHelper()
        repeat_video = db.select(sql=sql)
        if repeat_video:
            return False
        return True


if __name__ == "__main__":
    XiaoNianGaoZH()