浏览器自动化

# 安装 Playwright
pip install playwright
 
# 安装浏览器驱动
playwright install chromium

from playwright.async_api import async_playwright
 
async def main():
    # 启动 Playwright
    playwright = await async_playwright().start()
 
    # 启动浏览器
    browser = await playwright.chromium.launch(
        headless=False,  # 显示浏览器窗口
        args=["--start-maximized"]
    )
 
    # 创建上下文和页面
    context = await browser.new_context(viewport=None)
    page = await context.new_page()
 
    # 导航到页面
    await page.goto("https://example.com")
 
    # 关闭
    await browser.close()
    await playwright.stop()

launch_options = {
    "headless": False,                    # 是否无头模式
    "args": [
        "--start-maximized",              # 最大化窗口
        "--disable-blink-features=AutomationControlled",  # 禁用自动化检测
        "--disable-infobars",             # 禁用信息栏
    ],
    "executable_path": "/path/to/chrome", # 自定义 Chrome 路径
    "slow_mo": 100,                       # 操作间隔（毫秒）
    "devtools": True,                     # 打开开发者工具
}
 
browser = await playwright.chromium.launch(**launch_options)

# 每个上下文独立的 Cookie、LocalStorage
context1 = await browser.new_context()
context2 = await browser.new_context()
 
page1 = await context1.new_page()
page2 = await context2.new_page()
 
# 设置上下文选项
context = await browser.new_context(
    viewport={"width": 1920, "height": 1080},
    user_agent="Mozilla/5.0 ...",
    locale="ru-RU",
    timezone_id="Europe/Moscow",
)

# 命令行启动 Chrome
chrome.exe \
  --remote-debugging-port=9222 \
  --user-data-dir=/path/to/profile \
  --no-first-run \
  --new-window \
  https://example.com

async def connect_via_cdp():
    playwright = await async_playwright().start()
 
    # 连接到已运行的 Chrome
    browser = await playwright.chromium.connect_over_cdp(
        "http://127.0.0.1:9222"
    )
 
    # 获取现有上下文和页面
    context = browser.contexts[0]
    page = context.pages[0] if context.pages else await context.new_page()
 
    return browser, page

import urllib.request
 
def is_cdp_ready(cdp_url: str) -> bool:
    """检测 CDP 接口是否可用"""
    try:
        req = urllib.request.Request(f"{cdp_url}/json/version")
        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
        with opener.open(req, timeout=0.5) as resp:
            return resp.status == 200
    except Exception:
        return False

import asyncio
import time
 
async def wait_until_ready(cdp_url: str, timeout: float = 20) -> bool:
    """等待 CDP 接口就绪"""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if is_cdp_ready(cdp_url):
            return True
        await asyncio.sleep(0.25)
    return False

# CSS 选择器
page.locator("button.submit")
page.locator("[data-testid='login']")
page.locator("input[type='email']")
 
# 文本选择器
page.locator("text=登录")
page.locator("button:has-text('提交')")
page.locator(":text('Sign in')")
 
# XPath 选择器
page.locator("//button[@type='submit']")
 
# 组合选择器
page.locator("button:has-text('登录'), a:has-text('登录')")

# 定义多个备选选择器
LOGIN_SELECTORS = [
    "[data-testid='login-button']",
    "button:has-text('登录')",
    "button:has-text('Sign in')",
    "a[href*='login']",
    "[role='button']:has-text('登录')",
]
 
async def find_and_click(page, selectors: list[str]) -> bool:
    """尝试多个选择器，找到第一个可用的"""
    for selector in selectors:
        locator = page.locator(selector).first
        try:
            if await locator.count() == 0:
                continue
            await locator.wait_for(state="visible", timeout=3000)
            await locator.click()
            return True
        except Exception:
            continue
    return False

# 按优先级排序选择器
MESSAGE_BUTTONS = [
    # 最高优先级：精确 data-testid
    "[data-testid='message-button']",
 
    # 次优先级：模糊 data-testid
    "[data-testid*='chat' i]",
 
    # 中等优先级：精确文本
    "button:has-text('Написать')",
 
    # 较低优先级：ARIA 标签
    "[aria-label*='чат' i]",
 
    # 最低优先级：URL 匹配
    "[href*='chat']",
]

# 等待元素可见
await locator.wait_for(state="visible", timeout=5000)
 
# 等待元素隐藏
await locator.wait_for(state="hidden", timeout=5000)
 
# 等待元素附加到 DOM
await locator.wait_for(state="attached", timeout=5000)
 
# 等待元素从 DOM 移除
await locator.wait_for(state="detached", timeout=5000)

async def find_button_near_input(page, input_selector: str, button_candidates: str):
    """查找输入框附近的按钮"""
    input_box = None
    input_locator = page.locator(input_selector).first
 
    if await input_locator.count() > 0:
        input_box = await input_locator.bounding_box()
 
    candidates = []
    locator = page.locator(button_candidates)
    count = await locator.count()
 
    for index in range(count):
        button = locator.nth(index)
        try:
            box = await button.bounding_box(timeout=1000)
        except Exception:
            continue
 
        if not box or box["width"] < 20 or box["height"] < 20:
            continue
 
        score = calculate_position_score(input_box, box)
        candidates.append((score, button))
 
    # 返回最高分的按钮
    if candidates:
        candidates.sort(key=lambda x: x[0], reverse=True)
        return candidates[0][1] if candidates[0][0] > 100 else None
 
    return None
 
 
def calculate_position_score(input_box: dict, button_box: dict) -> int:
    """计算按钮相对于输入框的位置分数"""
    score = 0
 
    if not input_box:
        return score
 
    # 计算中心点
    button_center_y = button_box["y"] + button_box["height"] / 2
    input_center_y = input_box["y"] + input_box["height"] / 2
 
    # 1. 水平距离：按钮在输入框右侧
    distance_right = button_box["x"] - (input_box["x"] + input_box["width"])
    if 0 <= distance_right <= 180:
        score += 150  # 理想距离
    elif button_box["x"] > input_box["x"] + input_box["width"]:
        score += 40   # 在右侧但较远
 
    # 2. 垂直对齐：中心点接近
    vertical_distance = abs(button_center_y - input_center_y)
    if vertical_distance <= 45:
        score += 120  # 完美对齐
    elif vertical_distance <= 90:
        score += 40   # 大致对齐
 
    # 3. 按钮尺寸：合适的大小
    if 28 <= button_box["width"] <= 72 and 28 <= button_box["height"] <= 72:
        score += 25
 
    # 4. 位置惩罚：按钮在输入框左侧
    if button_box["x"] + button_box["width"] < input_box["x"]:
        score -= 200
 
    return score

async def calculate_feature_score(button) -> int:
    """基于按钮特征计算分数"""
    score = 0
 
    try:
        html = await button.evaluate("(el) => el.outerHTML")
    except Exception:
        html = ""
 
    # SVG 路径特征匹配（用于识别特定图标）
    svg_patterns = {
        "send_icon": ["M5.086 3.537", "8.933-3.49", "21 11.999"],
        "attach_icon": ["M11.055 3.703", "4.565 4.556", "5.835"],
    }
 
    for icon_name, patterns in svg_patterns.items():
        if any(pattern in html for pattern in patterns):
            score += 60
            break
 
    # 文本内容匹配
    text_patterns = ["发送", "Send", "Отправить", "提交", "Submit"]
    if any(pattern in html for pattern in text_patterns):
        score += 50
 
    return score

async def find_best_button(page, input_selector: str, button_selector: str) -> bool:
    """综合评分找到最佳按钮"""
    input_box = None
    try:
        input_locator = page.locator(input_selector).first
        if await input_locator.count() > 0:
            input_box = await input_locator.bounding_box(timeout=1500)
    except Exception:
        pass
 
    candidates = []
    locator = page.locator(button_selector)
    count = await locator.count()
 
    for index in range(count):
        button = locator.nth(index)
        try:
            box = await button.bounding_box(timeout=800)
        except Exception:
            continue
 
        if not box or box["width"] < 20 or box["height"] < 20:
            continue
 
        # 位置分数
        position_score = calculate_position_score(input_box, box)
 
        # 特征分数
        feature_score = await calculate_feature_score(button)
 
        total_score = position_score + feature_score
        candidates.append((total_score, button))
 
    # 尝试点击最高分的按钮
    seen = set()
    for score, button in sorted(candidates, key=lambda x: x[0], reverse=True):
        if score < 100:
            continue
 
        try:
            handle = await button.element_handle()
            marker = str(handle)
 
            if marker in seen:
                continue
            seen.add(marker)
 
            # 检查按钮是否可用
            disabled = await button.evaluate(
                "(el) => el.disabled || el.getAttribute('aria-disabled') === 'true'"
            )
            if disabled:
                continue
 
            await button.click(timeout=3000)
            return True
        except Exception:
            continue
 
    return False

import random
import asyncio
 
async def human_pause(min_seconds: float = 0.8, max_seconds: float = 3.0):
    """模拟人类操作间隔"""
    delay = random.uniform(min_seconds, max_seconds)
    await asyncio.sleep(delay)
 
# 使用示例
await human_pause(1.5, 4.0)  # 随机等待 1.5-4 秒

async def human_scroll(page, min_pixels: int = 200, max_pixels: int = 600):
    """模拟人类滚动行为"""
    pixels = random.randint(min_pixels, max_pixels)
    await page.mouse.wheel(0, pixels)
    await asyncio.sleep(random.uniform(0.5, 1.5))
 
# 渐进式滚动
async def smooth_scroll(page, total_pixels: int):
    """平滑滚动"""
    scrolled = 0
    while scrolled < total_pixels:
        step = random.randint(50, 150)
        await page.mouse.wheel(0, step)
        scrolled += step
        await asyncio.sleep(random.uniform(0.1, 0.3))

async def human_fill(locator, text: str):
    """模拟人类逐字输入"""
    await locator.click()
    for char in text:
        await locator.type(char, delay=random.randint(30, 120))
 
# 带随机停顿的输入
async def human_type_with_pauses(locator, text: str):
    """带随机停顿的输入"""
    await locator.click()
    for i, char in enumerate(text):
        await locator.type(char, delay=random.randint(50, 150))
 
        # 每输入 5-10 个字符后随机停顿
        if i > 0 and i % random.randint(5, 10) == 0:
            await asyncio.sleep(random.uniform(0.3, 0.8))

async def human_click(page, locator):
    """模拟人类点击行为"""
    # 1. 滚动到元素可见
    await locator.scroll_into_view_if_needed()
 
    # 2. 悬停
    await locator.hover()
 
    # 3. 随机短暂停顿
    await asyncio.sleep(random.uniform(0.1, 0.3))
 
    # 4. 点击
    await locator.click()

async def human_mouse_move(page, target_x: int, target_y: int):
    """模拟人类鼠标移动轨迹"""
    # 获取当前鼠标位置（假设在屏幕中心）
    current_x, current_y = 960, 540
 
    # 生成贝塞尔曲线控制点
    steps = random.randint(10, 20)
    for i in range(steps + 1):
        t = i / steps
        # 简单的线性插值 + 随机抖动
        x = current_x + (target_x - current_x) * t + random.randint(-5, 5)
        y = current_y + (target_y - current_y) * t + random.randint(-5, 5)
 
        await page.mouse.move(x, y)
        await asyncio.sleep(random.uniform(0.01, 0.03))

async def upload_file_direct(page, file_path: str):
    """直接设置文件输入框"""
    file_input = page.locator("input[type='file']").first
 
    if await file_input.count() == 0:
        return False
 
    await file_input.set_input_files(file_path)
    return True

async def upload_via_file_chooser(page, button_selector: str, file_path: str):
    """通过点击按钮触发文件选择器"""
    button = page.locator(button_selector).first
 
    if await button.count() == 0:
        return False
 
    # 监听文件选择器事件
    async with page.expect_file_chooser(timeout=5000) as chooser_info:
        await button.click()
 
    chooser = await chooser_info.value
    await chooser.set_files(file_path)
    return True

async def upload_file_with_fallback(page, file_path: str) -> bool:
    """多策略文件上传"""
 
    # 策略1: 直接设置文件输入框
    if await upload_file_direct(page, file_path):
        return True
 
    # 策略2: 通过附件按钮
    attach_selectors = [
        "[data-testid='attach-button']",
        "button:has-text('附件')",
        "button:has-text('Attach')",
        "[aria-label*='attach' i]",
    ]
 
    for selector in attach_selectors:
        if await upload_via_file_chooser(page, selector, file_path):
            return True
 
    # 策略3: 智能定位附近按钮
    input_box = await get_input_box_position(page)
    if input_box:
        button = await find_button_near_input(
            page,
            input_selector="textarea, [contenteditable='true']",
            button_candidates="button, [role='button']"
        )
        if button:
            async with page.expect_file_chooser(timeout=3000) as chooser_info:
                await button.click()
            chooser = await chooser_info.value
            await chooser.set_files(file_path)
            return True
 
    return False

class AutomationError(Exception):
    """自动化异常基类"""
    pass
 
class ManualRequiredError(AutomationError):
    """需要人工介入的异常"""
    pass
 
class SendFailedError(AutomationError):
    """发送失败异常"""
    pass
 
class TimeoutError(AutomationError):
    """超时异常"""
    pass

async def check_page_status(page) -> str:
    """检测页面状态"""
    # 检测限流/验证码
    limit_patterns = [
        "text=/лимит|слишком часто|验证码|captcha/i",
        "[role='alert']:has-text('limit')",
    ]
 
    for pattern in limit_patterns:
        if await page.locator(pattern).count() > 0:
            return "rate_limited"
 
    # 检测 WAF/安全检查
    waf_patterns = [
        "text=/WAF|Access denied|Доступ ограничен/i",
        "text=/captcha|капча|验证码/i",
    ]
 
    for pattern in waf_patterns:
        if await page.locator(pattern).count() > 0:
            return "waf_blocked"
 
    # 检测错误提示
    error_patterns = [
        ".toast-error",
        ".error-message",
        "[role='alert']",
    ]
 
    for pattern in error_patterns:
        if await page.locator(pattern).count() > 0:
            return "error"
 
    return "normal"

async def retry_on_failure(func, max_retries: int = 3, delay: float = 1.0):
    """失败自动重试"""
    last_exception = None
 
    for attempt in range(max_retries):
        try:
            return await func()
        except AutomationError as e:
            last_exception = e
            if attempt < max_retries - 1:
                await asyncio.sleep(delay * (attempt + 1))
 
    raise last_exception
 
 
# 使用示例
result = await retry_on_failure(
    lambda: send_message(page, message),
    max_retries=3,
    delay=2.0
)

from datetime import datetime
from pathlib import Path
 
async def save_failure_artifacts(page, account_id: str, reason: str) -> tuple[Path, Path]:
    """保存失败截图和页面 HTML"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    safe_reason = "".join(c if c.isalnum() else "_" for c in reason[:24])
 
    screenshot_path = Path(f"screenshots/{timestamp}_{account_id}_{safe_reason}.png")
    html_path = Path(f"html/{timestamp}_{account_id}_{safe_reason}.html")
 
    # 确保目录存在
    screenshot_path.parent.mkdir(parents=True, exist_ok=True)
    html_path.parent.mkdir(parents=True, exist_ok=True)
 
    try:
        await page.screenshot(path=str(screenshot_path), full_page=True)
    except Exception as e:
        print(f"截图保存失败: {e}")
 
    try:
        html_path.write_text(await page.content(), encoding="utf-8")
    except Exception as e:
        print(f"HTML 保存失败: {e}")
 
    return screenshot_path, html_path

async def handle_manual_required(page, reason: str):
    """处理需要人工介入的情况"""
    # 1. 保存当前状态
    screenshot_path, html_path = await save_failure_artifacts(
        page, "current", reason
    )
 
    # 2. 暂停自动化
    print(f"需要人工介入: {reason}")
    print(f"截图已保存: {screenshot_path}")
    print(f"HTML 已保存: {html_path}")
 
    # 3. 等待人工处理
    # 可以通过 WebSocket 通知前端，或简单等待
    await wait_for_human_intervention()
 
    # 4. 恢复自动化
    print("人工处理完成，继续自动化...")

from pydantic_settings import BaseSettings
from typing import Optional
from pathlib import Path
 
class BrowserConfig(BaseSettings):
    """浏览器配置"""
 
    # 浏览器模式
    browser_mode: str = "cdp"  # "playwright" | "cdp" | "custom"
 
    # Chrome 配置
    chrome_executable_path: Optional[str] = None
    chrome_cdp_url: str = "http://127.0.0.1:9222"
    chrome_user_data_dir: str = ".chrome-profile"
 
    # 浏览器选项
    headless: bool = False
    page_timeout_ms: int = 30000
 
    # 拟人化配置
    min_delay: float = 0.8
    max_delay: float = 3.0
 
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# .env 文件
BROWSER_MODE=cdp
CHROME_EXECUTABLE_PATH=C:\Program Files\Google\Chrome\Application\chrome.exe
CHROME_CDP_URL=http://127.0.0.1:9222
HEADLESS=false
PAGE_TIMEOUT_MS=30000

from pathlib import Path
 
def find_system_browser() -> Optional[str]:
    """自动检测系统安装的浏览器"""
    candidates = [
        # Windows
        Path("C:/Program Files/Google/Chrome/Application/chrome.exe"),
        Path("C:/Program Files (x86)/Google/Chrome/Application/chrome.exe"),
        Path.home() / "AppData/Local/Google/Chrome/Application/chrome.exe",
        Path("C:/Program Files/Microsoft/Edge/Application/msedge.exe"),
 
        # macOS
        Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"),
        Path("/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge"),
 
        # Linux
        Path("/usr/bin/google-chrome"),
        Path("/usr/bin/chromium-browser"),
        Path("/usr/bin/microsoft-edge"),
    ]
 
    for path in candidates:
        if path.exists():
            return str(path)
 
    return None

from loguru import logger
import sys
 
# 添加控制台输出
logger.add(sys.stderr, level="DEBUG")
 
# 添加文件输出
logger.add(
    "debug.log",
    level="DEBUG",
    rotation="10 MB",
    retention="7 days"
)

# 设置环境变量启用调试模式
export PWDEBUG=1  # Linux/macOS
set PWDEBUG=1     # Windows
 
# 运行脚本
python your_script.py

# 查看 CDP 状态
import requests
 
def check_cdp_status(cdp_url: str):
    """检查 CDP 连接状态"""
    try:
        response = requests.get(f"{cdp_url}/json/version", timeout=1)
        if response.status_code == 200:
            data = response.json()
            print(f"Browser: {data.get('Browser')}")
            print(f"Protocol: {data.get('Protocol-Version')}")
            print(f"User-Agent: {data.get('User-Agent')}")
            return True
    except Exception as e:
        print(f"CDP 连接失败: {e}")
 
    return False

# 1. 复用浏览器实例
browser = await playwright.chromium.launch()
context = await browser.new_context()
 
# 多个任务共享同一个浏览器
pages = [await context.new_page() for _ in range(5)]
 
# 2. 并行操作
import asyncio
 
async def process_page(page, url):
    await page.goto(url)
    # 处理页面...
    return result
 
# 并行处理多个页面
results = await asyncio.gather(*[
    process_page(page, url)
    for page, url in zip(pages, urls)
])
 
# 3. 资源限制
context = await browser.new_context(
    # 禁用图片加载
    extra_http_headers={"Accept": "text/html"}
)

# 导航
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
await page.reload()
await page.go_back()
await page.go_forward()
 
# 等待
await page.wait_for_load_state("networkidle")
await page.wait_for_url("**/dashboard")
await page.wait_for_selector(".loading", state="hidden")
await page.wait_for_timeout(1000)
 
# 执行 JavaScript
result = await page.evaluate("() => document.title")
result = await page.evaluate("(arg) => arg * 2", 21)

# 定位
locator = page.locator("selector")
locator = page.locator("selector").first
locator = page.locator("selector").nth(2)
 
# 状态检查
count = await locator.count()
visible = await locator.is_visible()
enabled = await locator.is_enabled()
 
# 内容获取
text = await locator.inner_text()
html = await locator.inner_html()
value = await locator.input_value()
 
# 操作
await locator.click()
await locator.fill("text")
await locator.type("text", delay=100)
await locator.select_option("value")
await locator.check()
await locator.uncheck()
 
# 位置信息
box = await locator.bounding_box()
await locator.scroll_into_view_if_needed()

await page.keyboard.press("Enter")
await page.keyboard.press("Control+A")
await page.keyboard.press("Control+C")
await page.keyboard.press("Control+V")
await page.keyboard.press("Backspace")
await page.keyboard.press("Escape")
 
# 组合键
await page.keyboard.press("Shift+Tab")
await page.keyboard.press("Alt+F4")

await page.mouse.click(x, y)
await page.mouse.dblclick(x, y)
await page.mouse.right_click(x, y)
await page.mouse.move(x, y)
await page.mouse.down()
await page.mouse.up()
await page.mouse.wheel(0, 100)  # 垂直滚动