automatic_crawler/spider/crawler_offline/shipinshuashua_id.py

# -*- coding: utf-8 -*-
# @Time: 2024/01/04

import re
import json
import os
import sys
import time
import requests


sys.path.append(os.getcwd())
from datetime import date, timedelta
from application.functions import clean_title
from appium import webdriver
from appium.webdriver.extensions.android.nativekey import AndroidKey
from bs4 import BeautifulSoup
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
from application.common.mysql import MysqlHelper

class SPSSIdRecommend:

    def __init__(self, env):
        self.env = env
        chromedriverExecutable = "/Users/tzld/Downloads/chromedriver_v111/chromedriver"
        # 微信的配置文件
        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(120):
            try:
                if self.driver.find_elements(By.ID, "com.tencent.mm:id/f2s"):
                    break
                elif self.driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"):
                    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 NoSuchElementException:
                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)
        self.driver.find_elements(By.XPATH, '//*[@text="视频刷刷"]')[-1].click()
        time.sleep(5)
        print("打开小程序")

        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)

                return
            except NoSuchElementException:
                time.sleep(1)

    def swipe_up(self):
        self.search_elements('//*[@class="dynamic--nick-top"]')
        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)

    def get_video_id(self, video_name_element):
        for i in range(3):
            self.search_elements('//*[@class="dynamic--nick-top"]')
            time.sleep(1)
            self.driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'});",
                                       video_name_element[0])
            time.sleep(3)
            video_name_element[0].click()
            self.check_to_applet(xpath=r'//wx-view[@class="small--mid"]')
            time.sleep(10)
            video_id_elements = self.search_elements(
                '//wx-view[@class="small--mid"]')
            if video_id_elements:
                return video_id_elements[0].text

    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):
        # 用户名
        username = video_element.find("wx-view", class_="dynamic--nick-top").text



        username_element = self.search_elements(f'//*[contains(text(), "{username}")]')
        if username_element is None:
            return
        video_id = self.get_video_id(username_element)

        if video_id is None:
            self.driver.press_keycode(AndroidKey.BACK)
            time.sleep(5)
            return
        video_id = re.sub(r'\D', '', video_id)
        video_id = int(video_id)
        repeat_video_id = self.repeat_video_id(video_id)
        if repeat_video_id != 0:
            self.driver.press_keycode(AndroidKey.BACK)
            return
        data_list = self.get_user_list(video_id)
        if len(data_list) == 0:
            self.driver.press_keycode(AndroidKey.BACK)
            return
        else:
            status = 1
            localtime = time.localtime(time.time())
            formatted_time = time.strftime("%Y-%m-%d", localtime)
            print(formatted_time)
            self.insert_user(video_id, username, data_list, status, formatted_time)
            self.driver.press_keycode(AndroidKey.BACK)
            time.sleep(2)

    def insert_user(self, mid, user_name, data_list, status, formatted_time):
        insert_sql = f"""insert into crawler_xng_userid( user_id , user_name , user_title_text , status, time) values ({mid},"{user_name}", "{data_list}",{status}, "{formatted_time}")"""
        print(insert_sql)
        MysqlHelper(self.env).update(insert_sql)

    def repeat_video_id(self, mid):
        sql = f"SELECT `link`  FROM `crawler_user_v3` WHERE  `source` = 'xiaoniangao'  and `link` = {mid}"
        repeat_video_id = MysqlHelper(self.env).select(sql)
        return len(repeat_video_id)

    def get_user_list(self, mid):
        next_t = -1
        url = "https://kapi-xng-app.xiaoniangao.cn/v1/album/user_public"
        headers = {
            'Host': 'kapi-xng-app.xiaoniangao.cn',
            'content-type': 'application/json; charset=utf-8',
            'accept': '*/*',
            'authorization': 'hSNQ2s9pvPxvFn4LaQJxKQ6/7Is=',
            'verb': 'POST',
            'content-md5': 'c7b7f8663984e8800e3bcd9b44465083',
            'x-b3-traceid': '2f9da41f960ae077',
            'accept-language': 'zh-cn',
            'date': 'Mon, 19 Jun 2023 06:41:17 GMT',
            'x-token-id': '',
            'x-signaturemethod': 'hmac-sha1',
            'user-agent': 'xngapp/157 CFNetwork/1335.0.3.1 Darwin/21.6.0'
        }
        payload = {
            "token": "",
            "limit": 20,
            "start_t": next_t,
            "visited_mid": mid,
            "share_width": 300,
            "share_height": 240,
        }
        response = requests.request(
            "POST",
            url,
            headers=headers,
            data=json.dumps(payload),
        )
        data_list = []
        if "data" not in response.text or response.status_code != 200:
            return data_list
        elif "list" not in response.json()["data"]:
            return data_list
        elif len(response.json()["data"]["list"]) == 0:
            return data_list
        list = response.json()["data"]["list"]
        for video_obj in list:
            video_title = clean_title(video_obj.get("title", ""))
            # 发布时间
            publish_time_stamp = int(int(video_obj.get("t", 0)) / 1000)
            publish_time_str = time.strftime(
                "%Y-%m-%d", time.localtime(publish_time_stamp)
            )
            date_three_days_ago_string = (date.today() + timedelta(days=-7)).strftime("%Y-%m-%d")
            rule = publish_time_str >= date_three_days_ago_string
            if rule == False:
                return ""
            v_url = video_obj.get("v_url")
            data_list.append(video_title + ":" + v_url)
        return data_list


    def get_video_info(self, video_element):
        try:
            self.get_video_info_2(video_element)
        except Exception as e:
            print(e)
            self.driver.press_keycode(AndroidKey.BACK)

    def get_videoList(self):
        self.driver.implicitly_wait(20)
        # 切换到 web_view
        self.check_to_applet(xpath='//*[@class="dynamic--nick-top"]')
        print("切换到 webview 成功")
        time.sleep(1)
        if self.search_elements('//*[@class="dynamic--nick-top"]') is None:
            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)
        print("下滑完成")
        time.sleep(5)


if __name__ == '__main__':
    SPSSIdRecommend(
            env="prod"
)