# -*- coding: utf-8 -*- # @Time: 2023/12/20 import json import multiprocessing import os import random import re import sys import time import uuid from urllib.parse import urlparse from hashlib import md5 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.functions import get_redirect_url from application.common.log import AliyunLogger, Local sys.path.append(os.getcwd()) from application.pipeline import PiaoQuanPipeline class ZhuFuQuanZiRecommend(object): def __init__(self, log_type, crawler, env, rule_dict, our_uid): self.mq = None self.platform = "zhufuquanzi" 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 chromedriverExecutable = "/usr/bin/chromedriver" self.aliyun_log = AliyunLogger(platform=crawler, mode=log_type, env=env) Local.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"}, } try: self.driver = webdriver.Remote("http://localhost:4750/wd/hub", caps) except Exception as e: self.aliyun_log.logging( code="3002", message=f'appium 启动异常: {e}' ) return self.driver.implicitly_wait(30) for i in range(5): 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: self.aliyun_log.logging( code="3002", message=f'打开微信异常: {e}' ) time.sleep(1) Local.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) self.driver.find_elements(By.XPATH, '//*[@text="祝福圈子"]')[-1].click() self.aliyun_log.logging( code="1000", message="打开小程序祝福圈子成功" ) time.sleep(5) self.get_videoList() time.sleep(1) self.driver.quit() def get_videoList(self): self.driver.implicitly_wait(20) # 切换到 web_view self.check_to_applet(xpath='//*[@class="tags--tag tags--tag-0 tags--checked"]') print("切换到 webview 成功") time.sleep(1) name = ["推荐", "搞笑", "大雪", "亲子"] selected_text = random.choice(name) try: self.driver.find_element(By.XPATH, f"//wx-button[contains(., '{selected_text}')]").click() time.sleep(2) except NoSuchElementException: pass if self.search_elements('//*[@class="bless--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(1) if self.swipe_count > 100: return print("已抓取完一组，休眠 5 秒\n") time.sleep(5) 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(self, video_element): try: self.get_video_info_2(video_element) except Exception as e: self.driver.press_keycode(AndroidKey.BACK) Local.logger(self.log_type, self.crawler).error(f"抓取单条视频异常:{e}\n") self.aliyun_log.logging( code="3001", message=f"抓取单条视频异常:{e}\n" ) def swipe_up(self): self.search_elements('//*[@class="bless--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 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) Local.logger(self.log_type, self.crawler).info("切换到WebView成功\n") self.aliyun_log.logging( code="1000", message="成功切换到 webview" ) return except NoSuchElementException: time.sleep(1) def get_video_url(self, video_title_element): for i in range(3): self.search_elements('//*[@class="bless--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() time.sleep(5) video_url_elements = self.search_elements( '//*[@class="index--video-item index--video"]' ) return video_url_elements[0].get_attribute("src") def get_video_info_2(self, video_element): Local.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 Local.logger(self.log_type, self.crawler).info(f"第{self.count}条视频") # 获取 trace_id, 并且把该 id 当做视频生命周期唯一索引 trace_id = self.crawler + str(uuid.uuid1()) self.aliyun_log.logging( code="1001", trace_id=trace_id, message="扫描到一条视频", ) video_title = video_element.find("wx-view", class_="dynamic--title").text play_str = video_element.find("wx-view", class_="dynamic--views").text like_str = video_element.findAll("wx-view", class_="dynamic--commerce-btn-text")[0].text comment_str = video_element.findAll("wx-view", class_="dynamic--commerce-btn-text")[1].text duration_str = video_element.find("wx-view", class_="dynamic--duration").text user_name = video_element.find("wx-view", class_="dynamic--nick-top").text avatar_url = video_element.find("wx-image", class_="avatar--avatar")["src"] cover_url = video_element.find("wx-image", class_="dynamic--bg-image")["src"] play_cnt = int(play_str.replace("+", "").replace("次播放", "")) duration = int(duration_str.split(":")[0].strip()) * 60 + int(duration_str.split(":")[-1].strip()) if "点赞" in like_str: like_cnt = 0 elif "万" in like_str: like_cnt = int(like_str.split("万")[0]) * 10000 else: like_cnt = int(like_str) if "评论" in comment_str: comment_cnt = 0 elif "万" in comment_str: comment_cnt = int(comment_str.split("万")[0]) * 10000 else: comment_cnt = int(comment_str) out_video_id = md5(video_title.encode('utf8')).hexdigest() out_user_id = md5(user_name.encode('utf8')).hexdigest() video_dict = { "video_title": video_title, "video_id": out_video_id, 'out_video_id': out_video_id, "duration_str": duration_str, "duration": duration, "play_str": play_str, "play_cnt": play_cnt, "like_str": like_str, "like_cnt": like_cnt, "comment_cnt": comment_cnt, "share_cnt": 0, "user_name": user_name, "user_id": out_user_id, 'publish_time_stamp': int(time.time()), 'publish_time_str': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()))), 'update_time_stamp': int(time.time()), "avatar_url": avatar_url, "cover_url": cover_url, "session": f"zhufuquanzi-{int(time.time())}" } pipeline = PiaoQuanPipeline( platform=self.platform, mode=self.log_type, rule_dict=self.rule_dict, env=self.env, item=video_dict, trace_id=trace_id, ) flag = pipeline.process_item() if flag: video_title_element = self.search_elements(f'//*[contains(text(), "{video_title}")]') if video_title_element is None: return Local.logger(self.log_type, self.crawler).info("点击标题，进入视频详情页") self.aliyun_log.logging( code="1000", message="点击标题，进入视频详情页", ) video_url = self.get_video_url(video_title_element) video_url = get_redirect_url(video_url) if video_url is None: self.driver.press_keycode(AndroidKey.BACK) time.sleep(5) return video_dict['video_url'] = video_url video_dict["platform"] = self.crawler video_dict["strategy"] = self.log_type video_dict["out_video_id"] = video_dict["video_id"] video_dict["crawler_rule"] = json.dumps(self.rule_dict) video_dict["user_id"] = self.our_uid video_dict["publish_time"] = video_dict["publish_time_str"] self.mq.send_msg(video_dict) self.download_cnt += 1 self.driver.press_keycode(AndroidKey.BACK) 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}} ZhuFuQuanZiRecommend("recommend", "zhufuquanzi", "prod", rule_dict1, [64120158, 64120157, 63676778]) 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)