基于发布-订阅模式实现多浏览器并行独立自动化操作

pip install pika selenium webdriver_manager

# leader.py (概念性代码，需根据实际消息队列库调整)
import json
import pika # 假设使用RabbitMQ

class Leader:
    def __init__(self, queue_name='browser_commands'):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=queue_name)
        self.queue_name = queue_name

    def publish_command(self, browser_id, command_type, payload):
        """
        发布一个指令到消息队列。
        :param browser_id: 目标浏览器实例的ID。
        :param command_type: 指令类型，如 'mouse_move', 'click', 'navigate'。
        :param payload: 指令的具体参数，如坐标、URL等。
        """
        message = {
            "browser_id": browser_id,
            "type": command_type,
            "payload": payload
        }
        self.channel.basic_publish(
            exchange='',
            routing_key=self.queue_name,
            body=json.dumps(message)
        )
        print(f"Leader published: {message}")

    def close(self):
        self.connection.close()

if __name__ == "__main__":
    leader = Leader()
    # 模拟为不同的浏览器实例发布指令
    leader.publish_command(1, 'navigate', {'url': 'https://www.example.com'})
    leader.publish_command(2, 'navigate', {'url': 'https://www.google.com'})
    leader.publish_command(1, 'mouse_move', {'x': 100, 'y': 200})
    leader.publish_command(1, 'click', {'x': 100, 'y': 200})
    leader.publish_command(2, 'mouse_move', {'x': 50, 'y': 50})
    leader.publish_command(2, 'click', {'x': 50, 'y': 50})
    leader.close()

# follower.py (概念性代码，需根据实际消息队列库调整)
import json
import pika
from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
import time

class Follower:
    def __init__(self, browser_id, queue_name='browser_commands'):
        self.browser_id = browser_id
        # 初始化Selenium WebDriver
        options = webdriver.ChromeOptions()
        # 可以添加headless模式，或者为每个浏览器设置不同的用户数据目录以完全隔离
        # options.add_argument('--headless')
        # options.add_argument(f'--user-data-dir=/tmp/chrome-profile-{browser_id}')
        self.driver = webdriver.Chrome(service=ChromeService(ChromeDriverManager().install()), options=options)
        print(f"Follower {self.browser_id}: Browser initialized.")

        # 初始化消息队列连接
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=queue_name)
        self.queue_name = queue_name

    def _execute_command(self, command):
        """根据接收到的指令在浏览器中执行操作。"""
        if command['browser_id'] != self.browser_id:
            return # 忽略不属于本实例的指令

        command_type = command['type']
        payload = command['payload']
        print(f"Follower {self.browser_id}: Executing {command_type} with payload {payload}")

        try:
            if command_type == 'navigate':
                self.driver.get(payload['url'])
            elif command_type == 'mouse_move':
                # Selenium的ActionChains模拟鼠标移动到相对页面左上角的坐标
                # 注意：这只是将Selenium的内部光标移动到该位置，不影响物理光标
                body = self.driver.find_element(By.TAG_NAME, 'body')
                ActionChains(self.driver).move_to_element_with_offset(body, payload['x'], payload['y']).perform()
            elif command_type == 'click':
                # 模拟点击，通常需要先移动到目标位置再点击
                body = self.driver.find_element(By.TAG_NAME, 'body')
                ActionChains(self.driver).move_to_element_with_offset(body, payload['x'], payload['y']).click().perform()
            # 可以添加更多指令类型，如 keyboard_input, find_element_and_click 等
            else:
                print(f"Follower {self.browser_id}: Unknown command type {command_type}")
        except Exception as e:
            print(f"Follower {self.browser_id}: Error executing command {command_type}: {e}")

    def start_listening(self):
        """开始监听消息队列并处理指令。"""
        def callback(ch, method, properties, body):
            command = json.loads(body)
            self._execute_command(command)
            ch.basic_ack(delivery_tag=method.delivery_tag) # 确认消息已处理

        self.channel.basic_consume(
            queue=self.queue_name,
            on_message_callback=callback,
            auto_ack=False # 手动确认，确保消息被成功处理
        )
        print(f"Follower {self.browser_id}: Started listening for commands...")
        self.channel.start_consuming()

    def close(self):
        """关闭浏览器和消息队列连接。"""
        self.driver.quit()
        self.connection.close()
        print(f"Follower {self.browser_id}: Closed.")

if __name__ == "__main__":
    # 启动多个跟随者实例，每个实例在独立的终端或进程中运行
    # 例如，在不同的终端分别运行：
    # python follower.py 1
    # python follower.py 2
    import sys
    if len(sys.argv) > 1:
        browser_id = int(sys.argv[1])
        follower = Follower(browser_id)
        try:
            follower.start_listening()
        except KeyboardInterrupt:
            print(f"Follower {browser_id}: Shutting down...")
        finally:
            follower.close()
    else:
        print("Usage: python follower.py <browser_id>")