# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2022/9/19 import time import os import sys from appium import webdriver from selenium.common import NoSuchElementException from selenium.webdriver.common.by import By from appium.webdriver.webdriver import WebDriver sys.path.append(os.getcwd()) from main.common import Common from main.feishu_lib import Feishu class Topic: # 搜索词列表 search_words = ['#正能量', '#老外真会玩', '#令人震惊的宠物', '#搞笑', '#娱乐', '#热门', '#我要上热门', '#微信娱乐时刻', '#搞笑视频', '#航拍'] # 单个话题抓取视频数 video_cnt = [] # 下载规则 @staticmethod def download_rule(duration, like_cnt, share_cnt, favorite_cnt, comment_cnt): if int(duration) >= 30: if int(like_cnt) >= 0: if int(share_cnt) >= 0: if int(favorite_cnt) >= 100000: if int(comment_cnt) >= 0: return True else: return False else: return False else: return False else: return False else: return False @classmethod def get_topics(cls, log_type): try: topic_sht = Feishu.get_values_batch(log_type, 'shipinhao', 'TZuDRX') topic_list = [] for x in topic_sht: for y in x: if y is None: pass else: topic_list.append(y) return topic_list except Exception as e: Common.logger(log_type).error('get_topics异常:{}', e) @classmethod def start_topic(cls, log_type, topic): # try: Common.logger(log_type).info('启动微信') caps = { "platformName": "Android", # 手机操作系统 Android / iOS "deviceName": "Android", # 连接的设备名（模拟器或真机），安卓可以随便写 "platforVersion": "11", # 手机对应的系统版本（Android 11） "appPackage": "com.tencent.mm", # 被测APP的包名，乐活圈 Android "appActivity": ".ui.LauncherUI", # 启动的Activity名 "autoGrantPermissions": "true", # 让 appium 自动授权 base 权限， # 如果 noReset 为 True，则该条不生效（该参数为 Android 独有），对应的值为 True 或 False "unicodekeyboard": True, # 使用自带输入法，输入中文时填True "resetkeyboard": True, # 执行完程序恢复原来输入法 "noReset": True, # 不重置APP "recreateChromeDriverSessions": True, # 切换到非 chrome-Driver 会 kill 掉 session，就不需要手动 kill 了 "printPageSourceOnFailure": True, # 找不到元素时，appium log 会完整记录当前页面的 pagesource "newCommandTimeout": 6000, # 初始等待时间 "automationName": "UiAutomator2", # 使用引擎，默认为 Appium， # 其中 Appium、UiAutomator2、Selendroid、Espresso 用于 Android，XCUITest 用于 iOS "showChromedriverLog": True, # "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"}, "chromeOptions": {"androidProcess": "com.tencent.mm:tools"}, 'enableWebviewDetailsCollection': True, 'setWebContentsDebuggingEnabled': True } # global driver driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) driver.implicitly_wait(10) Common.logger(log_type).info('点击发现TAB') driver.find_elements(By.ID, 'com.tencent.mm:id/f2s')[2].click() Common.logger(log_type).info('进入视频号') driver.find_elements(By.ID, 'com.tencent.mm:id/gv6')[1].click() time.sleep(1) Common.logger(log_type).info('进入搜索页\n') driver.find_element(By.ID, 'com.tencent.mm:id/j4j').click() cls.get_topic_feeds(log_type, topic, driver) Common.logger(log_type).info('休眠 3s，退出微信') time.sleep(3) cls.quit(log_type, driver) # except Exception as e: # Common.logger(log_type).error('start_wechat异常\n', e) # 退出 APP @classmethod def quit(cls, log_type, driver: WebDriver): driver.quit() Common.logger(log_type).info('退出 APP 成功\n') # 操作安卓手机，自己滑动首页视频，并获取视频信息 @classmethod def get_feeds(cls, log_type, driver: WebDriver): try: driver.implicitly_wait(10) # for i in range(2): while True: # 视频标题 try: title_id = driver.find_element(By.ID, 'com.tencent.mm:id/ki5') video_title = title_id.get_attribute('name').split('\n')[0].replace('#', '').strip() except NoSuchElementException: video_title = '' driver.swipe(500, 1600, 500, 300, 200) # 点击播放器，获取视频时长 # Common.logger(log_type).info('暂停播放') pause_btn = driver.find_element(By.ID, 'com.tencent.mm:id/eh4') pause_btn.click() start_time = driver.find_element(By.ID, 'com.tencent.mm:id/l59').get_attribute('name') start_time = int(start_time.split(':')[0]) * 60 + int(start_time.split(':')[-1]) try: end_time = driver.find_element(By.ID, 'com.tencent.mm:id/l7i').get_attribute('name') except NoSuchElementException: end_time = driver.find_element(By.ID, 'com.tencent.mm:id/g73').get_attribute('name') end_time = int(end_time.split(':')[0]) * 60 + int(end_time.split(':')[-1]) duration = start_time + end_time # 点赞 like_id = driver.find_element(By.ID, 'com.tencent.mm:id/k04') like_cnt = like_id.get_attribute('name') if like_cnt == "" or like_cnt == "喜欢": like_cnt = 0 elif '万' in like_cnt: like_cnt = float(like_cnt.split('万')[0]) * 10000 elif '万+' in like_cnt: like_cnt = float(like_cnt.split('万+')[0]) * 10000 else: like_cnt = float(like_cnt) # 分享 share_id = driver.find_element(By.ID, 'com.tencent.mm:id/jhv') share_cnt = share_id.get_attribute('name') if share_cnt == "" or share_cnt == "转发": share_cnt = 0 elif '万' in share_cnt: share_cnt = float(share_cnt.split('万')[0]) * 10000 elif '万+' in share_cnt: share_cnt = float(share_cnt.split('万+')[0]) * 10000 else: share_cnt = float(share_cnt) # 收藏 favorite_id = driver.find_element(By.ID, 'com.tencent.mm:id/fnp') favorite_cnt = favorite_id.get_attribute('name') if favorite_cnt == "" or favorite_cnt == "收藏": favorite_cnt = 0 elif '万' in favorite_cnt: favorite_cnt = float(favorite_cnt.split('万')[0]) * 10000 elif '万+' in favorite_cnt: favorite_cnt = float(favorite_cnt.split('万+')[0]) * 10000 else: favorite_cnt = float(favorite_cnt) # 评论 comment_id = driver.find_element(By.ID, 'com.tencent.mm:id/bje') comment_cnt = comment_id.get_attribute('name') if comment_cnt == "" or comment_cnt == "评论": comment_cnt = 0 elif '万' in comment_cnt: comment_cnt = float(comment_cnt.split('万')[0]) * 10000 elif '万+' in comment_cnt: comment_cnt = float(comment_cnt.split('万+')[0]) * 10000 else: comment_cnt = float(comment_cnt) # 用户名 username_id = driver.find_element(By.ID, 'com.tencent.mm:id/hft') user_name = username_id.get_attribute('name') Common.logger(log_type).info('video_title:{}', video_title) Common.logger(log_type).info('user_name:{}', user_name) Common.logger(log_type).info('duration:{}', duration) Common.logger(log_type).info('like_cnt:{}', like_cnt) # Common.logger(log_type).info('share_cnt:{}', share_cnt) # Common.logger(log_type).info('favorite_cnt:{}', favorite_cnt) # Common.logger(log_type).info('comment_cnt:{}', comment_cnt) # 判断无效视频 if video_title == '' or user_name == '': Common.logger(log_type).info('向上滑动页面') driver.swipe(500, 1600, 500, 300, 200) Common.logger(log_type).info('无效视频\n') # 判断下载规则 elif cls.download_rule(duration, like_cnt, share_cnt, favorite_cnt, comment_cnt) is False: Common.logger(log_type).info('不满足抓取规则，滑动到下一个视频\n') driver.swipe(500, 1600, 500, 300, 200) # 已下载表去重 elif str(video_title) in [x for y in Feishu.get_values_batch(log_type, 'shipinhao', 'c77cf9') for x in y]: Common.logger(log_type).info('向上滑动页面') driver.swipe(500, 1600, 500, 300, 200) Common.logger(log_type).info('视频已下载\n') # feeds 表去重 elif str(video_title) in [x for y in Feishu.get_values_batch(log_type, 'shipinhao', 'FSDlBy') for x in y]: Common.logger(log_type).info('向上滑动页面') driver.swipe(500, 1600, 500, 300, 200) Common.logger(log_type).info('视频已存在\n') # 分享给 windows 爬虫机 else: share_id.click() driver.find_element(By.XPATH, '//*[@text="转发给朋友"]').click() driver.find_element(By.XPATH, '//*[@text="爬虫群"]').click() driver.find_element(By.ID, 'com.tencent.mm:id/guw').click() # 把视频信息写入飞书feeds文档 Feishu.insert_columns(log_type, 'shipinhao', 'FSDlBy', 'ROWS', 1, 2) get_feeds_time = int(time.time()) values = [[time.strftime('%Y/%m/%d %H:%M:%S', time.localtime(get_feeds_time)), '话题榜', str(video_title), duration, like_cnt, share_cnt, favorite_cnt, comment_cnt, str(user_name)]] time.sleep(1) Feishu.update_values(log_type, 'shipinhao', 'FSDlBy', 'A2:Z2', values) Common.logger(log_type).info('视频信息写入飞书文档成功\n') while True: if Feishu.get_values_batch(log_type, 'shipinhao', 'FSDlBy')[1][11] is None: Common.logger(log_type).info('等待更新 URL 信息') time.sleep(10) else: Common.logger(log_type).info('URL 信息已更新\n') cls.video_cnt.append(video_title) driver.swipe(500, 1600, 500, 300, 200) break if len(cls.video_cnt) < 5: Common.logger(log_type).info('已抓取{}条视频\n', len(cls.video_cnt)) else: Common.logger(log_type).info('已抓取{}条视频\n', len(cls.video_cnt)) cls.video_cnt = [] return except Exception as e: Common.logger(log_type).error('get_feeds异常\n', e) driver.swipe(500, 1600, 500, 300, 200) @classmethod def get_topic_feeds(cls, log_type, topic, driver: WebDriver): driver.implicitly_wait(15) # webview = driver.contexts Common.logger(log_type).info('切换到webview') driver.switch_to.context('WEBVIEW_com.tencent.mm:appbrand0') time.sleep(5) Common.logger(log_type).info('展开更多搜索历史') for i in range(3): try: search_history = driver.find_element(By.XPATH, '//*[@class="arrow-wrap"]') search_history.click() break except NoSuchElementException: pass Common.logger(log_type).info('查找所有搜索历史') search_words = driver.find_elements(By.XPATH, '//*[@class="history__item__text"]') for search_word in search_words: # print(f'search_word:{search_word.text}') if search_word.text == topic: search_word.click() time.sleep(3) Common.logger(log_type).info('点击搜索结果第一个视频') driver.find_elements(By.XPATH, '//*[@class="waterfall__item"]')[0].click() Common.logger(log_type).info('切回NATIVE_APP\n') driver.switch_to.context('NATIVE_APP') time.sleep(3) cls.get_feeds(log_type, driver) return @classmethod def search_topic(cls, log_type): for topic in cls.search_words: Common.logger(log_type).info('搜索话题:{}\n', topic) cls.start_topic(log_type, topic) time.sleep(3) if __name__ == '__main__': # Topic.get_topics('topic') Topic.search_topic('topic') pass