Browser Agent：Web 操作的自动化

# Playwright 自动等待示例
async def test_login(page):
    await page.goto("https://example.com/login")
    # Playwright 自动等待 #username 元素可见并可交互
    await page.fill("#username", "testuser")
    await page.fill("#password", "password123")
    await page.click("button[type='submit']")
    # 自动等待导航完成
    await page.wait_for_url("**/dashboard")

// Puppeteer 核心架构示例
const puppeteer = require('puppeteer');

(async () => {
    // 启动浏览器，暴露调试端口
    const browser = await puppeteer.launch({
        headless: true,
        args: ['--remote-debugging-port=9222']
    });
    
    // 通过 CDP 连接浏览器
    const browserWSEndpoint = browser.wsEndpoint();
    
    // 创建页面
    const page = await browser.newPage();
    
    // 拦截网络请求
    await page.setRequestInterception(true);
    page.on('request', request => {
        if (request.url().includes('analytics')) {
            request.abort(); // 屏蔽分析请求
        } else {
            request.continue();
        }
    });
    
    await page.goto('https://example.com');
    await browser.close();
})();

/* 后代选择器：匹配 form 内的所有 input */
form input

/* 子选择器：匹配 form 的直接子元素 input */
form > input

/* 相邻兄弟选择器：匹配 label 后的 input */
label + input

/* 通用兄弟选择器：匹配 label 后的所有 input */
label ~ input

/* 复合选择器：匹配 type 为 text 的 username 输入框 */
input[type="text"][id="username"]

/* 否定选择器：匹配非禁用状态的按钮 */
button:not([disabled])

# Playwright 扩展选择器示例
await page.locator("::-p-ashadow").click()          # Shadow DOM 选择器
await page.locator("::-p-text(/close/i)").click()   # 文本内容选择器
await page.locator("::-p-before(content)").click()  # ::before 伪元素

<!-- 定位第一个 td -->
td[1]

<!-- 定位包含特定文本的 a 标签 -->
a[contains(text(), '查看详情')]

<!-- 定位具有多个属性的 input -->
input[@type='text' and @name='email']

<!-- 定位父元素下的第二个子元素 -->
div[2]/input

<!-- 定位倒数第一个 tr -->
tr[last()]

<!-- 定位非隐藏的 tr -->
tr[not(@hidden)]

# 文本匹配：定位包含"登录"文本的按钮
xpath = "//button[contains(text(), '登录')]"

# 模糊匹配：定位 title 包含"邮箱"的 div
xpath = "//div[contains(@title, '邮箱')]"

# 逻辑或：定位 type 为 submit 或 button 的元素
xpath = "//input[@type='submit' or @type='button']"

# 层级穿越：跳过中间节点定位深层元素
xpath = "//form//table//tr[3]//input[@name='data']"

# 组合定位：先按文本定位父元素，再找兄弟元素
xpath = "//span[text()='用户名']/following-sibling::input"

# Playwright 语义定位示例
# 定位角色为 button 的元素
await page.locator("role=button").click()

# 定位角色为 dialog 的模态框
modal = page.locator("role=dialog")

# 定位 name 为 "Search" 的搜索框
search_input = page.locator("role=searchbox[name='Search']")

# 定位带有 checked 状态的 checkbox
checked_box = page.locator("role=checkbox[checked]")

# 定位按钮文本为"提交"的元素
await page.get_by_role("button", name="提交").click()

# 定位 placeholder 为"输入邮箱"的输入框
await page.get_by_placeholder("输入邮箱").fill("user@example.com")

# 定位 alt 文本为"logo"的图片
await page.get_by_alt_text("logo").click()

# 定位 title 属性为"设置"的元素
await page.get_by_title("设置").click()

# 定位 testid 属性为 login-button 的元素
await page.get_by_test_id("login-button").click()

# 最佳实践：优先使用语义定位
# 理由：对页面结构变化鲁棒，代码可读性高
await page.get_by_role("button", name="登录").click()
await page.get_by_label("用户名").fill("admin")

# 测试 ID 通常由开发团队维护，稳定性较高
await page.get_by_test_id("checkout-button").click()

# ID 定位性能最优，但需确保 ID 唯一性
await page.locator("#main-form").submit()
# 类选择器适合定位同类型元素组
await page.locator(".menu-item").first.click()

# XPath 用于复杂层级关系或文本匹配场景
await page.locator("xpath=//table//tr[last()]//button").click()

class SmartLocator:
    """智能定位器：根据页面特征自动选择最优定位策略"""
    
    def __init__(self, page):
        self.page = page
    
    def find_login_form(self):
        """智能定位登录表单"""
        # 策略1：ARIA 角色定位
        if locator := self.page.get_by_role("form", name=re.compile("登录|login", re.I)):
            return locator
        
        # 策略2：ID 定位
        if locator := self.page.locator("#login-form, #LoginForm, #frmLogin"):
            return locator
        
        # 策略3：XPath 文本匹配
        return self.page.locator("xpath=//form[.//button[contains(text(), '登录')]]")
    
    def find_submit_button(self, parent_locator):
        """在父元素内智能定位提交按钮"""
        return parent_locator.get_by_role("button", name=re.compile("提交|确定|submit", re.I))

# 策略1：等待 load 事件（保守但可靠）
await page.goto("https://example.com", wait_until="load")

# 策略2：等待 DOMContentLoaded（更快）
await page.goto("https://example.com", wait_until="domcontentloaded")

# 策略3：等待网络空闲（适合动态内容）
await page.goto("https://example.com", wait_until="networkidle")

# 策略4：自定义等待条件
await page.goto("https://example.com")
await page.wait_for_function("document.querySelector('#app')._isReady")

async def wait_for_app_ready(page: Page):
    """等待应用完全就绪"""
    # 等待基本 DOM 加载
    await page.wait_for_load_state("domcontentloaded")
    
    # 等待特定元素出现
    await page.wait_for_selector("#app", state="attached")
    
    # 等待应用初始化完成
    await page.wait_for_function(
        """() => {
            const app = document.querySelector('#app');
            return app && app.dataset.ready === 'true';
        }""",
        timeout=30000
    )
    
    # 额外等待动画完成
    await page.wait_for_timeout(500)

# Playwright 网络拦截示例
async def wait_for_api_response(page: Page, url_pattern: str):
    """等待特定 API 返回成功响应"""
    response = await page.wait_for_response(
        lambda r: re.match(url_pattern, r.url) and r.status in [200, 201],
        timeout=10000
    )
    return response

# 使用示例：等待搜索 API 返回
async with page.expect_response(r"**/api/search\?q=.*") as response_info:
    await page.click("#search-btn")
response = await response_info.value
data = await response.json()

# 方法1：等待函数执行结果变化
async def wait_for_data_update(page: Page):
    await page.wait_for_function(
        """() => {
            const counter = document.querySelector('#result-count');
            return counter && parseInt(counter.textContent) > 0;
        }"""
    )

# 方法2：监听自定义事件
async def wait_for_custom_event(page: Page):
    async with page.expect_event("dataLoaded"):
        await page.evaluate("""() => {
            document.addEventListener('dataReady', () => {
                window.dispatchEvent(new CustomEvent('dataLoaded'));
            });
        }""")

# 方法3：轮询检查数据模型
async def wait_for_condition(page: Page, condition_fn, timeout=30000):
    start = time.time()
    while time.time() - start < timeout:
        result = await page.evaluate(condition_fn)
        if result:
            return True
        await page.wait_for_timeout(100)
    raise TimeoutError(f"Condition not met within {timeout}ms")

async def robust_api_call(page: Page, url: str, max_retries: int = 3):
    """带重试机制的 API 调用"""
    for attempt in range(max_retries):
        try:
            response = await page.request.get(url)
            if response.ok:
                return response
            # 非成功状态码，重试
            logging.warning(f"Attempt {attempt + 1} failed: {response.status}")
        except APIError as e:
            logging.warning(f"Attempt {attempt + 1} exception: {e}")
        
        if attempt < max_retries - 1:
            await page.wait_for_timeout(1000 * (attempt + 1))  # 指数退避
    
    raise APIError(f"Failed after {max_retries} attempts")

<!-- Shadow DOM 示例 -->
<my-component>
    #shadow-root (open)
        <div class="inner">Shadow Content</div>
</my-component>

# 方法1：使用 Playwright 的 Shadow DOM 选择器
await page.locator("my-component >> ::p-ashadow >> .inner").click()

# 方法2：链式 shadow() 方法
await page.locator("my-component").locator(
    page.locator(".inner", has=page.locator("..", has=page.locator(":scope"))
).locator(":scope", has_text="Shadow Content")
).click()

# 方法3：JavaScript 递归穿透
async def shadow_click(page: Page, selector: str):
    """通过 JS 穿透 Shadow DOM 点击元素"""
    await page.evaluate("""(sel) => {
        const findInShadow = (root, selector) => {
            const result = root.querySelector(selector);
            if (result) return result;
            
            const shadows = root.querySelectorAll('*');
            for (const el of shadows) {
                if (el.shadowRoot) {
                    const found = findInShadow(el.shadowRoot, selector);
                    if (found) return found;
                }
            }
            return null;
        };
        
        const el = findInShadow(document, sel);
        if (el) el.click();
    }""", selector)

async def deep_shadow_query(page: Page, path: list):
    """
    穿透多层嵌套的 Shadow DOM
    path: Shadow DOM 标签路径，如 ['my-app', 'user-panel', 'settings-form']
    """
    js_code = """([...path]) => {
        const getShadowHost = (element) => {
            return element.parentNode;
        };
        
        let current = document;
        for (const tag of path) {
            let found = false;
            const walkShadow = (root) => {
                const el = root.querySelector(tag);
                if (el) {
                    current = el;
                    found = true;
                    return true;
                }
                const all = root.querySelectorAll('*');
                for (const e of all) {
                    if (e.shadowRoot && walkShadow(e.shadowRoot)) {
                        found = true;
                        return true;
                    }
                }
                return false;
            };
            
            if (found) {
                if (current.shadowRoot) {
                    current = current.shadowRoot;
                }
            } else {
                return null;
            }
        }
        return current;
    }"""
    
    return await page.evaluate(js_code, path)

from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Callable, Optional
import asyncio

class PageState(Enum):
    IDLE = auto()
    LOADING = auto()
    DOM_READY = auto()
    READY = auto()
    PROCESSING = auto()
    ERROR = auto()

@dataclass
class StateTransition:
    from_state: PageState
    to_state: PageState
    trigger: Callable
    condition: Optional[Callable] = None
    timeout: float = 30.0

class PageStateMachine:
    """页面状态机管理器"""
    
    def __init__(self, page: Page):
        self.page = page
        self.state = PageState.IDLE
        self.state_history: list[tuple[float, PageState]] = []
        self.listeners: dict[PageState, list[Callable]] = {}
        
        # 绑定页面事件监听
        self._setup_listeners()
    
    def _setup_listeners(self):
        """设置页面事件监听"""
        self.page.on("domcontentloaded", lambda: self._on_dom_content_loaded())
        self.page.on("load", lambda: self._on_load())
        self.page.on("requestfailed", lambda: self._on_request_failed())
    
    async def _on_dom_content_loaded(self):
        await self._transition(PageState.DOM_READY)
    
    async def _on_load(self):
        await self._transition(PageState.READY)
    
    async def _on_request_failed(self):
        if self.state != PageState.ERROR:
            await self._transition(PageState.ERROR)
    
    async def _transition(self, new_state: PageState):
        old_state = self.state
        self.state = new_state
        self.state_history.append((asyncio.get_event_loop().time(), new_state))
        
        # 触发状态变更监听器
        if new_state in self.listeners:
            for callback in self.listeners[new_state]:
                await callback(old_state, new_state)
    
    def on_state_change(self, state: PageState, callback: Callable):
        """注册状态变更监听器"""
        if state not in self.listeners:
            self.listeners[state] = []
        self.listeners[state].append(callback)
    
    async def wait_for_state(self, target_state: PageState, timeout: float = 30.0):
        """等待达到目标状态"""
        if self.state == target_state:
            return
        
        event = asyncio.Event()
        
        def listener(old, new):
            if new == target_state:
                event.set()
        
        self.on_state_change(target_state, listener)
        
        try:
            await asyncio.wait_for(event.wait(), timeout)
        except asyncio.TimeoutError:
            raise TimeoutError(f"Timeout waiting for state {target_state}")
    
    async def execute_with_state_guard(self, action: Callable):
        """在正确状态下执行操作"""
        await self.wait_for_state(PageState.READY)
        await self._transition(PageState.PROCESSING)
        
        try:
            result = await action()
            await self._transition(PageState.READY)
            return result
        except Exception as e:
            await self._transition(PageState.ERROR)
            raise

# 基础 fill 方法 - 直接设置值（触发输入事件）
await page.fill("#username", "admin")

# type 方法 - 模拟逐字符输入（触发 keydown/keyup/input 事件）
await page.locator("#username").press_sequentially("admin", delay=50)

# 清除后输入
await page.locator("#username").clear()
await page.locator("#username").type("admin")

# 日期输入
await page.fill('input[type="date"]', "2024-06-15")

# 邮箱输入（带验证）
await page.fill('input[type="email"]', "user@example.com")

# 数字输入（带步进）
number_input = page.locator('input[type="number"]')
await number_input.fill("10")
# 或使用步进按钮
await number_input.click()
await page.keyboard.press("ArrowUp")  # 加1
await page.keyboard.press("ArrowDown")  # 减1

# 特殊字符输入
await page.locator("#bio").fill("John <script>alert('xss')</script>")
# 如果需要转义
await page.locator("#bio").fill("第一行\n第二行")

# 方法1：ContentEditable div
await page.evaluate("""() => {
    const editor = document.querySelector('#editor');
    editor.innerHTML = '<p>Hello <strong>World</strong></p>';
    editor.dispatchEvent(new Event('input', { bubbles: true }));
}""")

# 方法2：使用 Playwright 的 contenteditable 支持
await page.locator("#editor").fill("<p>Hello World</p>")

# 方法3：模拟粘贴
await page.locator("#editor").click()
await page.keyboard.press("Control+A")
await page.keyboard.type("Pasted content")

# 单个复选框
await page.check("#agree-terms")       # 选中
await page.uncheck("#agree-terms")      # 取消选中
await page.check("#agree-terms", force=True)  # 强制选中（跳过状态检查）

# 检查状态
is_checked = await page.is_checked("#agree-terms")

# 切换状态
if await page.is_checked("#agree-terms"):
    await page.uncheck("#agree-terms")
else:
    await page.check("#agree-terms")

# 选择单选框
await page.check('input[name="gender"][value="male"]')

# 通过标签定位
await page.check("label:has-text('男') >> input[type='radio']")

# 获取当前选中值
selected = await page.locator('input[name="gender"]:checked').get_attribute("value")

# 标准下拉菜单
await page.select_option("#country", "CN")           # 按值选择
await page.select_option("#country", label="中国")    # 按标签选择
await page.select_option("#country", index=1)        # 按索引选择

# 多选下拉菜单
await page.select_option("#languages", ["Python", "JavaScript"])

# 获取所有选项
options = await page.locator("#country option").all_text_contents()
values = await page.locator("#country option").all()

async def select_custom_option(page: Page, dropdown_selector: str, option_text: str):
    """处理自定义下拉菜单"""
    # 点击打开下拉菜单
    await page.click(dropdown_selector)
    
    # 等待选项出现
    await page.wait_for_selector(".dropdown-menu", state="visible")
    
    # 查找并点击选项（支持模糊匹配）
    option_locator = page.locator(".dropdown-option").filter(has_text=re.compile(option_text, re.I))
    await option_locator.click()
    
    # 等待下拉菜单关闭
    await page.wait_for_selector(".dropdown-menu", state="hidden")

# 组合键操作（如选择全部）
async def select_all_options(page: Page, dropdown_selector: str):
    await page.click(dropdown_selector)
    await page.wait_for_selector(".dropdown-menu", state="visible")
    
    # Ctrl+A 全选
    await page.keyboard.press("Control+a")
    await page.click(".dropdown-menu .select-all-option")

# 单文件上传
await page.set_input_files('input[type="file"]', "/path/to/file.pdf")

# 多文件上传
await page.set_input_files('input[type="file"]', [
    "/path/to/file1.pdf",
    "/path/to/file2.pdf"
])

# 清除已选文件
await page.set_input_files('input[type="file"]', [])

# HTML5 拖拽上传
async def upload_via_drag_drop(page: Page, file_path: str):
    # 创建 DataTransfer 对象
    await page.evaluate("""(filePath) => {
        const dropZone = document.querySelector('.drop-zone');
        
        const dataTransfer = new DataTransfer();
        const file = new File(['content'], filePath.split('/').pop());
        dataTransfer.items.add(file);
        
        const dropEvent = new DragEvent('drop', {
            bubbles: true,
            cancelable: true,
            dataTransfer: dataTransfer
        });
        
        dropZone.dispatchEvent(dropEvent);
    }""", file_path)

async def upload_with_progress(page: Page, file_path: str):
    """带进度监控的上传"""
    progress = []
    
    # 监听 XHR 上传进度
    await page.route("**/upload**", lambda route: route.continue_())
    
    # 注入进度监控脚本
    await page.evaluate("""() => {
        const originalFetch = window.fetch;
        window.fetch = async (...args) => {
            const response = await originalFetch(...args);
            if (args[0].url.includes('upload')) {
                // 上传完成
                window.dispatchEvent(new CustomEvent('uploadComplete'));
            }
            return response;
        };
    }""")
    
    # 执行上传
    async with page.expect_event("uploadComplete", timeout=60000):
        await page.set_input_files('input[type="file"]', file_path)
    
    return True

# 触发单个字段验证
await page.locator("#email").blur()  # 失去焦点触发验证
await page.locator("#email").focus()

# 手动触发 HTML5 验证
await page.evaluate("""() => {
    const input = document.querySelector('#email');
    input.reportValidity();
}""")

# 检查验证状态
is_valid = await page.evaluate("""() => {
    return document.querySelector('#form').checkValidity();
}""")

# 方法1：点击提交按钮
await page.click('button[type="submit"]')

# 方法2：直接提交表单
await page.evaluate("""() => {
    document.querySelector('#form').submit();
}""")

# 方法3：模拟回车提交
await page.locator("#last-input").press("Enter")

# 等待提交完成
await page.wait_for_response(lambda r: r.url.includes("/api/submit"))
await page.wait_for_load_state("networkidle")

async def submit_form_with_validation(page: Page, form_selector: str):
    """带验证的表单提交"""
    # 尝试提交
    await page.click(f"{form_selector} button[type='submit']")
    
    # 检查是否有验证错误
    errors = await page.locator(".form-error, .error-message").all_text_contents()
    
    if errors:
        error_msg = "\n".join(errors)
        raise ValidationError(f"表单验证失败: {error_msg}")
    
    # 等待服务器响应
    try:
        await page.wait_for_response(
            lambda r: r.url.contains("/api/") and r.status in [200, 201, 302],
            timeout=15000
        )
    except Exception:
        raise SubmitError("表单提交超时")
    
    # 检查业务错误
    error_toast = await page.locator(".toast.error, .alert-danger").is_visible()
    if error_toast:
        error_text = await page.locator(".toast.error, .alert-danger").text_content()
        raise BusinessError(error_text)

async def detect_captcha(page: Page) -> bool:
    """检测页面是否存在验证码"""
    captcha_indicators = [
        # reCAPTCHA
        ".g-recaptcha",
        "iframe[src*='recaptcha']",
        # hCaptcha
        ".h-captcha",
        "iframe[src*='hcaptcha']",
        # 图像验证码
        ".captcha-container",
        "#captcha-modal",
        # 行为验证
        "[data-callback*='verify']",
        # 通用
        "iframe[src*='captcha']",
        "img[alt*='captcha']"
    ]
    
    for indicator in captcha_indicators:
        try:
            if await page.locator(indicator).count() > 0:
                return True
        except:
            continue
    
    # 检查页面内容特征
    page_content = await page.content()
    captcha_keywords = ["captcha", "验证", "人机", "我不是机器人", "点击验证"]
    
    for keyword in captcha_keywords:
        if keyword.lower() in page_content.lower():
            return True
    
    return False

async def solve_recaptcha_v2(page: Page, api_key: str, site_key: str, url: str):
    """
    通过第三方服务解决 reCAPTCHA v2
    api_key: 2Captcha 或类似服务提供的 API 密钥
    """
    # 获取页面 sitekey
    captcha_site_key = await page.evaluate("""() => {
        const el = document.querySelector('.g-recaptcha');
        return el ? el.getAttribute('data-sitekey') : null;
    }""")
    
    if not captcha_site_key:
        captcha_site_key = site_key
    
    # 调用验证码解决服务
    # 这里以 2Captcha 为例
    import http.client
    import urllib.parse
    
    conn = http.client.HTTPConnection("2captcha.com")
    params = urllib.parse.urlencode({
        "key": api_key,
        "method": "userrecaptcha",
        "googlekey": captcha_site_key,
        "pageurl": url,
        "json": 1
    })
    
    # 请求验证码 token（轮询直到获得结果）
    import time
    conn.request("GET", f"/res.php?{params}")
    response = json.loads(conn.getresponse().read())
    
    if response["status"] == 1:
        captcha_id = response["request"]
    else:
        raise CaptchaError(f"获取验证码失败: {response}")
    
    # 等待验证码解决
    time.sleep(20)
    
    while True:
        conn.request("GET", f"/res.php?key={api_key}&action=get&id={captcha_id}&json=1")
        response = json.loads(conn.getresponse().read())
        
        if response["status"] == 1:
            g_recaptcha_response = response["request"]
            break
        elif response["request"] == "CAPCHA_NOT_READY":
            time.sleep(5)
        else:
            raise CaptchaError(f"验证码解决失败: {response}")
    
    # 注入 token 到页面
    await page.evaluate(f"""(token) => {{
        document.querySelector('#g-recaptcha-response').innerHTML = token;
        if (typeof ___grecaptcha_cfg !== 'undefined') {{
            ___grecaptcha_cfg['clients'].map(client => {{
                Object.values(client).forEach(c => {{
                    if (c.callback) c.callback(token);
                }});
            }});
        }}
    }}""", g_recaptcha_response)
    
    # 触发验证码回调
    callback = await page.evaluate("""() => {
        const iframes = document.querySelectorAll('iframe[src*="recaptcha"]');
        for (const iframe of iframes) {
            const win = iframe.contentWindow;
            if (win.callback) return 'callback';
        }
        return null;
    }""")
    
    return g_recaptcha_response

async def solve_image_captcha(page: Page, captcha_img_selector: str, api_key: str):
    """
    解决图像验证码
    依赖于 OCR 服务（如 2Captcha、Anti-Captcha）
    """
    # 截取验证码图片
    captcha_element = page.locator(captcha_img_selector)
    captcha_image = await captcha_element.screenshot()
    
    # 保存临时文件
    temp_path = "/tmp/captcha.png"
    with open(temp_path, "wb") as f:
        f.write(captcha_image)
    
    # 发送到 OCR 服务
    # 这里使用 base64 编码
    import base64
    with open(temp_path, "rb") as f:
        img_base64 = base64.b64encode(f.read()).decode()
    
    # 调用 OCR API
    # 示例使用自定义 OCR 服务
    result = await call_ocr_service(api_key, img_base64)
    
    # 输入验证码
    captcha_input = page.locator("input[type='text'], input[name='captcha']").first
    await captcha_input.fill(result.text)
    
    return result.text

import asyncio
from dataclasses import dataclass
from typing import Protocol
import aiohttp

@dataclass
class Proxy:
    host: str
    port: int
    username: str = None
    password: str = None
    
    @property
    def url(self) -> str:
        if self.username and self.password:
            return f"http://{self.username}:{self.password}@{self.host}:{self.port}"
        return f"http://{self.host}:{self.port}"

class ProxyPool:
    """代理池管理器"""
    
    def __init__(self, proxies: list[Proxy] = None):
        self.proxies = proxies or []
        self.failed_proxies: set[str] = set()
        self.proxy_stats: dict[str, dict] = {}
        
    def add_proxy(self, proxy: Proxy):
        self.proxies.append(proxy)
        self.proxy_stats[proxy.url] = {"success": 0, "fail": 0, "latency": []}
    
    def get_proxy(self) -> Proxy:
        """获取最优代理（根据成功率和延迟）"""
        available = [p for p in self.proxies if p.url not in self.failed_proxies]
        
        if not available:
            available = self.proxies  # 重试失败的
            self.failed_proxies.clear()
        
        # 选择评分最高的代理
        scored = []
        for p in available:
            stats = self.proxy_stats.get(p.url, {"success": 0, "fail": 0, "latency": [1]})
            success_rate = stats["success"] / max(stats["success"] + stats["fail"], 1)
            avg_latency = sum(stats["latency"]) / max(len(stats["latency"]), 1)
            score = success_rate * 0.7 - avg_latency * 0.3
            scored.append((score, p))
        
        return min(scored, key=lambda x: x[0])[1]
    
    def report_result(self, proxy: Proxy, success: bool, latency: float = None):
        """报告代理使用结果"""
        if proxy.url not in self.proxy_stats:
            self.proxy_stats[proxy.url] = {"success": 0, "fail": 0, "latency": []}
        
        stats = self.proxy_stats[proxy.url]
        if success:
            stats["success"] += 1
        else:
            stats["fail"] += 1
            if stats["fail"] >= 3:  # 连续失败3次，标记为不可用
                self.failed_proxies.add(proxy.url)
        
        if latency is not None:
            stats["latency"].append(latency)
            if len(stats["latency"]) > 10:  # 保留最近10次延迟
                stats["latency"] = stats["latency"][-10:]

# Playwright 使用代理
async def browse_with_proxy(page: Page, proxy: Proxy):
    """使用指定代理浏览"""
    context = await page.context
    
    # 注意：Playwright 支持在 launch 或 newContext 时设置代理
    # 这里展示的是如何动态切换代理
    await page.close()
    
    new_context = await page.browser.new_context(
        proxy={"server": f"http://{proxy.host}:{proxy.port}"}
    )
    new_page = await new_context.new_page()
    
    return new_page

from playwright.async_api import BrowserContext

async def create_stealth_context(browser: Browser) -> BrowserContext:
    """创建反检测的浏览器上下文"""
    
    # 随机 User-Agent
    user_agents = [
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
    ]
    user_agent = random.choice(user_agents)
    
    context = await browser.new_context(
        user_agent=user_agent,
        locale="zh-CN",  # 设置区域
        timezone_id="Asia/Shanghai",  # 设置时区
        permissions=["geolocation"],  # 地理位置权限
        viewport={"width": 1920, "height": 1080},
    )
    
    # 拦截并修改 WebGL 指纹
    await context.add_init_script("""() => {
        // 修改 Canvas 指纹
        const originalToDataURL = HTMLCanvasElement.prototype.toDataURL;
        HTMLCanvasElement.prototype.toDataURL = function(...args) {
            const ctx = this.getContext('2d');
            if (ctx) {
                // 添加随机噪声
                const imageData = ctx.getImageData(0, 0, this.width, this.height);
                for (let i = 0; i < imageData.data.length; i += 4) {
                    imageData.data[i] += Math.random() * 0.1;
                    imageData.data[i + 1] += Math.random() * 0.1;
                    imageData.data[i + 2] += Math.random() * 0.1;
                }
                ctx.putImageData(imageData, 0, 0);
            }
            return originalToDataURL.apply(this, args);
        };
        
        // 修改 WebGL 指纹
        const getParameter = WebGLRenderingContext.prototype.getParameter;
        WebGLRenderingContext.prototype.getParameter = function(param) {
            if (param === 37445) { // UNMASKED_VENDOR
                return 'Intel Inc.';
            }
            if (param === 37446) { // UNMASKED_RENDERER
                return 'Intel Iris OpenGL Engine';
            }
            return getParameter.apply(this, arguments);
        };
        
        // 修改 navigator 属性
        Object.defineProperty(navigator, 'webdriver', {
            get: () => false
        });
        
        // 添加随机化的语言
        Object.defineProperty(navigator, 'languages', {
            get: () => ['zh-CN', 'zh', 'en-US', 'en']
        });
    }""")
    
    # 拦截 navigator.permissions 查询
    await context.add_init_script("""() => {
        const originalQuery = window.navigator.permissions.query;
        window.navigator.permissions.query = (parameters) => (
            parameters.name === 'notifications' ?
                Promise.resolve({ state: Notification.permission }) :
                originalQuery(parameters)
        );
    }""")
    
    return context

async def set_realistic_headers(page: Page):
    """设置真实的请求头"""
    await page.set_extra_http_headers({
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
        "DNT": "1",
        "Upgrade-Insecure-Requests": "1",
        "Cache-Control": "max-age=0",
    })

import asyncio
import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class RateLimitConfig:
    max_requests_per_second: float = 1.0
    max_requests_per_minute: float = 60.0
    max_requests_per_hour: float = 1000.0
    burst_size: int = 3  # 允许的突发请求数

class TokenBucketRateLimiter:
    """令牌桶限速器"""
    
    def __init__(self, config: RateLimitConfig):
        self.config = config
        self.tokens = config.burst_size
        self.last_update = time.time()
        self.min_interval = 1.0 / config.max_requests_per_second
        
        # 分钟级和小时级计数
        self.minute_requests: list[float] = []
        self.hour_requests: list[float] = []
    
    async def acquire(self):
        """获取请求许可（阻塞直到可以发送）"""
        while True:
            now = time.time()
            
            # 更新令牌
            elapsed = now - self.last_update
            self.tokens = min(
                self.config.burst_size,
                self.tokens + elapsed * self.config.max_requests_per_second
            )
            self.last_update = now
            
            # 清理过期计数
            self.minute_requests = [t for t in self.minute_requests if now - t < 60]
            self.hour_requests = [t for t in self.hour_requests if now - t < 3600]
            
            # 检查各层级限制
            if self.hour_requests and len(self.hour_requests) >= self.config.max_requests_per_hour:
                sleep_time = 3600 - (now - self.hour_requests[0])
                await asyncio.sleep(sleep_time)
                continue
            
            if self.minute_requests and len(self.minute_requests) >= self.config.max_requests_per_minute:
                sleep_time = 60 - (now - self.minute_requests[0])
                await asyncio.sleep(sleep_time)
                continue
            
            if self.tokens < 1:
                await asyncio.sleep(self.min_interval)
                continue
            
            # 获取令牌
            self.tokens -= 1
            self.minute_requests.append(now)
            self.hour_requests.append(now)
            return
    
    def get_stats(self) -> dict:
        """获取当前限速状态"""
        now = time.time()
        return {
            "tokens_available": self.tokens,
            "requests_this_minute": len([t for t in self.minute_requests if now - t < 60]),
            "requests_this_hour": len([t for t in self.hour_requests if now - t < 3600]),
        }

# 使用示例
async def polite_scraping(page: Page, urls: list[str]):
    """礼貌的爬取：自动限速"""
    config = RateLimitConfig(
        max_requests_per_second=1.0,
        max_requests_per_minute=30,
        max_requests_per_hour=500
    )
    limiter = TokenBucketRateLimiter(config)
    
    for url in urls:
        await limiter.acquire()
        
        try:
            await page.goto(url)
            # 处理页面...
            print(f"成功抓取: {url}")
        except Exception as e:
            print(f"抓取失败: {url}, 错误: {e}")
        
        # 额外随机延迟
        await asyncio.sleep(random.uniform(0.5, 2.0))

"""
Browser Agent - 基于 Playwright 的 Web 自动化框架
作者：HOS(安全风信子)
版本：1.0.0
"""

import asyncio
import re
import json
import logging
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Optional, Callable, Any
from pathlib import Path
from urllib.parse import urljoin, urlparse
import random

from playwright.async_api import async_playwright, Page, Browser, BrowserContext, Playwright
from playwright.async_api.models import FilePayload

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class AgentState(Enum):
    """Agent 状态枚举"""
    IDLE = auto()
    INITIALIZING = auto()
    NAVIGATING = auto()
    EXECUTING = auto()
    WAITING = auto()
    ERROR = auto()
    STOPPED = auto()


@dataclass
class ActionResult:
    """动作执行结果"""
    success: bool
    message: str = ""
    data: Any = None
    error: Optional[Exception] = None
    
    def __bool__(self):
        return self.success


@dataclass
class BrowserConfig:
    """浏览器配置"""
    headless: bool = True
    user_agent: Optional[str] = None
    viewport_width: int = 1920
    viewport_height: int = 1080
    locale: str = "zh-CN"
    timezone: str = "Asia/Shanghai"
    proxy_server: Optional[str] = None
    proxy_username: Optional[str] = None
    proxy_password: Optional[str] = None
    slow_mo: int = 0  # 慢动作延迟（毫秒）
    Stealth: bool = True  # 启用反检测


@dataclass
class AgentConfig:
    """Agent 配置"""
    default_timeout: int = 30000
    navigation_timeout: int = 60000
    retry_count: int = 3
    retry_delay: int = 1000
    screenshot_on_error: bool = True
    trace_on_error: bool = True
    data_dir: str = "./data"


class InstructionParser:
    """
    指令解析器：将自然语言指令转换为可执行的动作序列
    """
    
    # 动作模式定义
    ACTION_PATTERNS = {
        "navigate": [
            r"打开\s*(.+?)(?:\s|$)",
            r"访问\s*(.+?)(?:\s|$)",
            r"转到\s*(.+?)(?:\s|$)",
            r"导航到\s*(.+?)(?:\s|$)",
        ],
        "click": [
            r"点击\s*(.+?)(?:\s|$)",
            r"单击\s*(.+?)(?:\s|$)",
            r"选择\s*(.+?)(?:\s|$)",
        ],
        "input": [
            r"在\s*(.+?)\s*(?:中|里)?输入\s*['\"](.+?)['\"](?:\s|$)",
            r"填写\s*(.+?)\s*(?:为|成)\s*['\"](.+?)['\"](?:\s|$)",
        ],
        "select": [
            r"选择\s*(.+?)\s*(?:的|为)?\s*['\"](.+?)['\"](?:\s|$)",
        ],
        "wait": [
            r"等待\s*(.+?)(?:\s|$)",
            r"等待\s*(\d+)\s*(?:秒|毫秒)(?:\s|$)",
        ],
        "extract": [
            r"提取\s*(.+?)(?:\s|$)",
            r"获取\s*(.+?)(?:\s|$)",
            r"抓取\s*(.+?)(?:\s|$)",
        ],
        "scroll": [
            r"滚动\s*(?:到)?\s*(.+?)(?:\s|$)",
            r"向下滚动",
            r"向上滚动",
        ],
        "screenshot": [
            r"截图",
            r"截屏",
        ],
    }
    
    def __init__(self):
        self.compiled_patterns = {}
        for action, patterns in self.ACTION_PATTERNS.items():
            self.compiled_patterns[action] = [re.compile(p) for p in patterns]
    
    def parse(self, instruction: str) -> list[dict]:
        """
        解析自然语言指令为动作序列
        
        Args:
            instruction: 自然语言指令
            
        Returns:
            动作列表，每个动作为字典，包含 type 和参数
        """
        instruction = instruction.strip()
        actions = []
        
        # 逐个检测动作模式
        for action, patterns in self.compiled_patterns.items():
            for pattern in patterns:
                match = pattern.search(instruction)
                if match:
                    action_def = {"type": action}
                    groups = match.groups()
                    
                    if action == "navigate" and groups:
                        action_def["target"] = groups[0]
                    elif action == "click" and groups:
                        action_def["target"] = groups[0]
                    elif action == "input" and len(groups) >= 2:
                        action_def["field"] = groups[0]
                        action_def["value"] = groups[1]
                    elif action == "select" and len(groups) >= 2:
                        action_def["field"] = groups[0]
                        action_def["value"] = groups[1]
                    elif action == "wait" and groups:
                        if groups[0].isdigit():
                            action_def["duration"] = int(groups[0])
                        else:
                            action_def["target"] = groups[0]
                    elif action == "extract" and groups:
                        action_def["target"] = groups[0]
                    elif action == "scroll" and groups:
                        action_def["direction"] = groups[0] if groups else "down"
                    
                    actions.append(action_def)
                    break  # 每个指令只匹配一个模式
        
        return actions


class ElementLocator:
    """
    智能元素定位器：支持多种定位策略的自动选择和回退
    """
    
    def __init__(self, page: Page):
        self.page = page
    
    async def locate(self, target: str) -> Optional[Any]:
        """
        智能定位元素
        
        Args:
            target: 定位目标（可以是文本、ID、类名、选择器等）
            
        Returns:
            定位到的元素
        """
        locator = None
        
        # 策略1：精确文本匹配（按钮、链接等）
        text_patterns = [
            (rf".*('{re.escape(target)}'|{re.escape(target)}).*", "text"),
            (rf".*{re.escape(target)}.*", "text"),
        ]
        
        for pattern, match_type in text_patterns:
            locator = self.page.get_by_text(re.compile(pattern, re.I))
            if await locator.count() > 0:
                logger.info(f"通过文本匹配定位: {target}")
                return locator
        
        # 策略2：ARIA 角色定位
        role_patterns = ["button", "link", "textbox", "checkbox", "radio", "menuitem"]
        for role in role_patterns:
            locator = self.page.get_by_role(role, name=re.compile(target, re.I))
            if await locator.count() > 0:
                logger.info(f"通过角色定位: {target}")
                return locator
        
        # 策略3：placeholder 定位
        locator = self.page.get_by_placeholder(re.compile(target, re.I))
        if await locator.count() > 0:
            logger.info(f"通过 placeholder 定位: {target}")
            return locator
        
        # 策略4：label 关联定位
        locator = self.page.get_by_label(re.compile(target, re.I))
        if await locator.count() > 0:
            logger.info(f"通过 label 定位: {target}")
            return locator
        
        # 策略5：CSS 选择器（ID、类、属性等）
        selector_patterns = [
            f"#{target}",
            f".{target}",
            f"[id*='{target}']",
            f"[class*='{target}']",
            f"[name='{target}']",
            target,
        ]
        
        for selector in selector_patterns:
            try:
                locator = self.page.locator(selector)
                if await locator.count() > 0:
                    logger.info(f"通过 CSS 选择器定位: {selector}")
                    return locator
            except Exception:
                continue
        
        # 策略6：XPath 文本包含
        xpath_patterns = [
            f"//*[contains(text(), '{target}')]",
            f"//button[contains(text(), '{target}')]",
            f"//a[contains(text(), '{target}')]",
            f"//input[@placeholder='{target}']",
        ]
        
        for xpath in xpath_patterns:
            locator = self.page.locator(f"xpath={xpath}")
            if await locator.count() > 0:
                logger.info(f"通过 XPath 定位: {xpath}")
                return locator
        
        logger.warning(f"无法定位元素: {target}")
        return None
    
    async def locate_with_retry(self, target: str, max_retries: int = 3, delay: int = 1000) -> Optional[Any]:
        """带重试的元素定位"""
        for attempt in range(max_retries):
            locator = await self.locate(target)
            if locator:
                return locator
            
            if attempt < max_retries - 1:
                await asyncio.sleep(delay / 1000)
        
        return None


class ActionExecutor:
    """
    动作执行器：执行各种浏览器操作
    """
    
    def __init__(self, page: Page, locator: ElementLocator):
        self.page = page
        self.locator = locator
    
    async def navigate(self, url: str) -> ActionResult:
        """导航到指定 URL"""
        try:
            if not url.startswith(("http://", "https://")):
                url = "https://" + url
            
            await self.page.goto(url, wait_until="domcontentloaded", timeout=30000)
            
            # 等待页面基本可交互
            await self.page.wait_for_load_state("networkidle", timeout=15000)
            
            return ActionResult(success=True, message=f"成功导航到 {url}")
        except Exception as e:
            logger.error(f"导航失败: {e}")
            return ActionResult(success=False, message=str(e), error=e)
    
    async def click(self, target: str) -> ActionResult:
        """点击元素"""
        try:
            locator = await self.locator.locate_with_retry(target)
            if not locator:
                return ActionResult(success=False, message=f"找不到元素: {target}")
            
            await locator.click(timeout=10000)
            await self.page.wait_for_load_state("domcontentloaded", timeout=10000)
            
            return ActionResult(success=True, message=f"成功点击: {target}")
        except Exception as e:
            logger.error(f"点击失败: {e}")
            return ActionResult(success=False, message=str(e), error=e)
    
    async def input_text(self, field: str, value: str) -> ActionResult:
        """输入文本"""
        try:
            locator = await self.locator.locate_with_retry(field)
            if not locator:
                return ActionResult(success=False, message=f"找不到输入框: {field}")
            
            await locator.clear()
            await locator.fill(value)
            
            return ActionResult(success=True, message=f"成功输入: {value}")
        except Exception as e:
            logger.error(f"输入失败: {e}")
            return ActionResult(success=False, message=str(e), error=e)
    
    async def select_option(self, field: str, value: str) -> ActionResult:
        """选择选项"""
        try:
            locator = await self.locator.locate_with_retry(field)
            if not locator:
                return ActionResult(success=False, message=f"找不到选择框: {field}")
            
            # 尝试多种选择方式
            try:
                await locator.select_option(value)
            except:
                try:
                    await locator.select_option(label=value)
                except:
                    await locator.select_option(index=int(value))
            
            return ActionResult(success=True, message=f"成功选择: {value}")
        except Exception as e:
            logger.error(f"选择失败: {e}")
            return ActionResult(success=False, message=str(e), error=e)
    
    async def wait(self, duration: int = None, target: str = None) -> ActionResult:
        """等待"""
        try:
            if duration:
                # 毫秒转秒
                await asyncio.sleep(duration / 1000)
            elif target:
                locator = await self.locator.locate_with_retry(target)
                if locator:
                    await locator.wait_for(state="visible", timeout=30000)
            
            return ActionResult(success=True, message="等待完成")
        except Exception as e:
            return ActionResult(success=False, message=str(e), error=e)
    
    async def extract(self, selector: str, attribute: str = "text") -> ActionResult:
        """提取内容"""
        try:
            locator = self.page.locator(selector)
            count = await locator.count()
            
            if count == 0:
                return ActionResult(success=False, message=f"找不到元素: {selector}")
            
            if attribute == "text":
                data = await locator.all_text_contents()
            elif attribute == "inner_html":
                data = await locator.all_inner_htmls()
            elif attribute == "value":
                data = await locator.all_input_values()
            else:
                data = [await el.get_attribute(attribute) for el in [locator] if await locator.count() > 0]
            
            return ActionResult(success=True, message="提取成功", data=data)
        except Exception as e:
            return ActionResult(success=False, message=str(e), error=e)
    
    async def scroll(self, direction: str = "down") -> ActionResult:
        """滚动页面"""
        try:
            if direction == "down":
                await self.page.evaluate("window.scrollBy(0, window.innerHeight)")
            elif direction == "up":
                await self.page.evaluate("window.scrollBy(0, -window.innerHeight)")
            elif "px" in direction:
                await self.page.evaluate(f"window.scrollBy(0, {direction})")
            
            return ActionResult(success=True, message=f"滚动{direction}完成")
        except Exception as e:
            return ActionResult(success=False, message=str(e), error=e)
    
    async def screenshot(self, path: str = None) -> ActionResult:
        """截图"""
        try:
            if not path:
                path = f"screenshot_{int(asyncio.get_event_loop().time())}.png"
            
            await self.page.screenshot(path=path, full_page=True)
            
            return ActionResult(success=True, message=f"截图保存到 {path}", data=path)
        except Exception as e:
            return ActionResult(success=False, message=str(e), error=e)


class BrowserAgent:
    """
    Browser Agent 主类：协调各模块工作
    """
    
    def __init__(self, config: AgentConfig = None):
        self.config = config or AgentConfig()
        self.playwright: Optional[Playwright] = None
        self.browser: Optional[Browser] = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None
        
        self.state = AgentState.IDLE
        self.parser = InstructionParser()
        self.locator: Optional[ElementLocator] = None
        self.executor: Optional[ActionExecutor] = None
        
        self.session_data: dict = {}
        self.action_history: list[ActionResult] = []
    
    async def initialize(self, browser_config: BrowserConfig = None) -> bool:
        """初始化浏览器"""
        try:
            self.state = AgentState.INITIALIZING
            
            browser_config = browser_config or BrowserConfig()
            
            self.playwright = await async_playwright().start()
            
            # 启动浏览器
            launch_options = {
                "headless": browser_config.headless,
                "slow_mo": browser_config.slow_mo,
            }
            
            # 代理配置
            if browser_config.proxy_server:
                launch_options["proxy"] = {
                    "server": browser_config.proxy_server,
                    "username": browser_config.proxy_username,
                    "password": browser_config.proxy_password,
                }
            
            self.browser = await self.playwright.chromium.launch(**launch_options)
            
            # 创建上下文
            context_options = {
                "viewport": {
                    "width": browser_config.viewport_width,
                    "height": browser_config.viewport_height,
                },
                "locale": browser_config.locale,
                "timezone_id": browser_config.timezone,
            }
            
            if browser_config.user_agent:
                context_options["user_agent"] = browser_config.user_agent
            
            self.context = await self.browser.new_context(**context_options)
            
            # 反检测脚本
            if browser_config.stealth:
                await self._apply_stealth_scripts()
            
            self.page = await self.context.new_page()
            
            # 初始化组件
            self.locator = ElementLocator(self.page)
            self.executor = ActionExecutor(self.page, self.locator)
            
            self.state = AgentState.IDLE
            logger.info("Browser Agent 初始化成功")
            return True
            
        except Exception as e:
            logger.error(f"初始化失败: {e}")
            self.state = AgentState.ERROR
            return False
    
    async def _apply_stealth_scripts(self):
        """应用反检测脚本"""
        await self.context.add_init_script("""() => {
            // 修改 navigator.webdriver
            Object.defineProperty(navigator, 'webdriver', {
                get: () => false
            });
            
            // 修改 Canvas 指纹
            const originalToDataURL = HTMLCanvasElement.prototype.toDataURL;
            HTMLCanvasElement.prototype.toDataURL = function(...args) {
                const ctx = this.getContext('2d');
                if (ctx) {
                    const imageData = ctx.getImageData(0, 0, this.width, this.height);
                    for (let i = 0; i < imageData.data.length; i += 4) {
                        imageData.data[i] += Math.random() * 0.0001;
                    }
                    ctx.putImageData(imageData, 0, 0);
                }
                return originalToDataURL.apply(this, args);
            };
            
            // 修改 WebGL 指纹
            const getParameter = WebGLRenderingContext.prototype.getParameter;
            WebGLRenderingContext.prototype.getParameter = function(param) {
                if (param === 37445) return 'Intel Inc.';
                if (param === 37446) return 'Intel Iris OpenGL Engine';
                return getParameter.apply(this, arguments);
            };
            
            // 修改 permissions
            const originalQuery = navigator.permissions.query;
            navigator.permissions.query = (parameters) => (
                parameters.name === 'notifications' ?
                    Promise.resolve({ state: Notification.permission }) :
                    originalQuery(parameters)
            );
            
            // 添加真实的语言列表
            Object.defineProperty(navigator, 'languages', {
                get: () => ['zh-CN', 'zh', 'en-US', 'en']
            });
        }""")
    
    async def execute(self, instruction: str) -> ActionResult:
        """
        执行自然语言指令
        
        Args:
            instruction: 自然语言指令
            
        Returns:
            执行结果
        """
        if self.state == AgentState.ERROR:
            return ActionResult(success=False, message="Agent 处于错误状态")
        
        if not self.page:
            return ActionResult(success=False, message="Agent 未初始化")
        
        try:
            self.state = AgentState.EXECUTING
            
            # 解析指令
            actions = self.parser.parse(instruction)
            
            if not actions:
                return ActionResult(success=False, message=f"无法解析指令: {instruction}")
            
            logger.info(f"解析到 {len(actions)} 个动作: {actions}")
            
            # 执行动作序列
            results = []
            for action in actions:
                result = await self._execute_action(action)
                results.append(result)
                self.action_history.append(result)
                
                if not result.success:
                    # 动作失败，可选择是否继续
                    logger.warning(f"动作失败，继续执行: {result.message}")
            
            # 返回最后一个结果
            return results[-1] if results else ActionResult(success=False, message="无执行结果")
            
        except Exception as e:
            logger.error(f"执行异常: {e}")
            self.state = AgentState.ERROR
            return ActionResult(success=False, message=str(e), error=e)
        finally:
            self.state = AgentState.IDLE
    
    async def _execute_action(self, action: dict) -> ActionResult:
        """执行单个动作"""
        action_type = action["type"]
        
        self.state = AgentState.EXECUTING
        
        if action_type == "navigate":
            return await self.executor.navigate(action.get("target", ""))
        
        elif action_type == "click":
            return await self.executor.click(action.get("target", ""))
        
        elif action_type == "input":
            return await self.executor.input_text(
                action.get("field", ""),
                action.get("value", "")
            )
        
        elif action_type == "select":
            return await self.executor.select_option(
                action.get("field", ""),
                action.get("value", "")
            )
        
        elif action_type == "wait":
            return await self.executor.wait(
                duration=action.get("duration"),
                target=action.get("target")
            )
        
        elif action_type == "extract":
            return await self.executor.extract(action.get("target", ""))
        
        elif action_type == "scroll":
            return await self.executor.scroll(action.get("direction", "down"))
        
        elif action_type == "screenshot":
            return await self.executor.screenshot()
        
        else:
            return ActionResult(success=False, message=f"未知动作类型: {action_type}")
    
    async def execute_script(self, script: str) -> Any:
        """在页面中执行 JavaScript"""
        if not self.page:
            raise RuntimeError("Agent 未初始化")
        return await self.page.evaluate(script)
    
    async def save_session(self, path: str = None):
        """保存会话状态（Cookie、LocalStorage 等）"""
        if not self.context:
            return
        
        path = path or f"session_{int(asyncio.get_event_loop().time())}.json"
        
        storage = await self.context.storage_state()
        
        with open(path, "w", encoding="utf-8") as f:
            json.dump(storage, f, indent=2, ensure_ascii=False)
        
        logger.info(f"会话已保存到 {path}")
        return path
    
    async def load_session(self, path: str):
        """加载会话状态"""
        if not self.context:
            return
        
        await self.context.storage_state(path=path)
        logger.info(f"会话已从 {path} 加载")
    
    async def close(self):
        """关闭 Agent"""
        if self.page:
            await self.page.close()
        if self.context:
            await self.context.close()
        if self.browser:
            await self.browser.close()
        if self.playwright:
            await self.playwright.stop()
        
        self.state = AgentState.STOPPED
        logger.info("Browser Agent 已关闭")

"""
Browser Agent 使用示例
"""

import asyncio
from browser_agent import BrowserAgent, BrowserConfig, AgentConfig

async def demo_basic_usage():
    """基础用法演示"""
    agent = BrowserAgent()
    
    # 初始化
    await agent.initialize(BrowserConfig(headless=True))
    
    # 执行一系列操作
    # 1. 打开百度
    result = await agent.execute("打开 https://www.baidu.com")
    print(f"导航结果: {result.success}, {result.message}")
    
    # 2. 在搜索框输入 Python
    await agent.execute("在搜索框输入 'Python'")
    
    # 3. 点击搜索按钮
    await agent.execute("点击 百度一下")
    
    # 4. 截图
    await agent.execute("截图")
    
    # 关闭
    await agent.close()

async def demo_login_flow():
    """登录流程演示"""
    agent = BrowserAgent()
    await agent.initialize()
    
    # 打开登录页面
    await agent.execute("打开 https://example.com/login")
    
    # 填写用户名
    await agent.execute("在 用户名 输入 'admin'")
    
    # 填写密码
    await agent.execute("在 密码 输入 'password123'")
    
    # 点击登录按钮
    await agent.execute("点击 登录")
    
    # 等待页面跳转
    await asyncio.sleep(2)
    
    # 截图保存登录后状态
    await agent.executor.screenshot("after_login.png")
    
    await agent.close()

async def demo_data_extraction():
    """数据提取演示"""
    agent = BrowserAgent()
    await agent.initialize()
    
    await agent.execute("打开 https://news.ycombinator.com")
    
    # 提取标题列表
    result = await agent.execute("获取所有标题")
    
    if result.success:
        titles = result.data
        for i, title in enumerate(titles[:10], 1):
            print(f"{i}. {title}")
    
    await agent.close()

async def demo_session_persistence():
    """会话保持演示"""
    agent = BrowserAgent()
    
    # 尝试加载已保存的会话
    session_path = "session.json"
    try:
        await agent.load_session(session_path)
        print("加载已保存的会话")
    except:
        print("创建新会话")
    
    await agent.initialize()
    
    # 执行操作...
    await agent.execute("打开 https://example.com")
    
    # 保存会话
    await agent.save_session(session_path)
    
    await agent.close()

if __name__ == "__main__":
    # 运行基础演示
    asyncio.run(demo_basic_usage())

"""
Browser Agent 扩展模块
"""

import asyncio
from typing import Optional
import httpx
from browser_agent import BrowserAgent, BrowserConfig, ActionResult

class CaptchaIntegration:
    """验证码集成模块"""
    
    def __init__(self, agent: BrowserAgent, api_key: str, provider: str = "2captcha"):
        self.agent = agent
        self.api_key = api_key
        self.provider = provider
    
    async def solve_recaptcha(self, site_key: str = None) -> Optional[str]:
        """解决 reCAPTCHA"""
        if not site_key:
            site_key = await self.agent.execute_script("""
                document.querySelector('.g-recaptcha')?.getAttribute('data-sitekey')
            """)
        
        if not site_key:
            return None
        
        # 调用验证码解决服务
        async with httpx.AsyncClient() as client:
            if self.provider == "2captcha":
                # 获取验证码 ID
                resp = await client.get(
                    f"http://2captcha.com/res.php?key={self.api_key}&method=userrecaptcha&googlekey={site_key}&pageurl={self.agent.page.url}&json=1"
                )
                data = resp.json()
                
                if data["status"] != 1:
                    return None
                
                captcha_id = data["request"]
                
                # 轮询获取结果
                await asyncio.sleep(20)
                
                for _ in range(30):
                    resp = await client.get(
                        f"http://2captcha.com/res.php?key={self.api_key}&action=get&id={captcha_id}&json=1"
                    )
                    data = resp.json()
                    
                    if data["status"] == 1:
                        return data["request"]
                    
                    await asyncio.sleep(5)
        
        return None
    
    async def detect_and_solve(self) -> bool:
        """检测并解决验证码"""
        page = self.agent.page
        
        # 检测 reCAPTCHA
        recaptcha = await page.query_selector(".g-recaptcha")
        if recaptcha:
            token = await self.solve_recaptcha()
            if token:
                # 注入 token
                await self.agent.execute_script(f"""
                    document.getElementById('g-recaptcha-response').innerHTML = '{token}';
                """)
                return True
        
        return False


class ProxyRotator:
    """代理轮换模块"""
    
    def __init__(self, agent: BrowserAgent, proxy_list: list[dict]):
        self.agent = agent
        self.proxies = proxy_list
        self.current_index = 0
    
    async def rotate(self):
        """切换到下一个代理"""
        if not self.proxies:
            return
        
        proxy = self.proxies[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxies)
        
        # 关闭当前页面
        await self.agent.page.close()
        await self.agent.context.close()
        
        # 使用新代理重新创建上下文
        browser_config = BrowserConfig(
            proxy_server=proxy.get("server"),
            proxy_username=proxy.get("username"),
            proxy_password=proxy.get("password"),
        )
        
        await self.agent.initialize(browser_config)
        
        print(f"已切换到代理: {proxy.get('server')}")
    
    def get_current_proxy(self) -> Optional[dict]:
        """获取当前代理"""
        if self.proxies:
            return self.proxies[self.current_index - 1] if self.current_index > 0 else self.proxies[0]
        return None


class RateLimiter:
    """请求限流器"""
    
    def __init__(self, max_per_second: float = 1.0):
        self.max_per_second = max_per_second
        self.min_interval = 1.0 / max_per_second
        self.last_request = 0
    
    async def wait(self):
        """等待以满足限流要求"""
        now = asyncio.get_event_loop().time()
        elapsed = now - self.last_request
        
        if elapsed < self.min_interval:
            await asyncio.sleep(self.min_interval - elapsed)
        
        self.last_request = asyncio.get_event_loop().time()

class SmartWaiter:
    """智能等待器"""
    
    def __init__(self, page: Page):
        self.page = page
    
    async def wait_for_element(self, selector: str, timeout: int = 30000):
        """等待元素出现"""
        locator = self.page.locator(selector)
        await locator.wait_for(state="attached", timeout=timeout)
        return locator
    
    async def wait_for_text(self, text: str, timeout: int = 30000):
        """等待文本出现"""
        await self.page.wait_for_function(
            f"""() => document.body.innerText.includes('{text}')""",
            timeout=timeout
        )
    
    async def wait_for_navigation(self, timeout: int = 30000):
        """等待导航完成"""
        await asyncio.gather(
            self.page.wait_for_load_state("networkidle", timeout=timeout),
            self.page.wait_for_load_state("domcontentloaded", timeout=timeout),
        )
    
    async def wait_for_network_idle_with_timeout(self, timeout: int = 30000):
        """带自定义超时的网络空闲等待"""
        try:
            await self.page.wait_for_load_state("networkidle", timeout=timeout)
        except Exception:
            # 网络空闲等待超时，强制继续
            pass

class ErrorRecovery:
    """错误恢复机制"""
    
    def __init__(self, agent: BrowserAgent):
        self.agent = agent
        self.max_retries = 3
    
    async def with_recovery(self, action: Callable, *args, **kwargs) -> ActionResult:
        """带错误恢复的动作执行"""
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                result = await action(*args, **kwargs)
                
                if result.success:
                    return result
                
                last_error = result.error
                
            except Exception as e:
                last_error = e
                logger.error(f"尝试 {attempt + 1} 失败: {e}")
            
            # 重试前等待
            if attempt < self.max_retries - 1:
                delay = 1000 * (2 ** attempt)  # 指数退避
                await asyncio.sleep(delay / 1000)
                
                # 尝试刷新页面恢复状态
                await self._refresh_state()
        
        return ActionResult(
            success=False,
            message=f"重试 {self.max_retries} 次后仍失败: {last_error}",
            error=last_error
        )
    
    async def _refresh_state(self):
        """刷新页面状态"""
        try:
            await self.agent.page.reload()
            await self.agent.page.wait_for_load_state("domcontentloaded")
        except Exception as e:
            logger.warning(f"状态刷新失败: {e}")

async def parallel_navigation(pages: list[Page], urls: list[str]):
    """并行导航多个页面"""
    tasks = [page.goto(url) for page, url in zip(pages, urls)]
    await asyncio.gather(*tasks)

async def parallel_extraction(page: Page, selectors: list[str]):
    """并行提取多个选择器的内容"""
    tasks = [
        page.locator(selector).all_text_contents()
        for selector in selectors
    ]
    results = await asyncio.gather(*tasks)
    return dict(zip(selectors, results))

async def batch_operations(agent: BrowserAgent, urls: list[str]):
    """批量操作复用上下文"""
    # 在同一上下文中打开多个页面
    pages = []
    
    for url in urls:
        page = await agent.context.new_page()
        pages.append(page)
        
        await page.goto(url)
        # 执行操作...
    
    # 批量关闭
    for page in pages:
        await page.close()

# 禁用不必要的资源加载
await page.route("**/*", lambda route: route.abort() if route.request.resource_type in ["image", "stylesheet", "font"] else route.continue_())

# 启用请求缓存
await page.context.set_extra_http_headers({
    "Cache-Control": "max-age=86400"
})

# config.py
from dataclasses import dataclass

@dataclass
class Environment:
    base_url: str
    api_timeout: int
    headless: bool
    max_retries: int

ENVIRONMENTS = {
    "dev": Environment(
        base_url="https://dev.example.com",
        api_timeout=30000,
        headless=False,
        max_retries=3,
    ),
    "prod": Environment(
        base_url="https://www.example.com",
        api_timeout=60000,
        headless=True,
        max_retries=5,
    ),
}

# 使用
import os
env_name = os.getenv("ENV", "dev")
config = ENVIRONMENTS[env_name]

import logging
from functools import wraps

def log_action(action_name: str):
    """动作执行日志装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            logger.info(f"[ACTION] 开始执行: {action_name}")
            try:
                result = await func(*args, **kwargs)
                if result.success:
                    logger.info(f"[ACTION] 成功: {action_name}")
                else:
                    logger.warning(f"[ACTION] 失败: {action_name} - {result.message}")
                return result
            except Exception as e:
                logger.error(f"[ACTION] 异常: {action_name} - {e}")
                raise
        return wrapper
    return decorator

import pytest

@pytest.fixture
async def browser_agent():
    """Browser Agent 测试 fixture"""
    agent = BrowserAgent()
    await agent.initialize()
    yield agent
    await agent.close()

@pytest.mark.asyncio
async def test_login_flow(browser_agent):
    """登录流程测试"""
    await browser_agent.execute("打开 https://example.com/login")
    await browser_agent.execute("在用户名输入 'test'")
    await browser_agent.execute("在密码输入 'password'")
    result = await browser_agent.execute("点击登录")
    
    assert result.success

"""
Browser Agent - 完整实现
作者：HOS(安全风信子)
日期：2026-05-24
版本：1.0.0
"""

import asyncio
import re
import json
import logging
import random
import time
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Optional, Callable, Any, List
from urllib.parse import urlparse

from playwright.async_api import async_playwright, Page, Browser, BrowserContext, Playwright

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class AgentState(Enum):
    """Agent 状态枚举"""
    IDLE = auto()
    INITIALIZING = auto()
    NAVIGATING = auto()
    EXECUTING = auto()
    WAITING = auto()
    ERROR = auto()
    STOPPED = auto()


@dataclass
class ActionResult:
    """动作执行结果"""
    success: bool
    message: str = ""
    data: Any = None
    error: Optional[Exception] = None


@dataclass
class BrowserConfig:
    """浏览器配置"""
    headless: bool = True
    user_agent: Optional[str] = None
    viewport_width: int = 1920
    viewport_height: int = 1080
    locale: str = "zh-CN"
    timezone: str = "Asia/Shanghai"
    proxy_server: Optional[str] = None
    proxy_username: Optional[str] = None
    proxy_password: Optional[str] = None
    slow_mo: int = 0
    stealth: bool = True


class InstructionParser:
    """自然语言指令解析器"""
    
    ACTION_PATTERNS = {
        "navigate": [r"(?:打开|访问|转到|导航到)\s*(.+?)(?:\s*$)",],
        "click": [r"(?:点击|单击|选择)\s*(.+?)(?:\s*$)",],
        "input": [r"在\s*(.+?)\s*(?:中|里)?输入\s*['\"](.+?)['\"](?:\s*$)",],
        "wait": [r"等待\s*(\d+)?\s*(?:秒|毫秒)?(?:\s*$)",],
        "screenshot": [r"(?:截图|截屏)(?:\s*$)",],
        "extract": [r"(?:提取|获取|抓取)\s*(.+?)(?:\s*$)",],
    }
    
    def __init__(self):
        self.compiled_patterns = {}
        for action, patterns in self.ACTION_PATTERNS.items():
            self.compiled_patterns[action] = [re.compile(p) for p in patterns]
    
    def parse(self, instruction: str) -> List[dict]:
        """解析自然语言指令"""
        instruction = instruction.strip()
        actions = []
        
        for action, patterns in self.compiled_patterns.items():
            for pattern in patterns:
                match = pattern.search(instruction)
                if match:
                    action_def = {"type": action}
                    groups = match.groups()
                    
                    if action == "navigate" and groups:
                        action_def["target"] = groups[0]
                    elif action == "click" and groups:
                        action_def["target"] = groups[0]
                    elif action == "input" and len(groups) >= 2:
                        action_def["field"] = groups[0]
                        action_def["value"] = groups[1]
                    elif action == "wait" and groups and groups[0]:
                        action_def["duration"] = int(groups[0]) * (1000 if '毫秒' not in instruction else 1)
                    elif action == "extract" and groups:
                        action_def["target"] = groups[0]
                    
                    actions.append(action_def)
                    break
        
        return actions


class BrowserAgent:
    """
    Browser Agent 主类
    """
    
    def __init__(self):
        self.playwright: Optional[Playwright] = None
        self.browser: Optional[Browser] = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None
        self.state = AgentState.IDLE
        self.parser = InstructionParser()
    
    async def initialize(self, config: BrowserConfig = None) -> bool:
        """初始化浏览器"""
        try:
            self.state = AgentState.INITIALIZING
            config = config or BrowserConfig()
            
            self.playwright = await async_playwright().start()
            
            launch_options = {
                "headless": config.headless,
                "slow_mo": config.slow_mo,
            }
            
            if config.proxy_server:
                launch_options["proxy"] = {
                    "server": config.proxy_server,
                    "username": config.proxy_username,
                    "password": config.proxy_password,
                }
            
            self.browser = await self.playwright.chromium.launch(**launch_options)
            
            context_options = {
                "viewport": {"width": config.viewport_width, "height": config.viewport_height},
                "locale": config.locale,
                "timezone_id": config.timezone,
            }
            
            if config.user_agent:
                context_options["user_agent"] = config.user_agent
            
            self.context = await self.browser.new_context(**context_options)
            
            if config.stealth:
                await self._apply_stealth()
            
            self.page = await self.context.new_page()
            self.state = AgentState.IDLE
            logger.info("Browser Agent 初始化成功")
            return True
            
        except Exception as e:
            logger.error(f"初始化失败: {e}")
            self.state = AgentState.ERROR
            return False
    
    async def _apply_stealth(self):
        """应用反检测脚本"""
        await self.context.add_init_script("""() => {
            Object.defineProperty(navigator, 'webdriver', { get: () => false });
            const originalToDataURL = HTMLCanvasElement.prototype.toDataURL;
            HTMLCanvasElement.prototype.toDataURL = function(...args) {
                const ctx = this.getContext('2d');
                if (ctx) {
                    const imageData = ctx.getImageData(0, 0, this.width, this.height);
                    for (let i = 0; i < imageData.data.length; i += 4) {
                        imageData.data[i] += Math.random() * 0.0001;
                    }
                    ctx.putImageData(imageData, 0, 0);
                }
                return originalToDataURL.apply(this, args);
            };
            Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh', 'en-US', 'en'] });
        }""")
    
    async def execute(self, instruction: str) -> ActionResult:
        """执行自然语言指令"""
        if not self.page:
            return ActionResult(success=False, message="Agent 未初始化")
        
        try:
            self.state = AgentState.EXECUTING
            actions = self.parser.parse(instruction)
            
            if not actions:
                return ActionResult(success=False, message=f"无法解析指令: {instruction}")
            
            result = ActionResult(success=True, message="执行成功")
            
            for action in actions:
                result = await self._execute_action(action)
                if not result.success:
                    break
            
            self.state = AgentState.IDLE
            return result
            
        except Exception as e:
            logger.error(f"执行异常: {e}")
            self.state = AgentState.ERROR
            return ActionResult(success=False, message=str(e), error=e)
    
    async def _execute_action(self, action: dict) -> ActionResult:
        """执行单个动作"""
        action_type = action["type"]
        
        if action_type == "navigate":
            target = action.get("target", "")
            if not target.startswith(("http://", "https://")):
                target = "https://" + target
            try:
                await self.page.goto(target, wait_until="domcontentloaded", timeout=30000)
                return ActionResult(success=True, message=f"导航到 {target}")
            except Exception as e:
                return ActionResult(success=False, message=str(e), error=e)
        
        elif action_type == "click":
            target = action.get("target", "")
            try:
                # 尝试多种定位方式
                locator = self._create_locator(target)
                await locator.click(timeout=10000)
                return ActionResult(success=True, message=f"点击 {target}")
            except Exception as e:
                return ActionResult(success=False, message=str(e), error=e)
        
        elif action_type == "input":
            field = action.get("field", "")
            value = action.get("value", "")
            try:
                locator = self._create_locator(field)
                await locator.clear()
                await locator.fill(value)
                return ActionResult(success=True, message=f"输入 {value}")
            except Exception as e:
                return ActionResult(success=False, message=str(e), error=e)
        
        elif action_type == "wait":
            duration = action.get("duration", 1000)
            await asyncio.sleep(duration / 1000)
            return ActionResult(success=True, message=f"等待 {duration}ms")
        
        elif action_type == "screenshot":
            path = f"screenshot_{int(time.time())}.png"
            await self.page.screenshot(path=path, full_page=True)
            return ActionResult(success=True, message=f"截图保存到 {path}", data=path)
        
        elif action_type == "extract":
            target = action.get("target", "")
            try:
                locator = self._create_locator(target)
                texts = await locator.all_text_contents()
                return ActionResult(success=True, message="提取成功", data=texts)
            except Exception as e:
                return ActionResult(success=False, message=str(e), error=e)
        
        return ActionResult(success=False, message=f"未知动作: {action_type}")
    
    def _create_locator(self, target: str):
        """创建元素定位器"""
        # 尝试不同的定位策略
        locators = [
            self.page.get_by_text(re.compile(target, re.I)),
            self.page.get_by_role("button", name=re.compile(target, re.I)),
            self.page.get_by_label(re.compile(target, re.I)),
            self.page.get_by_placeholder(re.compile(target, re.I)),
            self.page.locator(f"#{target}"),
            self.page.locator(f".{target}"),
            self.page.locator(f"[name='{target}']"),
            self.page.locator(target),
        ]
        
        # 返回第一个能匹配到的定位器
        for loc in locators:
            try:
                if loc.count() > 0:
                    return loc
            except:
                continue
        
        # 最后返回原始选择器
        return self.page.locator(target)
    
    async def close(self):
        """关闭 Agent"""
        if self.page:
            await self.page.close()
        if self.context:
            await self.context.close()
        if self.browser:
            await self.browser.close()
        if self.playwright:
            await self.playwright.stop()
        
        self.state = AgentState.STOPPED
        logger.info("Browser Agent 已关闭")


# 使用示例
async def main():
    """主函数"""
    agent = BrowserAgent()
    
    # 初始化
    if not await agent.initialize(BrowserConfig(headless=True)):
        print("初始化失败")
        return
    
    # 执行操作
    result = await agent.execute("打开 https://www.baidu.com")
    print(f"导航结果: {result.success}")
    
    if result.success:
        result = await agent.execute("在搜索框输入 'Playwright'")
        print(f"输入结果: {result.success}")
        
        result = await agent.execute("点击 百度一下")
        print(f"点击结果: {result.success}")
        
        await agent.execute("截图")
    
    # 关闭
    await agent.close()


if __name__ == "__main__":
    asyncio.run(main())