# -*- 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" )