# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2022/10/26 import difflib import os import sys import time from appium import webdriver from appium.webdriver.extensions.android.nativekey import AndroidKey from appium.webdriver.webdriver import WebDriver from selenium.common import NoSuchElementException from selenium.webdriver.common.by import By sys.path.append(os.getcwd()) from main.common import Common from main.feishu_lib import Feishu from xinshi.xinshi_publish import Publish class XinshiAPP: i = 0 @classmethod def start_wechat(cls, log_type, env): 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, # 'chromedriverExecutable': '/Users/wangkun/Downloads/chromedriver_v86/chromedriver', 'chromedriverExecutable': '/Users/piaoquan/Downloads/chromedriver', } driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) driver.implicitly_wait(10) time.sleep(5) # 搜索 新视-热门 cls.search_video(log_type, driver, 'gO4Sn4', env) # 搜索 新视-推荐 cls.search_video(log_type, driver, 'aOjaIU', env) except Exception as e: Common.logger(log_type).error('start_wechat异常，重启APP:{}\n', e) cls.start_wechat(log_type, env) @classmethod def search_element(cls, log_type, driver: WebDriver, element): try: windowHandles = driver.window_handles # 遍历所有的handles，找到当前页面所在的handle：如果pageSource有包含你想要的元素，就是所要找的handle # 小程序的页面来回切换也需要：遍历所有的handles，切换到元素所在的handle for handle in windowHandles: driver.switch_to.window(handle) time.sleep(3) if len(driver.find_elements(By.XPATH, element)) != 0: return driver.find_element(By.XPATH, element) else: pass except Exception as e: Common.logger(log_type).warning('search_element异常:{}\n', e) @classmethod def search_video(cls, log_type, driver: WebDriver, sheetid, env): try: sheet = Feishu.get_values_batch(log_type, 'shipinhao', sheetid) if len(sheet) == 1: Common.logger(log_type).info('暂无数据\n') return else: for i in range(1, len(sheet)): driver.implicitly_wait(10) try: video_title = sheet[i][2] video_user = sheet[i][3] Common.logger(log_type).info('点击搜索按钮') driver.find_element(By.ID, 'com.tencent.mm:id/j5t').click() Common.logger(log_type).info('输入视频标题:{}', video_title) time.sleep(3) driver.find_element(By.ID, 'com.tencent.mm:id/cd7').clear().send_keys(video_title[:40]) Common.logger(log_type).info('点击搜索') driver.press_keycode(AndroidKey.SEARCH) driver.find_element(By.ID, 'com.tencent.mm:id/lm0').click() Common.logger(log_type).info('切换到webview') webview = driver.contexts driver.switch_to.context(webview[1]) time.sleep(5) video_list = cls.search_element(log_type, driver, '//div[@class="unit"]/*[2]') Common.logger(log_type).info('点击"视频"分类') video_list.click() time.sleep(8) # webview搜索并点击目标视频 while True: cls.i += 1 title_element = cls.search_element( log_type, driver, '//div[@class="double-rich double-rich_vertical"]' '/*['+str(cls.i)+']//div[@class="title ellipsis_2"]') driver.execute_script( "arguments[0].scrollIntoView({block:'center',inline:'center'})", title_element) download_title = title_element.text download_user = cls.search_element( log_type, driver, '//div[@class="double-rich double-rich_vertical"]' '/*[' + str(cls.i) + ']//div[@class="vc-source"]').text.split('\n')[0] Common.logger(log_type).info('title:{}', download_title) Common.logger(log_type).info('user:{}\n', download_user) # 标题相似度>=50%, 用户名相似度>=80% # if video_title[:30] in download_title and video_user in download_user: # if difflib.SequenceMatcher(None, video_title, download_title).quick_ratio() >= 0.5 \ # and difflib.SequenceMatcher(None, video_user, download_user).quick_ratio() >= 0.8: if difflib.SequenceMatcher(None, video_user, download_user).quick_ratio() >= 0.8: Common.logger(log_type).info('点击进入该视频详情') title_element.click() # 获取视频播放数据及播放地址 cls.get_video_info(log_type, driver, sheetid) break elif cls.search_element(log_type, driver, '//span[@class="page-end"]'): Feishu.dimension_range(log_type, 'shipinhao', sheetid, 'ROWS', i + 1, i + 1) Common.logger(log_type).info('未搜索到该视频，删除视频信息\n') break else: pass except Exception as e: Common.logger(log_type).error('获取视频列表数据异常，重启微信:{}\n', e) cls.start_wechat(log_type, env) # 下载/上传 cls.xinshi_download_publish(log_type, sheetid, env) except Exception as e: Common.logger(log_type).error('search_video异常:{}\n', e) @classmethod def get_video_info(cls, log_type, driver: WebDriver, sheetid): try: # webview = driver.contexts Common.logger(log_type).info('切回NATIVE_APP\n') driver.switch_to.context('NATIVE_APP') # 点击播放器，获取视频时长 # 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) # 把视频信息写入飞书feeds文档 values = [[duration, like_cnt, share_cnt, favorite_cnt, comment_cnt]] time.sleep(1) Feishu.update_values(log_type, 'shipinhao', sheetid, 'E2:I2', values) Common.logger(log_type).info('视频信息写入飞书文档成功\n') # 分享给 windows 爬虫机 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() while True: if Feishu.get_values_batch(log_type, 'shipinhao', sheetid)[1][11] is None: Common.logger(log_type).info('等待更新 URL 信息') time.sleep(10) else: Common.logger(log_type).info('URL 信息已更新\n') break except Exception as e: Common.logger(log_type).error('get_video_info异常:{}\n', e) @classmethod def xinshi_download_publish(cls, log_type, sheetid, env): try: download_sheet = Feishu.get_values_batch(log_type, 'shipinhao', sheetid) for i in range(1, len(download_sheet)): download_title = download_sheet[i][2].strip().replace('"', '') \ .replace('“', '').replace('“', '…').replace("\n", "") \ .replace("/", "").replace("\r", "").replace("#", "") \ .replace(".", "。").replace("\\", "").replace("&NBSP", "") \ .replace(":", "").replace("*", "").replace("？", "") \ .replace("?", "").replace('"', "").replace("<", "") \ .replace(">", "").replace("|", "").replace(" ", "") download_duration = download_sheet[i][4] download_like_cnt = download_sheet[i][5] download_share_cnt = download_sheet[i][6] download_favorite_cnt = download_sheet[i][7] download_comment_cnt = download_sheet[i][8] download_username = download_sheet[i][3] download_head_url = download_sheet[i][9] download_cover_url = download_sheet[i][10] download_video_url = download_sheet[i][11] Common.logger(log_type).info("download_title:{}", download_title) Common.logger(log_type).info("download_username:{}", download_username) Common.logger(log_type).info("download_video_url:{}", download_video_url) if download_title is None or download_duration is None or download_video_url is None: Feishu.dimension_range(log_type, 'shipinhao', sheetid, 'ROWS', i + 1, i + 1) Common.logger(log_type).info('空行，删除成功\n') return elif str(download_title) in [x for y in Feishu.get_values_batch(log_type, 'shipinhao', 'c77cf9') for x in y]: Feishu.dimension_range(log_type, 'shipinhao', sheetid, 'ROWS', i + 1, i + 1) Common.logger(log_type).info('视频已下载，删除成功\n') return elif str(download_title) in [x for y in Feishu.get_values_batch(log_type, 'shipinhao', 'WAG7Dq') for x in y]: Feishu.dimension_range(log_type, 'shipinhao', sheetid, 'ROWS', i + 1, i + 1) Common.logger(log_type).info('视频已下载，删除成功\n') return elif str(download_title) in [x for y in Feishu.get_values_batch(log_type, 'shipinhao', '0i4jmV') for x in y]: Feishu.dimension_range(log_type, 'shipinhao', sheetid, 'ROWS', i + 1, i + 1) Common.logger(log_type).info('视频已下载，删除成功\n') return elif str(download_title) in [x for y in Feishu.get_values_batch(log_type, 'shipinhao', 'c77cf9') for x in y]: Feishu.dimension_range(log_type, 'shipinhao', sheetid, 'ROWS', i + 1, i + 1) Common.logger(log_type).info('视频已存在，删除成功\n') return else: # 下载封面 Common.download_method(log_type=log_type, text="cover", d_name=str(download_title), d_url=str(download_cover_url)) # 下载视频 Common.download_method(log_type=log_type, text="video", d_name=str(download_title), d_url=str(download_video_url)) # 保存视频信息至 "./videos/{download_video_title}/info.txt" with open("./videos/" + download_title + "/" + "info.txt", "a", encoding="UTF-8") as f_a: f_a.write('shipinhao' + str(int(time.time())) + "\n" + str(download_title) + "\n" + str(download_duration) + "\n" + str(download_favorite_cnt) + "\n" + str(download_comment_cnt) + "\n" + str(download_like_cnt) + "\n" + str(download_share_cnt) + "\n" + str(1920 * 1080) + "\n" + str(int(time.time())) + "\n" + str(download_username) + "\n" + str(download_head_url) + "\n" + str(download_video_url) + "\n" + str(download_cover_url) + "\n" + "shipinhao") Common.logger(log_type).info("==========视频信息已保存至info.txt==========") Common.logger(log_type).info("开始上传视频:{}".format(download_title)) if env == 'dev' and sheetid == 'gO4Sn4': our_video_id = Publish.upload_and_publish(log_type, env, "xinshi_hot") our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/"+str(our_video_id)+"/info" # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "shipinhao", '0i4jmV', "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)), "新视-热门榜", str(download_title), our_video_link, download_duration, download_like_cnt, download_share_cnt, download_favorite_cnt, download_comment_cnt, download_username, str(download_head_url), str(download_cover_url), str(download_video_url)]] time.sleep(1) Feishu.update_values(log_type, "shipinhao", '0i4jmV', "F2:V2", values) # 删除行或列，可选 ROWS、COLUMNS time.sleep(1) Feishu.dimension_range(log_type, "shipinhao", sheetid, "ROWS", i + 1, i + 1) Common.logger(log_type).info("视频上传完成:{}\n", our_video_link) return elif env == 'dev' and sheetid == 'aOjaIU': our_video_id = Publish.upload_and_publish(log_type, env, "xinshi_recommend") our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/"+str(our_video_id)+"/info" # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "shipinhao", 'WAG7Dq', "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)), "新视-推荐榜", str(download_title), our_video_link, download_duration, download_like_cnt, download_share_cnt, download_favorite_cnt, download_comment_cnt, download_username, str(download_head_url), str(download_cover_url), str(download_video_url)]] time.sleep(1) Feishu.update_values(log_type, "shipinhao", 'WAG7Dq', "F2:V2", values) # 删除行或列，可选 ROWS、COLUMNS time.sleep(1) Feishu.dimension_range(log_type, "shipinhao", sheetid, "ROWS", i + 1, i + 1) Common.logger(log_type).info("视频上传完成:{}\n", our_video_link) return elif env == 'prod' and sheetid == 'gO4Sn4': our_video_id = Publish.upload_and_publish(log_type, env, "xinshi_hot") our_video_link = "https://admin.piaoquantv.com/cms/post-detail/"+str(our_video_id)+"/info" # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "shipinhao", '0i4jmV', "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)), "新视-热门榜", str(download_title), our_video_link, download_duration, download_like_cnt, download_share_cnt, download_favorite_cnt, download_comment_cnt, download_username, str(download_head_url), str(download_cover_url), str(download_video_url)]] time.sleep(1) Feishu.update_values(log_type, "shipinhao", '0i4jmV', "F2:V2", values) # 删除行或列，可选 ROWS、COLUMNS time.sleep(1) Feishu.dimension_range(log_type, "shipinhao", sheetid, "ROWS", i + 1, i + 1) Common.logger(log_type).info("视频上传完成:{}\n", our_video_link) return elif env == 'prod' and sheetid == 'aOjaIU': our_video_id = Publish.upload_and_publish(log_type, env, "xinshi_recommend") our_video_link = "https://admin.piaoquantv.com/cms/post-detail/"+str(our_video_id)+"/info" # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "shipinhao", 'WAG7Dq', "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)), "新视-热门榜", str(download_title), our_video_link, download_duration, download_like_cnt, download_share_cnt, download_favorite_cnt, download_comment_cnt, download_username, str(download_head_url), str(download_cover_url), str(download_video_url)]] time.sleep(1) Feishu.update_values(log_type, "shipinhao", 'WAG7Dq', "F2:V2", values) # 删除行或列，可选 ROWS、COLUMNS time.sleep(1) Feishu.dimension_range(log_type, "shipinhao", sheetid, "ROWS", i + 1, i + 1) Common.logger(log_type).info("视频上传完成:{}\n", our_video_link) return else: our_video_id = Publish.upload_and_publish(log_type, env, "xinshi_hot") our_video_link = "https://admin.piaoquantv.com/cms/post-detail/"+str(our_video_id)+"/info" # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "shipinhao", 'WAG7Dq', "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)), "新视-热门榜", str(download_title), our_video_link, download_duration, download_like_cnt, download_share_cnt, download_favorite_cnt, download_comment_cnt, download_username, str(download_head_url), str(download_cover_url), str(download_video_url)]] time.sleep(1) Feishu.update_values(log_type, "shipinhao", 'WAG7Dq', "F2:V2", values) # 删除行或列，可选 ROWS、COLUMNS time.sleep(1) Feishu.dimension_range(log_type, "shipinhao", sheetid, "ROWS", i + 1, i + 1) Common.logger(log_type).info("视频上传完成:{}\n", our_video_link) return except Exception as e: Feishu.dimension_range(log_type, "shipinhao", "FSDlBy", "ROWS", 2, 2) Common.logger(log_type).error('download_publish异常，删除视频信息成功:{}\n', e) if __name__ == '__main__': XinshiAPP.start_wechat('xinshi', 'dev') pass