import time import random from typing import Optional, Tuple from playwright.sync_api import Page, ElementHandle from loguru import logger from config import Config class MIPAdAutomation: """MIP页面广告自动化操作""" def __init__(self, page: Page): self.page = page def check_and_click_ad(self, url: str) -> Tuple[bool, bool]: """ 检查并点击广告 Args: url: MIP页面链接 Returns: (是否点击成功, 是否获得回复) """ try: # 访问链接 logger.info(f"访问链接: {url}") self.page.goto(url, wait_until='domcontentloaded') # 等待页面加载 time.sleep(3) # 检查是否存在商业广告 has_ad, ad_element = self._detect_commercial_ad() if not has_ad: logger.info("未检测到商业广告，跳过该链接") return False, False # 点击广告 logger.info("检测到商业广告，准备点击") if not self._click_advertisement(ad_element): logger.warning("点击广告失败") return False, False # 等待并检查回复 has_reply = self._wait_for_reply() return True, has_reply except Exception as e: logger.error(f"处理链接异常: {str(e)}") return False, False finally: # 尝试关闭当前标签页，返回主窗口 self._close_current_tab() def _detect_commercial_ad(self) -> Tuple[bool, Optional[ElementHandle]]: """ 检测页面是否存在商业广告 Returns: (是否存在商业广告, 广告元素) """ try: # 等待评论区加载 time.sleep(2) # 方法1: 查找包含"广告"标识的元素 # 根据实际页面结构调整选择器 ad_selectors = [ "//div[contains(@class, 'ad') or contains(@class, 'advertisement')]", "//div[contains(text(), '广告')]", "//*[contains(text(), '广告')]//ancestor::div[contains(@class, 'card')]", "//a[contains(@class, 'ad-link')]", ] for selector in ad_selectors: try: elements = self.page.locator(f"xpath={selector}").all() if elements: # 检查元素是否可见 for elem in elements: if elem.is_visible(): # 进一步验证是否是商业广告（非AI健康管家） elem_text = elem.inner_text().lower() if '广告' in elem_text and 'ai健康' not in elem_text: logger.info("检测到商业广告") return True, elem except Exception: continue logger.info("未检测到商业广告") return False, None except Exception as e: logger.error(f"检测广告异常: {str(e)}") return False, None def _click_advertisement(self, ad_element: ElementHandle) -> bool: """ 点击广告元素 Args: ad_element: 广告元素 Returns: 是否点击成功 """ try: # 获取当前页面 context = self.page.context # 滚动到广告元素可见 ad_element.scroll_into_view_if_needed() time.sleep(1) # 监听新页面打开 with context.expect_page() as new_page_info: # 点击广告 ad_element.click() logger.info("已点击广告") # 等待新页面 new_page = new_page_info.value new_page.wait_for_load_state('domcontentloaded') # 切换到新页面 self.page = new_page logger.info("已切换到广告页面") return True except Exception as e: logger.error(f"点击广告异常: {str(e)}") return False def _wait_for_reply(self) -> bool: """ 等待广告主回复 Returns: 是否收到回复 """ try: logger.info(f"等待广告主回复（最多{Config.REPLY_WAIT_TIMEOUT}秒）") # 检查是否已经自动发送消息 time.sleep(2) # 等待并检查回复 start_time = time.time() timeout = Config.REPLY_WAIT_TIMEOUT # 根据实际页面结构调整回复检测逻辑 # 这里使用轮询方式检查是否有新消息 initial_msg_count = self._count_messages() while time.time() - start_time < timeout: time.sleep(2) current_msg_count = self._count_messages() # 如果消息数量增加，说明收到了回复 if current_msg_count > initial_msg_count: logger.info("收到广告主回复") return True logger.info("未收到广告主回复（超时）") return False except Exception as e: logger.error(f"等待回复异常: {str(e)}") return False def _count_messages(self) -> int: """ 统计当前页面的消息数量 Returns: 消息数量 """ try: # 根据实际页面结构调整选择器 # 这里是示例选择器，需要根据实际情况修改 message_selectors = [ "//div[contains(@class, 'message')]", "//div[contains(@class, 'chat-message')]", "//div[contains(@class, 'msg-item')]", ] for selector in message_selectors: try: messages = self.page.locator(f"xpath={selector}").all() if messages: return len(messages) except: continue return 0 except Exception as e: logger.error(f"统计消息数量异常: {str(e)}") return 0 def _close_current_tab(self): """关闭当前标签页并返回主窗口""" try: pages = self.page.context.pages if len(pages) > 1: self.page.close() self.page = pages[0] logger.info("已关闭广告页面") except Exception as e: logger.error(f"关闭标签页异常: {str(e)}") def random_delay(self, min_seconds: int = 2, max_seconds: int = 5): """随机延迟，模拟人工操作""" delay = random.uniform(min_seconds, max_seconds) time.sleep(delay)