# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2022/10/12 import os import shutil import sys import time import ffmpeg from appium import webdriver from appium.webdriver.common.touch_action import TouchAction from appium.webdriver.webdriver import WebDriver from selenium.common.exceptions import NoSuchElementException from selenium.webdriver.common.by import By sys.path.append(os.getcwd()) from main.common import Common from main.publish import Publish from main.feishu_lib import Feishu class Recommend: i = 0 @classmethod def get_video_info_from_local(cls, video_path): probe = ffmpeg.probe(video_path) # print('video_path: {}'.format(video_path)) # format1 = probe['format'] # bit_rate = int(format1['bit_rate']) / 1000 # duration = format['duration'] # size = int(format1['size']) / 1024 / 1024 video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None) if video_stream is None: print('No video stream found!') return width = int(video_stream['width']) height = int(video_stream['height']) # num_frames = int(video_stream['nb_frames']) # fps = int(video_stream['r_frame_rate'].split('/')[0]) / int(video_stream['r_frame_rate'].split('/')[1]) duration = float(video_stream['duration']) # print('width: {}'.format(width)) # print('height: {}'.format(height)) # print('num_frames: {}'.format(num_frames)) # print('bit_rate: {}k'.format(bit_rate)) # print('fps: {}'.format(fps)) # print('size: {}MB'.format(size)) # print('duration: {}'.format(duration)) return width, height, duration @classmethod def filter_words(cls, log_type): try: filter_words_sheet = Feishu.get_values_batch(log_type, 'zhiqingzongqun', 'i2kcSD') filter_words_list = [] for x in filter_words_sheet: for y in x: if y is None: pass else: filter_words_list.append(y) return filter_words_list except Exception as e: Common.logger(log_type).error('filter_words异常:{}', e) @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 "printPageSourceOnFailure": True, # 找不到元素时，appium log 会完整记录当前页面的 pagesource "newCommandTimeout": 6000, # 初始等待时间 "automationName": "UiAutomator2", # 使用引擎，默认为 Appium， # 其中 Appium、UiAutomator2、Selendroid、Espresso 用于 Android，XCUITest 用于 iOS "showChromedriverLog": True, 'enableWebviewDetailsCollection': True, 'setWebContentsDebuggingEnabled': True, 'recreateChromeDriverSessions': True, # 'chromedriverExecutable': '/Users/wangkun/Downloads/chromedriver', 'chromedriverExecutable': '/Users/piaoquan/Downloads/chromedriver', "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"}, # "chromeOptions": {"androidProcess": "com.tencent.mm:tools"}, 'browserName': '' } driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) driver.implicitly_wait(20) # 向下滑动页面，展示出小程序选择面板 time.sleep(20) Common.logger(log_type).info('下滑，展示小程序选择面板') 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(5) Common.logger(log_type).info('打开小程序"知青天天看"') driver.find_elements(By.XPATH, '//*[@text="知青天天看"]')[-1].click() # 获取视频信息 time.sleep(1) cls.get_recommend(log_type, driver, env) # 退出微信 time.sleep(3) Common.logger(log_type).info('退出微信') 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') # 切换 Handle @classmethod def switch_to_handle(cls, log_type, driver, env, text): try: windowHandles = driver.window_handles Common.logger(log_type).info('windowHandles:{}', windowHandles) # 遍历所有的handles，找到当前页面所在的handle：如果pageSource有包含你想要的元素，就是所要找的handle # 小程序的页面来回切换也需要：遍历所有的handles，切换到元素所在的handle for handle in windowHandles: Common.logger(log_type).info('切换到对应的windowHandle', handle) driver.switch_to.window(handle) time.sleep(3) if driver.page_source.find(text) != -1: break except Exception as e: Common.logger(log_type).warning('切换到小程序handle失败，重启APP:{}\n', e) cls.quit(log_type, driver) cls.start_wechat(log_type, env) @classmethod def get_recommend(cls, log_type, driver: WebDriver, env): try: driver.implicitly_wait(5) # 鼠标左键点击， 1为x坐标， 2为y坐标 time.sleep(10) Common.logger(log_type).info('关闭广告') size = driver.get_window_size() TouchAction(driver).tap(x=int(size['width'] * 0.5), y=int(size['height'] * 0.1)).perform() while True: try: # 切换到 webview webview = driver.contexts # Common.logger(log_type).info('webview:{}', webview) Common.logger(log_type).info('切换到小程序\n') # driver.switch_to.context('WEBVIEW_com.tencent.mm:appbrand0') driver.switch_to.context(webview[1]) time.sleep(5) cls.switch_to_handle(log_type, driver, env, '知青天天看') break except Exception as e: Common.logger(log_type).warning('切换到小程序失败，重启APP:{}\n', e) cls.quit(log_type, driver) cls.i = 0 cls.start_wechat(log_type, env) # Common.logger(log_type).info('点击"换"按钮') # try: # driver.find_element(By.XPATH, '//wx-view[@class="refreshbtnbox"]').click() # except NoSuchElementException: # Common.logger(log_type).warning('未点击到"换"按钮') time.sleep(5) Common.logger(log_type).info('获取推荐列表视频信息') while True: cls.i += 1 recommend_handles = driver.window_handles for recommend_handle in recommend_handles: try: driver.switch_to.window(recommend_handle) # ad try: ad = driver.find_element( By.XPATH, '//*[@class="videolistbox videolist--videolistbox"]' '/*[' + str(cls.i) + ']//*[@class="ad-_banner-_-full"]' ) except NoSuchElementException: ad = 0 # video_title try: title = driver.find_element( By.XPATH, '//*[@class="videolistbox videolist--videolistbox"]' '/*[' + str(cls.i) + ']//*[@class="video_title videolist--video_title"]') # 向上滚动至-元素可见 # Common.logger(log_type).info('滑动视频标题至屏幕中间') driver.execute_script( "arguments[0].scrollIntoView({block:'center',inline:'center'})", title) video_title = title.get_attribute('innerHTML') except NoSuchElementException: title = 0 video_title = 0 # play_cnt try: play_cnt = driver.find_element( By.XPATH, '//*[@class="videolistbox videolist--videolistbox"]' '/*[' + str(cls.i) + ']//*[@class="clickbox videolist--clickbox"]')\ .get_attribute('innerHTML') except NoSuchElementException: play_cnt = 0 # cover_url try: cover_url = driver.find_element( By.XPATH, '//*[@class="videolistbox videolist--videolistbox"]' '/*[' + str(cls.i) + ']//*[@class="itemimage videolist--itemimage"]')\ .get_attribute('src') except Exception: # Common.logger(log_type).error('cover_url异常:{}', e) cover_url = 0 if ad != 0: Common.logger(log_type).info('正在获取第{}条:广告\n', cls.i) break elif video_title == 0: pass else: Common.logger(log_type).info('正在获取第{}条:{}', cls.i, video_title) if video_title == 0 or cover_url == 0: Common.logger(log_type).info('无效视频\n') elif '精美图文' in video_title: Common.logger(log_type).info('精美图文\n') elif any(word if word in video_title else False for word in cls.filter_words(log_type)) is True: Common.logger(log_type).info('视频已中过滤词:{}\n', video_title) # driver.press_keycode(4) elif video_title in [x for y in Feishu.get_values_batch( log_type, 'zhiqingzongqun', 'Z48hlq') for x in y]: Common.logger(log_type).info('视频已存在\n') # driver.press_keycode(4) elif video_title in [x for y in Feishu.get_values_batch( log_type, 'zhiqingzongqun', '1a88b3') for x in y]: Common.logger(log_type).info('视频已下载\n') # driver.press_keycode(4) else: # video_url video_url = cls.get_url(log_type, driver, video_title, title) Common.logger(log_type).info('play_cnt:{}', play_cnt) Common.logger(log_type).info('video_url:{}', video_url) # 下载视频 Common.download_method(log_type, 'video', video_title, video_url) # 获取视频时长 video_info = cls.get_video_info_from_local( "./videos/" + video_title + "/video.mp4") download_width = str(video_info[0]) download_height = str(video_info[1]) download_duration = video_info[2] # 视频时长<60s，直接删除 if int(download_duration) < 60: # 删除视频文件夹 shutil.rmtree("./videos/" + video_title + "/") Common.logger(log_type).info("时长:{}<60秒，删除成功\n", int(download_duration)) return else: # 下载封面 Common.download_method(log_type, 'cover', video_title, cover_url) # 保存视频信息至 "./videos/{download_video_title}/info.txt" with open("./videos/" + video_title + "/" + "info.txt", "a", encoding="UTF-8") as f_a: f_a.write(str(int(time.time())) + "\n" + str(video_title) + "\n" + str(int(download_duration)) + "\n" + str(int(float( play_cnt.split(' ')[-1].split('万')[0]) * 10000)) + "\n" + '0' + "\n" + '0' + "\n" + '0' + "\n" + str(download_width) + '*' + str(download_height) + "\n" + str(int(time.time())) + "\n" + '知青天天看' + "\n" + str(cover_url) + "\n" + str(video_url) + "\n" + str(cover_url) + "\n" + "zhiqingzongqun" + str(int(time.time()))) Common.logger(log_type).info("==========视频信息已保存至info.txt==========") # 上传视频 Common.logger(log_type).info("开始上传视频:{}".format(video_title)) if env == 'dev' and int(download_width) >= int(download_height): our_video_id = Publish.upload_and_publish(log_type, env, "width") our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/" + str( our_video_id) + "/info" elif env == 'dev' and int(download_width) < int(download_height): our_video_id = Publish.upload_and_publish(log_type, env, "height") our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/" + str( our_video_id) + "/info" elif env == 'prod' and int(download_width) >= int(download_height): our_video_id = Publish.upload_and_publish(log_type, env, "width") our_video_link = "https://admin.piaoquantv.com/cms/post-detail/" + str( our_video_id) + "/info" elif env == 'prod' and int(download_width) < int(download_height): our_video_id = Publish.upload_and_publish(log_type, env, "height") our_video_link = "https://admin.piaoquantv.com/cms/post-detail/" + str( our_video_id) + "/info" else: our_video_id = Publish.upload_and_publish(log_type, env, "width") our_video_link = "https://admin.piaoquantv.com/cms/post-detail/" + str( our_video_id) + "/info" Common.logger(log_type).info("视频上传完成:{}", video_title) # 保存视频 ID 到已下载表 Common.logger(log_type).info("保存视频至已下载表:{}", video_title) # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "zhiqingzongqun", "1a88b3", "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(upload_time)), "推荐榜", video_title, our_video_link, play_cnt, int(download_duration), str(download_width) + '*' + str(download_height), cover_url, video_url]] time.sleep(1) Feishu.update_values(log_type, "zhiqingzongqun", "1a88b3", "F2:V2", values) Common.logger(log_type).info("视频:{}，下载/上传成功\n", video_title) driver.press_keycode(4) # Feishu.insert_columns(log_type, 'zhiqingzongqun', 'Z48hlq', '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), # play_cnt, # cover_url, # video_url]] # time.sleep(1) # Feishu.update_values(log_type, 'zhiqingzongqun', 'Z48hlq', 'A2:Z2', values) # Common.logger(log_type).info('视频信息写入飞书文档成功\n') # driver.press_keycode(4) # cls.download_publish(log_type, env) break except Exception: # Common.logger(log_type).error('switch_to.window(recommend_handle)异常:{}', e) pass if cls.i == 2000: cls.i = 0 break except Exception as e: Common.logger(log_type).error('get_recommend异常:{}，重启 APP\n', e) cls.quit(log_type, driver) cls.i = 0 cls.start_wechat(log_type, env) @classmethod def get_url(cls, log_type, driver: WebDriver, video_title, title): try: Common.logger(log_type).info('进入视频详情:{}', video_title) title.click() time.sleep(5) Common.logger(log_type).info('关闭广告') size = driver.get_window_size() TouchAction(driver).tap(x=int(size['width'] * 0.5), y=int(size['height'] * 0.1)).perform() time.sleep(10) info_handles = driver.window_handles for info_handle in info_handles: try: driver.switch_to.window(info_handle) url = driver.find_element( By.XPATH, '//*[@class="wx-swiper-slide-frame"]' '/*[2]//*[@class="video_item videoswiper--video_item"]') video_url = url.get_attribute('src') return video_url except NoSuchElementException: pass except Exception as e: Common.logger(log_type).error('get_url异常:{}\n', e) if __name__ == '__main__': # Recommend.start_wechat('recommend', 'prod') print(Recommend.filter_words('recommend')) pass