# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2023/4/25 import datetime import json import os import sys import time from datetime import date, timedelta from hashlib import md5 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 common.public import download_rule from common.mq import MQ from common.common import Common from common.scheduling_db import MysqlHelper class ShipinhaoSearchScheduling: platform = "视频号" i = 0 download_cnt = 0 @classmethod def start_wechat(cls, log_type, crawler, rule_dict, user_dict, env): Common.logger(log_type, crawler).info('启动微信') Common.logging(log_type, crawler, env, '启动微信') if env == "dev": chromedriverExecutable = "/Users/wangkun/Downloads/chromedriver/chromedriver_v107/chromedriver" else: chromedriverExecutable = "/Users/piaoquan/Downloads/chromedriver/chromedriver_v111/chromedriver" caps = { "platformName": "Android", # 手机操作系统 Android / iOS "deviceName": "Android", # 连接的设备名（模拟器或真机），安卓可以随便写 "platforVersion": "13", # 手机对应的系统版本（Android 13） "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"}, # "chromeOptions": {"androidProcess": "com.tencent.mm:toolsmp"}, # "chromeOptions": {"androidProcess": "com.tencent.mm"}, 'enableWebviewDetailsCollection': True, 'setWebContentsDebuggingEnabled': True, 'chromedriverExecutable': chromedriverExecutable, } driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) driver.implicitly_wait(10) # Common.logger(log_type, crawler).info("点击微信") # if len(driver.find_elements(By.ID, 'android:id/text1')) != 0: # driver.find_elements(By.ID, 'android:id/text1')[0].click() # Common.logger(log_type, crawler).info("等待 5s") time.sleep(5) cls.search_video(log_type=log_type, crawler=crawler, rule_dict=rule_dict, user_dict=user_dict, driver=driver, env=env) cls.close_wechat(log_type=log_type, crawler=crawler, env=env, driver=driver) @classmethod def close_wechat(cls, log_type, crawler, env, driver: WebDriver): driver.quit() Common.logger(log_type, crawler).info(f"微信退出成功\n") Common.logging(log_type, crawler, env, f"微信退出成功\n") @classmethod def is_contain_chinese(cls, strword): for ch in strword: if u'\u4e00' <= ch <= u'\u9fff': return True return False # 查找元素 @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: return elements except NoSuchElementException: pass @classmethod def check_to_webview(cls, log_type, crawler, driver: WebDriver): webviews = driver.contexts Common.logger(log_type, crawler).info(f"webviews:{webviews}") driver.switch_to.context(webviews[1]) Common.logger(log_type, crawler).info(driver.current_context) time.sleep(1) windowHandles = driver.window_handles for handle in windowHandles: try: driver.switch_to.window(handle) time.sleep(1) driver.find_element(By.XPATH, '//div[@class="unit"]') Common.logger(log_type, crawler).info('切换 webview 成功') return "成功" except Exception: Common.logger(log_type, crawler).info("切换 webview 失败") @classmethod def repeat_out_video_id(cls, log_type, crawler, out_video_id, env): sql = f""" select * from crawler_video where platform in ("{crawler}","{cls.platform}") and out_video_id="{out_video_id}"; """ repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env) return len(repeat_video) @classmethod def repeat_video_url(cls, log_type, crawler, video_url, env): sql = f""" select * from crawler_video where platform in ("{crawler}","{cls.platform}") and video_url="{video_url}"; """ repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env) return len(repeat_video) @classmethod def search_video(cls, log_type, crawler, rule_dict, driver: WebDriver, user_dict, env): mq = MQ(topic_name="topic_crawler_etl_" + env) # 点击微信搜索框，并输入搜索词 driver.implicitly_wait(10) Common.logger(log_type, crawler).info("点击搜索框") Common.logging(log_type, crawler, env, "点击搜索框") driver.find_element(By.ID, 'com.tencent.mm:id/j5t').click() # 微信8.0.30版本 # driver.find_element(By.ID, 'com.tencent.mm:id/he6').click() # 微信8.0.16版本 time.sleep(0.5) driver.find_element(By.ID, 'com.tencent.mm:id/cd7').clear().send_keys(user_dict["link"]) # 微信8.0.30版本 # driver.find_element(By.ID, 'com.tencent.mm:id/bxz').clear().send_keys(word) # 微信8.0.16版本 driver.press_keycode(AndroidKey.ENTER) Common.logger(log_type, crawler).info("进入搜索词页面") Common.logging(log_type, crawler, env, "进入搜索词页面") driver.find_elements(By.ID, 'com.tencent.mm:id/br8')[0].click() # 微信8.0.30版本 # driver.find_elements(By.ID, 'com.tencent.mm:id/jkg')[0].click() # 微信8.0.16版本 time.sleep(5) # 切换到微信搜索结果页 webview check_to_webview = cls.check_to_webview(log_type, crawler, driver) if check_to_webview is None: Common.logger(log_type, crawler).info("切换到视频号 webview 失败\n") Common.logging(log_type, crawler, env, "切换到视频号 webview 失败\n") return time.sleep(1) # 切换到"视频号"分类 shipinhao_tags = cls.search_elements(driver, '//div[@class="unit"]/*[2]') Common.logger(log_type, crawler).info('点击"视频号"分类') Common.logging(log_type, crawler, env, '点击"视频号"分类') shipinhao_tags[0].click() time.sleep(5) videos_cnt = rule_dict.get('videos_cnt', {}).get('min', 30) index = 0 while True: if cls.search_elements(driver, '//*[@class="mixed-box__bd"]') is None: Common.logger(log_type, crawler).info('窗口已销毁\n') Common.logging(log_type, crawler, env, '窗口已销毁\n') return Common.logger(log_type, crawler).info('获取视频列表\n') Common.logging(log_type, crawler, env, '获取视频列表\n') video_elements = cls.search_elements(driver, '//div[@class="rich-media active__absolute"]') if video_elements is None: Common.logger(log_type, crawler).warning(f'video_elements:{video_elements}') Common.logging(log_type, crawler, env, f'video_elements:{video_elements}') return video_element_temp = video_elements[index:] if len(video_element_temp) == 0: Common.logger(log_type, crawler).info('到底啦~~~~~~~~~~~~~\n') Common.logging(log_type, crawler, env, '到底啦~~~~~~~~~~~~~\n') return for i, video_element in enumerate(video_element_temp): try: Common.logger(log_type, crawler).info(f"download_cnt:{cls.download_cnt}") Common.logging(log_type, crawler, env, f"download_cnt:{cls.download_cnt}") if cls.download_cnt >= int(videos_cnt): Common.logger(log_type, crawler).info(f'搜索词:{user_dict["link"]}，已抓取视频数:{cls.download_cnt}') Common.logging(log_type, crawler, env, f'搜索词:{user_dict["link"]}，已抓取视频数:{cls.download_cnt}') cls.download_cnt = 0 return if video_element is None: Common.logger(log_type, crawler).info('到底啦~\n') Common.logging(log_type, crawler, env, '到底啦~\n') return cls.i += 1 cls.search_elements(driver, '//*[@class="rich-media active__absolute"]') Common.logger(log_type, crawler).info(f'拖动"视频"列表第{cls.i}个至屏幕中间') Common.logging(log_type, crawler, env, f'拖动"视频"列表第{cls.i}个至屏幕中间') time.sleep(3) driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'})", video_element) if len(video_element.find_elements(By.XPATH, "//*[@text='没有更多的搜索结果']")) != 0: Common.logger(log_type, crawler).info("没有更多的搜索结果\n") Common.logging(log_type, crawler, env, "没有更多的搜索结果\n") return video_title = \ video_element.find_elements(By.XPATH, '//div[@class="rich-media__title ellipsis_2"]/span')[ index + i].text[:40] video_url = video_element.find_elements(By.XPATH, '//div[@class="video-player"]')[ index + i].get_attribute('src') cover_url = video_element.find_elements(By.XPATH, '//div[@class="video-player__bd"]')[ index + i].get_attribute('style') cover_url = cover_url.split('url("')[-1].split('")')[0] duration = video_element.find_elements(By.XPATH, '//div[@class="video-player-mask__text"]')[ index + i].text duration = int(duration.split(':')[0]) * 60 + int(duration.split(':')[-1]) user_name = video_element.find_elements(By.XPATH, '//div[@class="rich-media__source__title"]')[ index + i].text avatar_url = video_element.find_elements(By.XPATH, '//div[@class="ui-image-image ui-image rich-media__source__thumb"]')[ index + i].get_attribute('style') avatar_url = avatar_url.split('url("')[-1].split('")')[0] 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, "play_cnt": 0, "duration": duration, # "duration": 60, "user_name": user_name, "user_id": out_user_id, "avatar_url": avatar_url, "cover_url": cover_url, "video_url": video_url, "session": f"shipinhao-search-{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}") if video_title is None or video_url is None: Common.logger(log_type, crawler).info("无效视频\n") Common.logging(log_type, crawler, env, "无效视频\n") elif cls.repeat_out_video_id(log_type, crawler, out_video_id, env) != 0: Common.logger(log_type, crawler).info('视频已下载\n') Common.logging(log_type, crawler, env, '视频已下载\n') elif cls.repeat_video_url(log_type, crawler, video_url, env) != 0: Common.logger(log_type, crawler).info('视频已下载\n') Common.logging(log_type, crawler, env, '视频已下载\n') else: video_element.click() time.sleep(3) video_info_dict = cls.get_video_info(driver) video_dict["like_cnt"] = video_info_dict["like_cnt"] video_dict["share_cnt"] = video_info_dict["share_cnt"] video_dict["favorite_cnt"] = video_info_dict["favorite_cnt"] video_dict["comment_cnt"] = video_info_dict["comment_cnt"] video_dict["publish_time_str"] = video_info_dict["publish_time_str"] + " 00:00:00" video_dict["publish_time_stamp"] = video_info_dict["publish_time_stamp"] Common.logger(log_type, crawler).info(f'publish_time:{video_dict["publish_time_str"]}') Common.logging(log_type, crawler, env, f'publish_time:{video_dict["publish_time_str"]}') if 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") else: video_dict["out_user_id"] = video_dict["user_id"] video_dict["platform"] = crawler video_dict["strategy"] = log_type video_dict["out_video_id"] = video_dict["video_id"] video_dict["width"] = 0 video_dict["height"] = 0 video_dict["crawler_rule"] = json.dumps(rule_dict) video_dict["user_id"] = user_dict["uid"] video_dict["publish_time"] = video_dict["publish_time_str"] mq.send_msg(video_dict) cls.download_cnt += 1 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('已抓取完一组视频，休眠1秒\n') Common.logging(log_type, crawler, env, '已抓取完一组视频，休眠1秒\n') time.sleep(1) index = index + len(video_element_temp) @classmethod def get_video_info(cls, driver: WebDriver): # Common.logger(log_type, crawler).info('切回NATIVE_APP') driver.switch_to.context('NATIVE_APP') # 点赞 like_id = driver.find_element(By.ID, 'com.tencent.mm:id/k04') # 微信版本 8.0.30 like_cnt = like_id.get_attribute('name') if '万' in like_cnt: like_cnt = int(float(like_cnt.split('万')[0]) * 10000) elif '万+' in like_cnt: like_cnt = int(float(like_cnt.split('万+')[0]) * 10000) elif like_cnt == "" or like_cnt == "喜欢" or like_cnt == "火" or cls.is_contain_chinese(like_cnt) is True: like_cnt = 0 else: like_cnt = int(float(like_cnt)) # 分享 share_id = driver.find_element(By.ID, 'com.tencent.mm:id/jhv') share_cnt = share_id.get_attribute('name') if '万' in share_cnt: share_cnt = int(float(share_cnt.split('万')[0]) * 10000) elif '万+' in share_cnt: share_cnt = int(float(share_cnt.split('万+')[0]) * 10000) elif share_cnt == "" or share_cnt == "转发" or cls.is_contain_chinese(share_cnt) is True: share_cnt = 0 else: share_cnt = int(float(share_cnt)) # 收藏 favorite_id = driver.find_element(By.ID, 'com.tencent.mm:id/fnp') favorite_cnt = favorite_id.get_attribute('name') if '万' in favorite_cnt: favorite_cnt = int(float(favorite_cnt.split('万')[0]) * 10000) elif '万+' in favorite_cnt: favorite_cnt = int(float(favorite_cnt.split('万+')[0]) * 10000) elif favorite_cnt == "" or favorite_cnt == "收藏" or favorite_cnt == "推荐" or favorite_cnt == "火" or cls.is_contain_chinese( favorite_cnt) is True: favorite_cnt = 0 else: favorite_cnt = int(float(favorite_cnt)) # 评论 comment_id = driver.find_element(By.ID, 'com.tencent.mm:id/bje') comment_cnt = comment_id.get_attribute('name') if '万' in comment_cnt: comment_cnt = int(float(comment_cnt.split('万')[0]) * 10000) elif '万+' in comment_cnt: comment_cnt = int(float(comment_cnt.split('万+')[0]) * 10000) elif comment_cnt == "" or comment_cnt == "评论" or cls.is_contain_chinese(comment_cnt) is True: comment_cnt = 0 else: comment_cnt = int(float(comment_cnt)) # 发布时间 comment_id.click() time.sleep(1) publish_time = driver.find_element(By.ID, "com.tencent.mm:id/bre").get_attribute("name") if "秒" in publish_time or "分钟" in publish_time or "小时" in publish_time: publish_time_str = (date.today() + timedelta(days=0)).strftime("%Y-%m-%d") elif "天前" in publish_time: days = int(publish_time.replace("天前", "")) publish_time_str = (date.today() + timedelta(days=-days)).strftime("%Y-%m-%d") elif "年" in publish_time: # publish_time_str = publish_time.replace("年", "-").replace("月", "-").replace("日", "") year_str = publish_time.split("年")[0] month_str = publish_time.split("年")[-1].split("月")[0] day_str = publish_time.split("月")[-1].split("日")[0] if int(month_str) < 10: month_str = f"0{month_str}" if int(day_str) < 10: day_str = f"0{day_str}" publish_time_str = f"{year_str}-{month_str}-{day_str}" else: year_str = str(datetime.datetime.now().year) month_str = publish_time.split("月")[0] day_str = publish_time.split("月")[-1].split("日")[0] if int(month_str) < 10: month_str = f"0{month_str}" if int(day_str) < 10: day_str = f"0{day_str}" publish_time_str = f"{year_str}-{month_str}-{day_str}" # publish_time_str = f'2023-{publish_time.replace("月", "-").replace("日", "")}' publish_time_stamp = int(time.mktime(time.strptime(publish_time_str, "%Y-%m-%d"))) # 收起评论 # Common.logger(log_type, crawler).info("收起评论") driver.find_element(By.ID, "com.tencent.mm:id/be_").click() time.sleep(0.5) # 返回 webview # Common.logger(log_type, crawler).info(f"操作手机返回按键") driver.find_element(By.ID, "com.tencent.mm:id/a2z").click() time.sleep(0.5) # driver.press_keycode(AndroidKey.BACK) # cls.check_to_webview(log_type=log_type, crawler=crawler, driver=driver) webviews = driver.contexts driver.switch_to.context(webviews[1]) video_dict = { "like_cnt": like_cnt, "share_cnt": share_cnt, "favorite_cnt": favorite_cnt, "comment_cnt": comment_cnt, "publish_time_str": publish_time_str, "publish_time_stamp": publish_time_stamp, } return video_dict @classmethod def get_search_videos(cls, log_type, crawler, rule_dict, user_list, env): Common.logger(log_type, crawler).info(f"搜索词总数:{len(user_list)}\n") Common.logging(log_type, crawler, env, f"搜索词总数:{len(user_list)}\n") if len(user_list) == 0: return for user_dict in user_list: try: cls.i = 0 cls.download_cnt = 0 Common.logger(log_type, crawler).info(f"开始抓取 {user_dict['link']}\n") Common.logging(log_type, crawler, env, f"开始抓取 {user_dict['link']}\n") cls.start_wechat(log_type=log_type, crawler=crawler, rule_dict=rule_dict, user_dict=user_dict, env=env) except Exception as e: Common.logger(log_type, crawler).error(f"抓取 {user_dict['link']} 时异常:{e}\n") Common.logging(log_type, crawler, env, f"抓取 {user_dict['link']} 时异常:{e}\n") if __name__ == '__main__': print(ShipinhaoSearchScheduling.repeat_out_video_id(log_type="search", crawler="shipinhao", out_video_id="123", env="dev")) pass