# -*- coding: utf-8 -*- # @Time: 2023/11/14 import json import os import random import sys import time from datetime import date, timedelta import requests 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 import multiprocessing from common import AliyunLogger from common.feishu import Feishu from common.public import clean_title, get_config_from_mysql sys.path.append(os.getcwd()) from common.common import Common from common.mq import MQ from common.scheduling_db import MysqlHelper def get_redirect_url(url): res = requests.get(url, allow_redirects=False) if res.status_code == 302 or res.status_code == 301: return res.headers['Location'] else: return url class XiaoNianGaoPlusRecommend: env = None driver = None log_type = None def __init__(self, log_type, crawler, env, rule_dict, our_uid): self.mq = None self.platform = "小年糕+主页账号ID" self.download_cnt = 0 self.element_list = [] self.count = 0 self.swipe_count = 0 self.log_type = log_type self.crawler = crawler self.env = env self.rule_dict = rule_dict self.our_uid = our_uid if self.env == "dev": chromedriverExecutable = "/Users/piaoquan/Downloads/chromedriver" else: chromedriverExecutable = "/Users/piaoquan/Downloads/chromedriver" Common.logger(self.log_type, self.crawler).info("启动微信") # 微信的配置文件 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"}, } self.driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) self.driver.implicitly_wait(30) for i in range(120): try: if self.driver.find_elements(By.ID, "com.tencent.mm:id/f2s"): Common.logger(self.log_type, self.crawler).info("微信启动成功") break elif self.driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"): Common.logger(self.log_type, self.crawler).info("发现并关闭系统下拉菜单") 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) Common.logger(self.log_type, self.crawler).info("下滑，展示小程序选择面板") 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) Common.logger(self.log_type, self.crawler).info('打开小程序"小年糕+"') self.driver.find_elements(By.XPATH, '//*[@text="小年糕+"]')[-1].click() 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) Common.logger(self.log_type, self.crawler).info("切换到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() self.driver.swipe(int(size["width"] * 0.5), int(size["height"] * 0.8), int(size["width"] * 0.5), int(size["height"] * 0.442), 200) self.swipe_count += 1 def get_video_url(self, video_title_element): for i in range(3): self.search_elements('//*[@class="list-list--list"]') Common.logger(self.log_type, self.crawler).info(f"video_title_element:{video_title_element[0]}") time.sleep(1) Common.logger(self.log_type, self.crawler).info("滑动标题至可见状态") self.driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'});", video_title_element[0]) time.sleep(3) Common.logger(self.log_type, self.crawler).info("点击标题") video_title_element[0].click() self.check_to_applet(xpath=r'//wx-video[@class="dynamic-index--video-item dynamic-index--video"]') Common.logger(self.log_type, self.crawler).info("点击标题完成") time.sleep(10) video_url_elements = self.search_elements( '//wx-video[@class="dynamic-index--video-item dynamic-index--video"]') if video_url_elements: 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): Common.logger(self.log_type, self.crawler).info(f"本轮已抓取{self.download_cnt}条视频\n") if self.download_cnt >= int(self.rule_dict.get("videos_cnt", {}).get("min", 10)): self.count = 0 self.download_cnt = 0 self.element_list = [] return self.count += 1 Common.logger(self.log_type, self.crawler).info(f"第{self.count}条视频") # 标题 video_title = video_element.find("wx-view", class_="dynamic--title").text # 用户名称 user_name = video_element.find("wx-view", class_="dynamic--nick-top").text video_title_element = self.search_elements(f'//*[contains(text(), "{video_title}")]') if video_title_element is None: Common.logger(self.log_type, self.crawler).warning( f"未找到该视频标题的element:{video_title_element}") return Common.logger(self.log_type, self.crawler).info("点击标题，进入视频详情页") 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")) repeat_video_id = self.repeat_video_id(mid) if repeat_video_id: Common.logger(self.log_type, self.crawler).info(f"该用户已经存在") # status = 0 # self.insert_user(mid, user_name, data_list, status) self.driver.press_keycode(AndroidKey.BACK) return data_list = self.get_user_list(mid) if len(data_list) == 0: Common.logger(self.log_type, self.crawler).info(f"不满足抓取条件") 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(mid, user_name, data_list, status, formatted_time) values = [[ mid, user_name, formatted_time, data_list ]] Feishu.insert_columns('xiaoniangao', 'xiaoniangao', "8zlceR", "ROWS", 1, 2) time.sleep(0.5) Feishu.update_values('xiaoniangao', 'xiaoniangao', "8zlceR", "A2:Z2", values) Common.logger(self.log_type, self.crawler).info(f"{mid}:{user_name}写入成功") AliyunLogger.logging( code="1010", platform=self.platform, mode=self.log_type, env=self.env, message=f"{mid}:{user_name}入库", ) 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.update_values(self.log_type, self.crawler, insert_sql, self.env, action='') 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 repeat_video_id(self,mid): sql = f"SELECT `link` FROM `crawler_user_v3` WHERE `source` = 'xiaoniangao' and `link` = {mid}" repeat_video_id = MysqlHelper.get_values(self.log_type, self.crawler, sql, self.env) return len(repeat_video_id) def get_video_info(self, video_element): try: self.get_video_info_2(video_element) except Exception as e: self.driver.press_keycode(AndroidKey.BACK) Common.logger(self.log_type, self.crawler).error(f"抓取单条视频异常:{e}\n") def get_videoList(self): self.mq = MQ(topic_name="topic_crawler_etl_" + self.env) 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: Common.logger(self.log_type, self.crawler).info("窗口已销毁\n") 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(1) if self.swipe_count > 100: return print("下滑完成") Common.logger(self.log_type, self.crawler).info("已抓取完一组，休眠 5 秒\n") time.sleep(5) def run(): rule_dict1 = {"period": {"min": 365, "max": 365}, "duration": {"min": 30, "max": 1800}, "favorite_cnt": {"min": 0, "max": 0}, "videos_cnt": {"min": 5000, "max": 0}, "share_cnt": {"min": 0, "max": 0}} XiaoNianGaoPlusRecommend("recommend", "xiaoniangao", "prod", rule_dict1, 6267141) if __name__ == "__main__": process = multiprocessing.Process( target=run ) process.start() while True: if not process.is_alive(): print("正在重启") process.terminate() time.sleep(60) os.system("adb forward --remove-all") process = multiprocessing.Process(target=run) process.start() time.sleep(60)