# -*- coding: utf-8 -*- import json import os import random import subprocess import sys import time import uuid from datetime import datetime import requests 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 from application.common import MysqlHelper, Feishu sys.path.append(os.getcwd()) class XiaoNianGaoZH(object): """ 小年糕+线下爬虫 """ def __init__(self): mid = 1160417293 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) print(tag_id) # 新增账号 pq_uid = self.insert_number(mid, tag_id) self.count = 0 self.swipe_count = 0 chromedriverExecutable = "/Users/tzld/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 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) # 获取当前时间 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()