增加视频刷刷获取用户id代码

app/off_line_controler.py

@@ -43,10 +43,17 @@ class SpiderScheduler(object):
     def run_spss(cls, hour):
         cls.protect_spider_timeout(function=cls.SH.run_spss, hour=hour)
 
+    @classmethod
+    def run_spss_id(cls, hour):
+        cls.protect_spider_timeout(function=cls.SH.run_spss_id, hour=hour)
+
 
 if __name__ == "__main__":
     SC = SpiderScheduler()
     # schedule.every().day.at("20:06").do(SC.run_xng_plus, hour=1)
     schedule.every().day.at("20:30").do(SC.run_spss, hour=1)
+
+    schedule.every().day.at("18:30").do(SC.run_spss_id, hour=1)
+
     while True:
         schedule.run_pending()

scheduler/spider_scheduler.py

@@ -1,10 +1,11 @@
 import os
 import sys
 
+
 sys.path.append(os.getcwd())
 
 from spider.crawler_offline import *
-
+from spider.crawler_offline.shipinshuashua_id import SPSSIdRecommend
 
 class SpiderHome(object):
     @classmethod
@@ -37,3 +38,9 @@ class SpiderHome(object):
             rule_dict=rule_dict,
             our_uid=[66433018, 66433020, 66433022, 66433023, 66433024]
         )
+
+    @classmethod
+    def run_spss_id(cls):
+        SPSSIdRecommend(
+            env="prod"
+        )

spider/crawler_offline/shipinshuashua_id.py

@@ -0,0 +1,256 @@
+# -*- 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)
+        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 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"
+)