Python中如何通过Flask控制Selenium实现自动化任务？

from flask import Flask, request, jsonify
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import threading
import time

app = Flask(__name__)

# 全局变量存储浏览器实例
driver = None
driver_lock = threading.Lock()

def init_browser():
    """初始化浏览器实例"""
    options = webdriver.ChromeOptions()
    options.add_argument('--headless')  # 无头模式
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-dev-shm-usage')
    return webdriver.Chrome(options=options)

@app.route('/start', methods=['POST'])
def start_browser():
    """启动浏览器"""
    global driver
    with driver_lock:
        if driver is None:
            driver = init_browser()
            return jsonify({'status': 'success', 'message': '浏览器已启动'})
        return jsonify({'status': 'error', 'message': '浏览器已在运行中'})

@app.route('/navigate', methods=['POST'])
def navigate():
    """导航到指定URL"""
    global driver
    url = request.json.get('url')
    
    if not url:
        return jsonify({'status': 'error', 'message': 'URL不能为空'})
    
    with driver_lock:
        if driver is None:
            return jsonify({'status': 'error', 'message': '请先启动浏览器'})
        
        try:
            driver.get(url)
            return jsonify({'status': 'success', 'message': f'已导航到 {url}'})
        except Exception as e:
            return jsonify({'status': 'error', 'message': str(e)})

@app.route('/click', methods=['POST'])
def click_element():
    """点击页面元素"""
    global driver
    data = request.json
    by = data.get('by', 'xpath')  # 支持xpath, css_selector, id等
    value = data.get('value')
    
    if not value:
        return jsonify({'status': 'error', 'message': '元素选择器不能为空'})
    
    with driver_lock:
        if driver is None:
            return jsonify({'status': 'error', 'message': '请先启动浏览器'})
        
        try:
            if by == 'xpath':
                element = WebDriverWait(driver, 10).until(
                    EC.element_to_be_clickable((By.XPATH, value))
                )
            elif by == 'css':
                element = WebDriverWait(driver, 10).until(
                    EC.element_to_be_clickable((By.CSS_SELECTOR, value))
                )
            elif by == 'id':
                element = WebDriverWait(driver, 10).until(
                    EC.element_to_be_clickable((By.ID, value))
                )
            else:
                return jsonify({'status': 'error', 'message': '不支持的定位方式'})
            
            element.click()
            return jsonify({'status': 'success', 'message': '点击成功'})
        except Exception as e:
            return jsonify({'status': 'error', 'message': str(e)})

@app.route('/input', methods=['POST'])
def input_text():
    """在输入框中输入文本"""
    global driver
    data = request.json
    by = data.get('by', 'xpath')
    value = data.get('value')
    text = data.get('text')
    
    if not all([value, text]):
        return jsonify({'status': 'error', 'message': '参数不完整'})
    
    with driver_lock:
        if driver is None:
            return jsonify({'status': 'error', 'message': '请先启动浏览器'})
        
        try:
            if by == 'xpath':
                element = WebDriverWait(driver, 10).until(
                    EC.presence_of_element_located((By.XPATH, value))
                )
            elif by == 'css':
                element = WebDriverWait(driver, 10).until(
                    EC.presence_of_element_located((By.CSS_SELECTOR, value))
                )
            else:
                return jsonify({'status': 'error', 'message': '不支持的定位方式'})
            
            element.clear()
            element.send_keys(text)
            return jsonify({'status': 'success', 'message': '输入成功'})
        except Exception as e:
            return jsonify({'status': 'error', 'message': str(e)})

@app.route('/screenshot', methods=['GET'])
def take_screenshot():
    """截取屏幕截图"""
    global driver
    with driver_lock:
        if driver is None:
            return jsonify({'status': 'error', 'message': '请先启动浏览器'})
        
        try:
            filename = f"screenshot_{int(time.time())}.png"
            driver.save_screenshot(filename)
            return jsonify({'status': 'success', 'filename': filename})
        except Exception as e:
            return jsonify({'status': 'error', 'message': str(e)})

@app.route('/stop', methods=['POST'])
def stop_browser():
    """停止浏览器"""
    global driver
    with driver_lock:
        if driver is not None:
            driver.quit()
            driver = None
            return jsonify({'status': 'success', 'message': '浏览器已停止'})
        return jsonify({'status': 'error', 'message': '浏览器未运行'})

@app.route('/execute_script', methods=['POST'])
def execute_script():
    """执行JavaScript代码"""
    global driver
    script = request.json.get('script')
    
    if not script:
        return jsonify({'status': 'error', 'message': '脚本不能为空'})
    
    with driver_lock:
        if driver is None:
            return jsonify({'status': 'error', 'message': '请先启动浏览器'})
        
        try:
            result = driver.execute_script(script)
            return jsonify({'status': 'success', 'result': result})
        except Exception as e:
            return jsonify({'status': 'error', 'message': str(e)})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

这个方案的核心是通过Flask创建REST API来控制Selenium浏览器实例。主要特点：

1. 线程安全：使用锁确保多请求时浏览器操作的安全
2. 完整的CRUD操作：启动、导航、点击、输入、截图、执行脚本、停止
3. 错误处理：每个端点都有完善的异常捕获
4. 等待机制：使用WebDriverWait确保元素加载完成

使用示例：

# 启动浏览器
curl -X POST http://localhost:5000/start -H "Content-Type: application/json"

# 导航到百度
curl -X POST http://localhost:5000/navigate -H "Content-Type: application/json" -d '{"url": "https://www.baidu.com"}'

# 在搜索框输入文本
curl -X POST http://localhost:5000/input -H "Content-Type: application/json" -d '{"by": "xpath", "value": "//input[@id=\\"kw\\"]", "text": "Flask Selenium"}'

记得先安装依赖：pip install flask selenium

总结：用Flask API包装Selenium操作，实现远程控制浏览器自动化。