使用OpenClaw Browser实现表单自动化的方法

def analyze_form(snapshot):
    """分析表单结构"""
    form_info = {
        "textboxes": [],
        "checkboxes": [],
        "radios": [],
        "selects": [],
        "buttons": []
    }
    
    for element in snapshot["elements"]:
        element_type = element.get("type")
        
        if element_type == "textbox":
            form_info["textboxes"].append({
                "ref": element["ref"],
                "label": element.get("label", ""),
                "placeholder": element.get("placeholder", ""),
                "required": element.get("required", False)
            })
        elif element_type == "checkbox":
            form_info["checkboxes"].append(element)
        elif element_type == "radio":
            form_info["radios"].append(element)
        elif element_type == "combobox":
            form_info["selects"].append(element)
        elif element_type == "button":
            form_info["buttons"].append(element)
    
    return form_info

def map_form_fields(form_info, data):
    """将数据映射到表单字段"""
    mapping = {}
    
    for field in form_info["textboxes"]:
        label = field.get("label", "").lower()
        placeholder = field.get("placeholder", "").lower()
        
        # 智能匹配
        for key, value in data.items():
            key_lower = key.lower()
            if key_lower in label or key_lower in placeholder:
                mapping[field["ref"]] = value
                break
    
    return mapping

def auto_login(url, username, password):
    """自动登录"""
    # 1. 打开登录页面
    browser(action="open", targetUrl=url)
    
    # 2. 等待页面加载
    browser(action="act", kind="wait", timeMs=2000)
    
    # 3. 获取页面快照
    snapshot = browser(
        action="snapshot",
        interactive=True
    )
    
    # 4. 查找用户名输入框
    username_field = find_element_by_keywords(
        snapshot, 
        "textbox", 
        ["用户名", "username", "账号", "email"]
    )
    
    # 5. 输入用户名
    browser(
        action="act",
        kind="type",
        ref=username_field["ref"],
        text=username
    )
    
    # 6. 查找密码输入框
    password_field = find_element_by_keywords(
        snapshot,
        "textbox",
        ["密码", "password", "pwd"]
    )
    
    # 7. 输入密码
    browser(
        action="act",
        kind="type",
        ref=password_field["ref"],
        text=password
    )
    
    # 8. 查找登录按钮
    login_btn = find_element_by_keywords(
        snapshot,
        "button",
        ["登录", "login", "signin", "提交"]
    )
    
    # 9. 点击登录
    browser(
        action="act",
        kind="click",
        ref=login_btn["ref"]
    )
    
    # 10. 等待登录完成
    browser(action="act", kind="wait", timeMs=3000)
    
    # 11. 验证登录结果
    new_snapshot = browser(action="snapshot")
    return check_login_success(new_snapshot)

def find_element_by_keywords(snapshot, element_type, keywords):
    """通过关键词查找元素"""
    for element in snapshot["elements"]:
        if element.get("type") != element_type:
            continue
        
        text = element.get("text", "").lower()
        label = element.get("label", "").lower()
        placeholder = element.get("placeholder", "").lower()
        
        for keyword in keywords:
            if keyword.lower() in text or keyword.lower() in label or keyword.lower() in placeholder:
                return element
    
    return None

def safe_auto_login(url, username, password, max_retries=3):
    """带错误处理的自动登录"""
    for attempt in range(max_retries):
        try:
            result = auto_login(url, username, password)
            return result
        except Exception as e:
            print(f"登录失败 (尝试 {attempt + 1}/{max_retries}): {e}")
            
            if attempt < max_retries - 1:
                # 刷新页面重试
                browser(action="navigate", targetUrl=url)
                browser(action="act", kind="wait", timeMs=2000)
    
    raise Exception("登录失败，已达最大重试次数")

def auto_register(url, user_data):
    """自动注册"""
    # 1. 打开注册页面
    browser(action="open", targetUrl=url)
    browser(action="act", kind="wait", timeMs=2000)
    
    # 2. 获取表单快照
    snapshot = browser(action="snapshot", interactive=True)
    
    # 3. 分析表单结构
    form_info = analyze_form(snapshot)
    
    # 4. 映射数据到字段
    field_mapping = map_form_fields(form_info, user_data)
    
    # 5. 填写文本字段
    for ref, value in field_mapping.items():
        browser(
            action="act",
            kind="type",
            ref=ref,
            text=str(value)
        )
        # 短暂等待，模拟人工输入
        browser(action="act", kind="wait", timeMs=300)
    
    # 6. 处理复选框（如同意条款）
    for checkbox in form_info["checkboxes"]:
        text = checkbox.get("text", "").lower()
        if "同意" in text or "条款" in text or "agree" in text.lower():
            browser(
                action="act",
                kind="click",
                ref=checkbox["ref"]
            )
    
    # 7. 处理下拉选择框
    for select in form_info["selects"]:
        label = select.get("label", "").lower()
        if "城市" in label or "city" in label:
            browser(
                action="act",
                kind="select",
                ref=select["ref"],
                value=user_data.get("city", "北京")
            )
    
    # 8. 提交表单
    submit_btn = find_element_by_keywords(
        snapshot,
        "button",
        ["注册", "register", "提交", "submit"]
    )
    
    if submit_btn:
        browser(
            action="act",
            kind="click",
            ref=submit_btn["ref"]
        )
    
    # 9. 等待响应
    browser(action="act", kind="wait", timeMs=3000)
    
    # 10. 检查结果
    result_snapshot = browser(action="snapshot")
    return check_register_result(result_snapshot)

def handle_captcha():
    """处理验证码"""
    # 获取快照
    snapshot = browser(action="snapshot")
    
    # 查找验证码图片
    captcha_img = find_element(snapshot, "image", "验证码")
    
    if captcha_img:
        # 截取验证码图片
        captcha_screenshot = browser(
            action="screenshot",
            element=captcha_img["ref"]
        )
        
        # 方案1：OCR识别
        captcha_text = ocr_recognize(captcha_screenshot)
        
        # 方案2：请求人工输入
        # captcha_text = request_human_input("请输入验证码")
        
        # 填写验证码
        captcha_input = find_element(snapshot, "textbox", "验证码")
        browser(
            action="act",
            kind="type",
            ref=captcha_input["ref"],
            text=captcha_text
        )

def multi_step_form(url, steps_data):
    """多步骤表单处理"""
    # 1. 打开页面
    browser(action="open", targetUrl=url)
    browser(action="act", kind="wait", timeMs=2000)
    
    # 2. 逐步处理
    for step_index, step_data in enumerate(steps_data):
        print(f"处理步骤 {step_index + 1}/{len(steps_data)}")
        
        # 获取当前页面快照
        snapshot = browser(action="snapshot", interactive=True)
        
        # 填写当前步骤的数据
        fill_current_step(snapshot, step_data)
        
        # 查找下一步按钮
        next_btn = find_element_by_keywords(
            snapshot,
            "button",
            ["下一步", "next", "继续", "continue"]
        )
        
        if next_btn:
            browser(
                action="act",
                kind="click",
                ref=next_btn["ref"]
            )
            browser(action="act", kind="wait", timeMs=2000)
        else:
            # 最后一步，查找提交按钮
            submit_btn = find_element_by_keywords(
                snapshot,
                "button",
                ["提交", "submit", "完成", "finish"]
            )
            
            if submit_btn:
                browser(
                    action="act",
                    kind="click",
                    ref=submit_btn["ref"]
                )
    
    # 3. 等待最终结果
    browser(action="act", kind="wait", timeMs=3000)
    
    # 4. 验证完成
    final_snapshot = browser(action="snapshot")
    return verify_completion(final_snapshot)

def fill_current_step(snapshot, step_data):
    """填写当前步骤的数据"""
    for field_name, value in step_data.items():
        # 查找对应字段
        element = find_element_by_keywords(
            snapshot,
            "textbox",
            [field_name]
        )
        
        if element:
            browser(
                action="act",
                kind="type",
                ref=element["ref"],
                text=str(value)
            )

def verify_step_completion(expected_elements):
    """验证步骤完成"""
    snapshot = browser(action="snapshot")
    
    for element_desc in expected_elements:
        found = find_element_by_keywords(
            snapshot,
            element_desc["type"],
            [element_desc["text"]]
        )
        
        if not found:
            return False
    
    return True

def smart_wait_for_element(element_type, keywords, timeout=10000):
    """智能等待元素出现"""
    start_time = time.time()
    
    while time.time() - start_time < timeout / 1000:
        snapshot = browser(action="snapshot", interactive=True)
        element = find_element_by_keywords(snapshot, element_type, keywords)
        
        if element:
            return element
        
        time.sleep(0.5)
    
    raise TimeoutError(f"等待元素超时: {element_type} - {keywords}")

def preprocess_form_data(data, form_info):
    """预处理表单数据"""
    processed = {}
    
    for field in form_info["textboxes"]:
        label = field.get("label", "").lower()
        placeholder = field.get("placeholder", "").lower()
        
        # 根据字段类型处理数据
        if "邮箱" in label or "email" in placeholder:
            # 邮箱格式验证
            email = data.get("email", "")
            if "@" in email:
                processed[field["ref"]] = email
        
        elif "电话" in label or "phone" in placeholder:
            # 电话格式化
            phone = data.get("phone", "")
            processed[field["ref"]] = phone.replace("-", "").replace(" ", "")
        
        elif "日期" in label or "date" in placeholder:
            # 日期格式化
            date = data.get("date", "")
            processed[field["ref"]] = format_date(date)
        
        else:
            # 默认处理
            for key, value in data.items():
                if key.lower() in label or key.lower() in placeholder:
                    processed[field["ref"]] = str(value)
                    break
    
    return processed

def form_fill_with_recovery(url, data, max_retries=3):
    """带错误恢复的表单填写"""
    saved_state = {}
    
    for attempt in range(max_retries):
        try:
            # 恢复之前的状态
            if saved_state:
                restore_state(saved_state)
            
            # 填写表单
            result = fill_form(url, data)
            return result
            
        except Exception as e:
            print(f"填写失败: {e}")
            
            # 保存当前状态
            saved_state = save_current_state()
            
            if attempt < max_retries - 1:
                # 刷新页面
                browser(action="navigate", targetUrl=url)
                browser(action="act", kind="wait", timeMs=2000)
    
    raise Exception("表单填写失败，已达最大重试次数")

# 批量操作
def batch_fill_fields(field_values):
    """批量填写字段"""
    snapshot = browser(action="snapshot", interactive=True)
    for ref, value in field_values.items():
        browser(
            action="act",
            kind="type",
            ref=ref,
            text=value
        )
        # 减少等待时间
        time.sleep(0.1)  # 最小等待