import re import threading import time from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.support.wait import WebDriverWait from selenium.webdriver.support import expected_conditions as EC class CommentManager: def __init__(self): self.options = Options() self.options.add_argument("--disable-blink-features") self.options.add_argument("--disable-blink-features=AutomationControlled") # driver = webdriver.Chrome(options=options) # 另外一种方式 # 谷歌浏览器位置 chrome_location = r'C:\Program Files (x86)\Google\Chrome\Application\chrome.exe' # 谷歌浏览器驱动地址 chromedriver_path = r'chromedriver.exe' self.options.binary_location = chrome_location # 指定chrome的路径 self.service = Service(chromedriver_path) # 获取正则表达式 comment_template_str = "" # 将中文替换为正则表达式支持的unicode编码 comment_template_str = comment_template_str.encode('unicode_escape').decode("utf-8") self.comment_templates = [(x.split("#")[0], x.split("#")[1]) if x.find("#")>=0 for x in comment_template_str.split("\n")] self.driver = None def __init(self): if not self.driver: self.driver = webdriver.Chrome(service=self.service, options=self.options) def start_process_comment(self): """ 开始处理评论 :return: """ self.__init() self.driver.get( "https://channels.weixin.qq.com/platform/comment?isImageMode=0") # 等到左侧菜单出现后才能执行后续操作 wait = WebDriverWait(self.driver, 100) # 最多等待10秒 element = wait.until(EC.visibility_of_element_located((By.ID, "side-bar"))) threading.Thread(target=lambda: self.click_like(self.driver), daemon=True).start() def __process_like(self,comment_content, comment_element): """ 处理单条评论的赞 :param comment_content: :param comment_element: :return: """ comment_pattern = re.compile(r'[\u4e00-\u9fa5]+') comment_content = re.sub("", "", comment_content) if not comment_pattern.search(comment_content): # 符合标准的评论 return comment_actions = comment_element.find_element(By.CLASS_NAME, "action-list").find_elements(By.CLASS_NAME, "action-item") classes: str = comment_actions[0].find_element(By.TAG_NAME, "svg").get_attribute("class") if classes and classes.find("fill-like") >= 0: # 已经点过赞了 return like = comment_element.find_element(By.CLASS_NAME, "like-action") like.click() self.driver.implicitly_wait(2) def __process_reply(self, comment_content, comment_element): """ 处理单条评论的回复 :param comment_content: :param comment_element: :return: """ def get_reply_comment(nick_name, content): # 内容是否符合标准 for t in self.comment_templates: if re.match(t[0], content): # 满足内容 retrun t[1].replace("[昵称]",nick_name) comment_pattern = re.compile(r'[\u4e00-\u9fa5]+') comment_content = re.sub("", "", comment_content) if not comment_pattern.search(comment_content): # 符合标准的评论 return try: comment_reply_list = comment_element.find_element(By.CLASS_NAME, "comment-reply-list") if comment_reply_list.find_element(By.XPATH, ".//div[normalize-space()='作者']"): print("已经回复。。。。") return except: print("还没回复。。。。") comment_actions = comment_element.find_element(By.CLASS_NAME, "action-list").find_elements(By.CLASS_NAME, "action-item") # 点击评论 comment_actions[1].click() wait = WebDriverWait(self.driver, 5) # 最多等待10秒 element = wait.until(EC.visibility_of_element_located((By.CLASS_NAME, "comment-create-content"))) replay_content = get_reply_comment(, comment_element) if not replay_content: # 不需要评论 return self.driver.find_element(By.CLASS_NAME, "comment-create-content").find_element(By.TAG_NAME, "textarea").send_keys(replay_content) self.driver.implicitly_wait(1) self.driver.find_element(By.CLASS_NAME, "comment-create-content").find_element(By.XPATH, "div[3]/div[2]").click() time.sleep(2) def __click_like_all(self, driver, start_index=0): scroll_list = driver.find_element(By.CLASS_NAME, "feed-comment__wrp") loadmore = scroll_list.find_element(By.CLASS_NAME, "loadmore__dot") # 获取父控件 loadmore = loadmore.find_element(By.XPATH, "./..") has_more = loadmore.value_of_css_property("display") == "none" comments = scroll_list.find_elements(By.XPATH, "div[2]/div/div/div[contains(@class,'comment-item')]") print("评论条数", len(comments)) for index in range(start_index, len(comments)): comment = comments[index] # 评论内容 comment_content = comment.find_element(By.CLASS_NAME, "comment-content").get_attribute("innerHTML") if not comment_content: continue print("=======评论内容：", comment_content) self.__process_reply(comment_content, comment) self.__process_like(comment_content, comment) # 往下滑动 if has_more: driver.execute_script( "var __comment_scroll = document.getElementsByClassName('feed-comment__wrp')[0]; __comment_scroll.scrollTop = __comment_scroll.scrollHeight;") print("往下滚动") time.sleep(3) self.__click_like_all(driver, start_index=len(comments)) else: print("没有更多了") def click_like(self, driver: webdriver.Chrome): wait = WebDriverWait(driver, 100) # 最多等待10秒 element = wait.until(EC.visibility_of_element_located((By.CLASS_NAME, "feeds-container"))) time.sleep(1) # 查找视频列表 videos_root = driver.find_element(By.CLASS_NAME, "feeds-container") videos = videos_root.find_elements(By.CLASS_NAME, "comment-feed-wrap") for video in videos: # 选择视频 video.click() driver.implicitly_wait(2) self.__click_like_all(driver) def close(self): if self.driver: try: self.driver.close() except: pass if __name__ == "__main__": # CommentManager().start_process_comment() s ="你好" print(s.encode('unicode_escape').decode("utf-8")) # if __name__ == "__main__": # data = """ # 213[呲牙][呲牙][呲牙] # """ # # result = re.compile(r'[\u4e00-\u9fa5]+').search(re.sub("", "", data)) # print(result)