# -*- coding: utf-8 -*- # @Author: luojunhui # @Time: 2023/9/27 import json import os import sys import time import random from hashlib import md5 from appium import webdriver from appium.webdriver.extensions.android.nativekey import AndroidKey from appium.webdriver.webdriver import WebDriver # 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 sys.path.append(os.getcwd()) from common.common import Common from common.mq import MQ from common.public import download_rule, get_config_from_mysql from common.scheduling_db import MysqlHelper class XiaoNianGaoPlusRecommend: platform = "小年糕" download_cnt = 0 element_list = [] i = 0 swipe_count = 0 @classmethod def start_wechat(cls, log_type, crawler, env, rule_dict, our_uid): if env == "dev": chromedriverExecutable = "/Users/a123456/Downloads/chromedriver_v111/chromedriver" else: chromedriverExecutable = "/Users/a123456/Downloads/chromedriver_v111/chromedriver" Common.logger(log_type, crawler).info("启动微信") Common.logging(log_type, crawler, env, '启动微信') # 微信的配置文件 caps = { "platformName": "Android", "devicesName": "Android", "platformVersion": "13", # "udid": "emulator-5554", "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"}, } driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) driver.implicitly_wait(30) # action = TouchAction(driver) for i in range(120): try: if driver.find_elements(By.ID, "com.tencent.mm:id/f2s"): Common.logger(log_type, crawler).info("微信启动成功") Common.logging(log_type, crawler, env, '微信启动成功') break elif driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"): Common.logger(log_type, crawler).info("发现并关闭系统下拉菜单") Common.logging(log_type, crawler, env, '发现并关闭系统下拉菜单') driver.find_element(By.ID, "com.android.system:id/dismiss_view").click() else: pass except NoSuchElementException: time.sleep(1) Common.logger(log_type, crawler).info("下滑，展示小程序选择面板") # Common.logging(log_type, crawler, env, '下滑，展示小程序选择面板') size = driver.get_window_size() 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(log_type, crawler).info('打开小程序"小年糕+"') # Common.logging(log_type, crawler, env, '打开小程序"小年糕+"') driver.find_elements(By.XPATH, '//*[@text="小年糕+"]')[-1].click() time.sleep(5) cls.get_videoList(log_type, crawler, driver, env, rule_dict, our_uid) time.sleep(1) driver.quit() @classmethod def search_elements(cls, driver: WebDriver, xpath): time.sleep(1) windowHandles = driver.window_handles for handle in windowHandles: driver.switch_to.window(handle) time.sleep(1) try: elements = driver.find_elements(By.XPATH, xpath) if elements: # print(" driver find elements") return elements # else: # print(" driver not find elements") except NoSuchElementException: pass @classmethod def check_to_applet(cls, log_type, crawler, env, driver: WebDriver, xpath): time.sleep(1) webViews = driver.contexts driver.switch_to.context(webViews[1]) windowHandles = driver.window_handles for handle in windowHandles: driver.switch_to.window(handle) time.sleep(1) try: driver.find_element(By.XPATH, xpath) Common.logger(log_type, crawler).info("切换到WebView成功\n") Common.logging(log_type, crawler, env, '切换到WebView成功\n') return except NoSuchElementException: time.sleep(1) @classmethod def repeat_video(cls, log_type, crawler, video_id, env): sql = f""" select * from crawler_video where platform in ("众妙音信", "刚刚都传", "吉祥幸福", "知青天天看", "zhufuquanzi", "祝福圈子", "haitunzhufu", "海豚祝福", "小年糕") and out_video_id="{video_id}"; """ repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env) return len(repeat_video) @classmethod def swipe_up(cls, driver: WebDriver): cls.search_elements(driver, '//*[@class="list-list--list"]') size = driver.get_window_size() driver.swipe(int(size["width"] * 0.5), int(size["height"] * 0.8), int(size["width"] * 0.5), int(size["height"] * 0.4), 200) cls.swipe_count += 1 @classmethod def get_video_url(cls, log_type, crawler, driver: WebDriver, video_title_element, env): for i in range(3): cls.search_elements(driver, '//*[@class="list-list--list"]') Common.logger(log_type, crawler).info(f"video_title_element:{video_title_element[0]}") time.sleep(1) Common.logger(log_type, crawler).info("滑动标题至可见状态") driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'});", video_title_element[0]) time.sleep(3) Common.logger(log_type, crawler).info("点击标题") video_title_element[0].click() cls.check_to_applet(log_type=log_type, crawler=crawler, driver=driver, env=env, xpath=r'//wx-video[@class="dynamic-index--video-item dynamic-index--video"]') Common.logger(log_type, crawler).info("点击标题完成") time.sleep(10) video_url_elements = cls.search_elements(driver, '//wx-video[@class="dynamic-index--video-item dynamic-index--video"]') if video_url_elements: return video_url_elements[0].get_attribute("src") @classmethod def get_videoList(cls, log_type, crawler, driver: WebDriver, env, rule_dict, our_uid): mq = MQ(topic_name="topic_crawler_etl_" + env) driver.implicitly_wait(20) # 切换到 web_view cls.check_to_applet(log_type=log_type, crawler=crawler, env=env, driver=driver, xpath='//*[@class="tab-bar--tab tab-bar--tab-selected"]') time.sleep(1) page = 0 while True: if cls.search_elements(driver, '//*[@class="list-list--list"]') is None: Common.logger(log_type, crawler).info("窗口已销毁\n") Common.logging(log_type, crawler, env, '窗口已销毁\n') cls.i = 0 cls.download_cnt = 0 cls.element_list = [] return cls.swipe_up(driver) page_source = driver.page_source soup = BeautifulSoup(page_source, 'html.parser') soup.prettify() video_list_elements = soup.findAll("wx-view", class_="expose--adapt-parent") # video_list_elements 有，cls.element_list 中没有的元素 video_list_elements = list(set(video_list_elements).difference(set(cls.element_list))) # video_list_elements 与 cls.element_list 的并集 cls.element_list = list(set(video_list_elements) | set(cls.element_list)) Common.logger(log_type, crawler).info(f"正在抓取第{page + 1}页，共:{len(video_list_elements)}条视频") Common.logging(log_type, crawler, env, f"正在抓取第{page + 1}页，共:{len(video_list_elements)}条视频") if len(video_list_elements) == 0: for i in range(10): Common.logger(log_type, crawler).info(f"向上滑动第{i + 1}次") # scroll_down(action, driver) # print(f"向上滑动第{i + 1}次 s") cls.swipe_up(driver) # print(f"向上滑动第{i + 1}次 e") time.sleep(0.5) continue for i, video_element in enumerate(video_list_elements): try: Common.logger(log_type, crawler).info(f"本轮已抓取{cls.download_cnt}条视频\n") Common.logging(log_type, crawler, env, f"本轮已抓取{cls.download_cnt}条视频\n") if cls.download_cnt >= int(rule_dict.get("videos_cnt", {}).get("min", 10)): cls.i = 0 cls.download_cnt = 0 cls.element_list = [] return cls.i += 1 Common.logger(log_type, crawler).info(f"第{cls.i}条视频") Common.logging(log_type, crawler, env, f"第{cls.i}条视频") # 标题 video_title = video_element.find("wx-view", class_="dynamic--title").text # 播放量字符串 play_str = video_element.find("wx-view", class_="dynamic--views").text info_list = video_element.findAll("wx-view", class_="dynamic--commerce-btn-text") # 点赞数量 like_str = info_list[1].text # 评论数量 comment_str = info_list[2].text # 视频时长 duration_str = video_element.find("wx-view", class_="dynamic--duration").text user_name = video_element.find("wx-view", class_="dynamic--nick-top").text # 头像 URL avatar_url = video_element.find("wx-image", class_="avatar--avatar")["src"] # 封面 URL 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, "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()))), "avatar_url": avatar_url, "cover_url": cover_url, "session": f"xiaoniangao-{int(time.time())}" } for k, v in video_dict.items(): Common.logger(log_type, crawler).info(f"{k}:{v}") Common.logging(log_type, crawler, env, f"video_dict:{video_dict}") # Common.logger(log_type, crawler).info(f"==========分割线==========\n") if video_title is None or cover_url is None: Common.logger(log_type, crawler).info("无效视频\n") Common.logging(log_type, crawler, env, '无效视频\n') cls.swipe_up(driver) time.sleep(0.5) elif download_rule(log_type=log_type, crawler=crawler, video_dict=video_dict, rule_dict=rule_dict) is False: Common.logger(log_type, crawler).info("不满足抓取规则\n") Common.logging(log_type, crawler, env, "不满足抓取规则\n") cls.swipe_up(driver) time.sleep(0.5) elif any(str(word) if str(word) in video_dict["video_title"] else False for word in get_config_from_mysql(log_type=log_type, source=crawler, env=env, text="filter", action="")) is True: Common.logger(log_type, crawler).info('已中过滤词\n') Common.logging(log_type, crawler, env, '已中过滤词\n') cls.swipe_up(driver) time.sleep(0.5) elif cls.repeat_video(log_type, crawler, out_video_id, env) != 0: Common.logger(log_type, crawler).info('视频已下载\n') Common.logging(log_type, crawler, env, '视频已下载\n') cls.swipe_up(driver) time.sleep(5) else: video_title_element = cls.search_elements(driver, f'//*[contains(text(), "{video_title}")]') if video_title_element is None: Common.logger(log_type, crawler).warning(f"未找到该视频标题的element:{video_title_element}") Common.logging(log_type, crawler, env, f"未找到该视频标题的element:{video_title_element}") continue Common.logger(log_type, crawler).info("点击标题，进入视频详情页") Common.logging(log_type, crawler, env, "点击标题，进入视频详情页") video_url = cls.get_video_url(log_type, crawler, driver, video_title_element, env=env) if video_url is None: Common.logger(log_type, crawler).info("未获取到视频播放地址\n") driver.press_keycode(AndroidKey.BACK) time.sleep(5) continue video_dict['video_url'] = video_url Common.logger(log_type, crawler).info(f"video_url:{video_url}") video_dict["platform"] = crawler video_dict["strategy"] = log_type video_dict["out_video_id"] = video_dict["video_id"] video_dict["crawler_rule"] = json.dumps(rule_dict) video_dict["user_id"] = our_uid video_dict["publish_time"] = video_dict["publish_time_str"] mq.send_msg(video_dict) # print(video_dict) cls.download_cnt += 1 driver.press_keycode(AndroidKey.BACK) time.sleep(5) # scroll_down(action, driver) cls.swipe_up(driver) if cls.swipe_count > 200: print("一共滑动超过200次") return except Exception as e: Common.logger(log_type, crawler).error(f"抓取单条视频异常:{e}\n") Common.logging(log_type, crawler, env, f"抓取单条视频异常:{e}\n") Common.logger(log_type, crawler).info("已抓取完一组，休眠 5 秒\n") Common.logging(log_type, crawler, env, "已抓取完一组，休眠 5 秒\n") time.sleep(5) page += 1 def scroll_down(action, driver: WebDriver): """ 刷视频函数，使用该函数可以往下滑动进入下一个视频 """ time.sleep(1) width = driver.get_window_size()['width'] # 获取屏幕宽 height = driver.get_window_size()['height'] # 获取屏幕高 # print(width, height) action.press(x=int(0.5 * width), y=int(0.25 * height)) action.wait(ms=random.randint(200, 400)) action.move_to(x=int(0.5 * width), y=int(0.75 * height)) action.release() action.perform() if __name__ == "__main__": rule_dict1 = {"period": {"min": 365, "max": 365}, "duration": {"min": 30, "max": 1800}, "favorite_cnt": {"min": 999999, "max": 0}, "videos_cnt": {"min": 100, "max": 0}, "share_cnt": {"min": 999999, "max": 0}} XiaoNianGaoPlusRecommend.start_wechat("recommend", "xiaoniangao", "dev", rule_dict1, 6267141)