实战案例库

本章提供五个可直接套用到真实项目中的完整脚本模板，覆盖多账号自动化的最常见场景。每个案例提供 JavaScript 和 Python 两个版本，按需取用。

---

案例一：自动登录并保持状态

场景：自动将指定的候鸟账号环境登录到目标平台。

// JavaScript 版
const { chromium } = require('playwright');
const axios = require('axios');

const API_URL    = 'http://127.0.0.1:8186';
const SESSION_ID = '替换为您的环境ID';

async function autoLogin() {
  // ── 开启候鸟环境 ─────────────────────────────
  const { data } = await axios.post(`${API_URL}/api/v1/browser/start`, {
    Session_ID: SESSION_ID
  });
  if (data.code !== 0) throw new Error(`环境启动失败: ${data.message}`);
  const wsEndpoint = data.data.ws;

  const browser = await chromium.connectOverCDP(wsEndpoint);
  const context = browser.contexts()[0];
  const page    = context.pages()[0] || await context.newPage();

  try {
    // ── 打开登录页并填表 ────────────────────────
    await page.goto('https://example.com/login', { waitUntil: 'domcontentloaded' });

    // getByRole 是 Playwright 最推荐的定位方式，比 CSS 选择器更稳
    await page.getByLabel('邮箱').fill('my_account@email.com');
    await page.getByLabel('密码').fill('my_password_123');
    await page.getByRole('button', { name: '登 录' }).click();

    // 等待跳转到首页（登录成功的标志）
    await page.waitForURL('**/dashboard', { timeout: 15000 });
    console.log('✅ 登录成功！当前 URL:', page.url());

    // 候鸟会自动持久化本次 Cookie，下次开启该环境无需重新登录
    console.log('🔒 Cookie 已自动保存到候鸟环境');
  } catch (e) {
    console.error('❌ 登录失败:', e.message);
    await page.screenshot({ path: 'login_error.png', fullPage: true });
  } finally {
    await browser.close();
  }
}

autoLogin().catch(console.error);

# Python 版
import requests
from playwright.sync_api import sync_playwright

API_URL    = "http://127.0.0.1:8186"
SESSION_ID = "替换为您的环境ID"

def auto_login():
    resp = requests.post(
        f"{API_URL}/api/v1/browser/start",
        json={"Session_ID": SESSION_ID}, timeout=30
    )
    data = resp.json()
    if data["code"] != 0:
        print(f"❌ 环境启动失败: {data['message']}")
        return

    ws = data["data"]["ws"]
    with sync_playwright() as p:
        browser = p.chromium.connect_over_cdp(ws)
        context = browser.contexts[0]
        page    = context.pages[0] if context.pages else context.new_page()

        try:
            page.goto("https://example.com/login", wait_until="domcontentloaded")
            page.get_by_label("邮箱").fill("my_account@email.com")
            page.get_by_label("密码").fill("my_password_123")
            page.get_by_role("button", name="登 录").click()
            page.wait_for_url("**/dashboard", timeout=15000)
            print(f"✅ 登录成功！当前 URL: {page.url}")
        except Exception as e:
            print(f"❌ 登录失败: {e}")
            page.screenshot(path="login_error.png", full_page=True)
        finally:
            browser.close()

if __name__ == "__main__":
    auto_login()

---

案例二：Cookie 注入直接进入登录态

场景：已有账号 Cookie 数据，无需走登录表单，直接注入后即进入登录状态。

// JavaScript 版
async function injectAndLogin(page, context) {
  // 注意：必须先打开目标域名，才能给该域注入 Cookie
  await page.goto('https://shop.example.com', { waitUntil: 'commit' });

  // 注入准备好的 Cookie 数组
  await context.addCookies([
    { name: 'session_id', value: 'SID_abcdef123456', domain: '.example.com', path: '/' },
    { name: 'auth_token', value: 'Bearer_xyz789',   domain: '.example.com', path: '/' }
  ]);

  // 刷新页面，Cookie 生效
  await page.reload({ waitUntil: 'domcontentloaded' });

  // 验证是否成功进入登录状态
  const accountEl = page.getByRole('button', { name: '我的账号' });
  if (await accountEl.isVisible()) {
    console.log('✅ Cookie 注入成功，已进入登录状态！');
  } else {
    console.warn('⚠️ Cookie 可能已过期，请重新采集');
  }
}

# Python 版
def inject_and_login(page, context):
    page.goto("https://shop.example.com", wait_until="commit")
    context.add_cookies([
        {"name": "session_id", "value": "SID_abcdef123456",
         "domain": ".example.com", "path": "/"},
        {"name": "auth_token", "value": "Bearer_xyz789",
         "domain": ".example.com", "path": "/"}
    ])
    page.reload(wait_until="domcontentloaded")

    if page.get_by_role("button", name="我的账号").is_visible():
        print("✅ Cookie 注入成功！")
    else:
        print("⚠️ Cookie 可能已过期")

NOTE

候鸟小秘密：在候鸟环境中成功登录过一次后，Cookie 会永久保存在该环境的数据目录中。下次重新开启该环境，无需重复注入，账号仍处于登录状态——这就是指纹浏览器"持久化环境"的核心价值！

---

案例三：利用网络拦截抓取 API 数据

场景：目标网站的数据通过 XHR/Fetch 请求加载，比直接解析网页 HTML 更方便、更稳定。

// JavaScript 版
async function interceptAndCollect(page) {
  const collectedData = [];

  // ── 监听目标 API 的响应 ──────────────────────────
  page.on('response', async response => {
    if (response.url().includes('/api/products') && response.status() === 200) {
      try {
        const json = await response.json();
        if (json.data?.list) {
          collectedData.push(...json.data.list);
          console.log(`📦 已采集 ${collectedData.length} 条商品数据`);
        }
      } catch (e) {
        // 部分响应不是 JSON，忽略
      }
    }
  });

  // ── 打开商品列表页，自动触发 API 请求 ───────────
  await page.goto('https://example.com/products');

  // ── 模拟翻页，采集多页数据 ──────────────────────
  for (let i = 0; i < 5; i++) {
    const nextBtn = page.getByRole('button', { name: '下一页' });
    if (!await nextBtn.isVisible()) break;
    // 点击翻页，并等待数据请求完成
    await Promise.all([
      page.waitForResponse(r => r.url().includes('/api/products')),
      nextBtn.click()
    ]);
  }

  console.log(`✅ 共采集 ${collectedData.length} 条数据`);
  return collectedData;
}

# Python 版（Python Playwright 默认是同步 API）
def intercept_and_collect(page):
    collected_data = []

    def handle_response(response):
        if "/api/products" in response.url and response.status == 200:
            try:
                json_data = response.json()
                if json_data.get("data", {}).get("list"):
                    collected_data.extend(json_data["data"]["list"])
                    print(f"📦 已采集 {len(collected_data)} 条数据")
            except Exception:
                pass

    page.on("response", handle_response)
    page.goto("https://example.com/products")

    for _ in range(5):
        next_btn = page.get_by_role("button", name="下一页")
        if not next_btn.is_visible():
            break
        with page.expect_response(lambda r: "/api/products" in r.url):
            next_btn.click()

    print(f"✅ 共采集 {len(collected_data)} 条数据")
    return collected_data

---

案例四：多账号并发批量操作

场景：同时操控多个候鸟环境，实现真正的并行效率（JavaScript Promise.all 天然支持）。

// JavaScript 版 —— 并发操作多个账号
const { chromium } = require('playwright');
const axios = require('axios');

const API_URL = 'http://127.0.0.1:8186';

// 定义要并发操作的环境列表
const SESSION_LIST = [
  '环境ID_1',
  '环境ID_2',
  '环境ID_3',
  '环境ID_4',
  '环境ID_5',
];

async function operateOneAccount(sessionId, index) {
  console.log(`[线程 ${index}] 启动环境 ${sessionId.slice(0, 8)}...`);
  try {
    const { data } = await axios.post(`${API_URL}/api/v1/browser/start`, {
      Session_ID: sessionId
    });
    if (data.code !== 0) {
      console.error(`[线程 ${index}] ❌ 启动失败: ${data.message}`);
      return;
    }

    const browser = await chromium.connectOverCDP(data.data.ws);
    const context = browser.contexts()[0];
    const page    = context.pages()[0] || await context.newPage();

    // ── 在此处放您的账号业务逻辑 ────────────────
    await page.goto('https://example.com');
    console.log(`[线程 ${index}] ✅ 操作完成: ${await page.title()}`);

    await browser.close();
  } catch (e) {
    console.error(`[线程 ${index}] ❌ 异常: ${e.message}`);
  } finally {
    // 关闭候鸟环境（释放内存）
    await axios.post(`${API_URL}/api/v1/browser/stop`, { Session_ID: sessionId })
      .catch(() => {});
  }
}

async function batchOperate() {
  // Promise.all 同时启动所有账号操作
  await Promise.all(
    SESSION_LIST.map((sessionId, i) => operateOneAccount(sessionId, i + 1))
  );
  console.log('\n🎉 所有账号操作完毕！');
}

batchOperate().catch(console.error);

# Python 版 —— 使用 threading 实现并发
import threading
import requests
from playwright.sync_api import sync_playwright

API_URL      = "http://127.0.0.1:8186"
SESSION_LIST = ["环境ID_1", "环境ID_2", "环境ID_3", "环境ID_4", "环境ID_5"]


def operate_one_account(session_id, thread_id):
    print(f"[线程 {thread_id}] 启动环境 {session_id[:8]}...")
    try:
        resp = requests.post(
            f"{API_URL}/api/v1/browser/start",
            json={"Session_ID": session_id}, timeout=30
        )
        data = resp.json()
        if data["code"] != 0:
            print(f"[线程 {thread_id}] ❌ 失败: {data['message']}")
            return

        ws = data["data"]["ws"]
        with sync_playwright() as p:
            browser = p.chromium.connect_over_cdp(ws)
            context = browser.contexts[0]
            page    = context.pages[0] if context.pages else context.new_page()

            # ── 您的账号业务逻辑 ─────────────────
            page.goto("https://example.com")
            print(f"[线程 {thread_id}] ✅ 完成: {page.title()}")

            browser.close()
    except Exception as e:
        print(f"[线程 {thread_id}] ❌ 异常: {e}")
    finally:
        requests.post(
            f"{API_URL}/api/v1/browser/stop",
            json={"Session_ID": session_id}, timeout=10
        )


def batch_operate():
    threads = [
        threading.Thread(target=operate_one_account, args=(sid, i + 1))
        for i, sid in enumerate(SESSION_LIST)
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print("\n🎉 所有账号操作完毕！")


if __name__ == "__main__":
    batch_operate()

IMPORTANT

并发数量建议：

• 16GB 内存 → 同时开启 10-15 个候鸟环境
• 32GB 内存 → 同时开启 20-30 个候鸟环境
  过多并发会导致内存不足，浏览器响应迟钝甚至崩溃，请根据实际硬件配置控制并发量。

---

案例五：生产级脚本框架（含重试与日志）

场景：正式生产环境使用，需要稳定性、可观测性和容错能力。

# Python 生产级框架
import threading
import requests
import logging
import time
from playwright.sync_api import sync_playwright, TimeoutError as PWTimeoutError

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(threadName)s] %(levelname)s %(message)s",
    datefmt="%H:%M:%S"
)
logger = logging.getLogger(__name__)

API_URL = "http://127.0.0.1:8186"


class MBPlaywrightSession:
    """候鸟 Playwright 会话管理器（支持自动重试、上下文管理）"""

    def __init__(self, session_id: str, max_retry: int = 3):
        self.session_id = session_id
        self.max_retry  = max_retry
        self._pw        = None
        self._browser   = None

    def start(self):
        """启动候鸟环境并初始化 Playwright 连接（失败自动重试）"""
        for attempt in range(1, self.max_retry + 1):
            try:
                logger.info(f"[{self.session_id[:8]}] 尝试开启环境（第 {attempt} 次）")
                resp = requests.post(
                    f"{API_URL}/api/v1/browser/start",
                    json={"Session_ID": self.session_id},
                    timeout=30
                )
                data = resp.json()
                if data["code"] == 0:
                    ws = data["data"]["ws"]
                    self._pw      = sync_playwright().start()
                    self._browser = self._pw.chromium.connect_over_cdp(ws)
                    logger.info(f"[{self.session_id[:8]}] ✅ 环境就绪")
                    return self
                logger.warning(f"[{self.session_id[:8]}] 开启失败: {data.get('message')}")
            except Exception as e:
                logger.error(f"[{self.session_id[:8]}] 异常: {e}")
            time.sleep(2 ** attempt)  # 指数退避：2s、4s、8s

        raise RuntimeError(f"环境 {self.session_id} 连续 {self.max_retry} 次启动失败")

    @property
    def page(self):
        """返回当前上下文的第一个页面（懒加载）"""
        context = self._browser.contexts[0]
        return context.pages[0] if context.pages else context.new_page()

    @property
    def context(self):
        return self._browser.contexts[0]

    def stop(self, close_browser: bool = False):
        """关闭会话"""
        try:
            if self._browser:
                self._browser.close()
            if self._pw:
                self._pw.stop()
        except Exception:
            pass
        if close_browser:
            requests.post(
                f"{API_URL}/api/v1/browser/stop",
                json={"Session_ID": self.session_id},
                timeout=10
            )
        logger.info(f"[{self.session_id[:8]}] 会话已关闭")

    def __enter__(self):
        return self.start()

    def __exit__(self, *args):
        self.stop()
        return False


# ── 使用示例 ──────────────────────────────────────────────
SESSION_ID = "替换为您的环境ID"

with MBPlaywrightSession(SESSION_ID) as session:
    page = session.page
    try:
        page.goto("https://www.mbbrowser.com")
        logger.info(f"页面标题: {page.title()}")
        page.screenshot(path="result.png")
    except PWTimeoutError:
        logger.error("页面超时，请检查代理或网络")
    except Exception as e:
        logger.error(f"执行异常: {e}")
        page.screenshot(path="error.png")

---

与 Selenium / Puppeteer 的对比总结

对比维度	Puppeteer	Selenium	Playwright
候鸟接入字段	data.ws	data.http	data.ws
连接方法	connectOverCDP(ws)	debuggerAddress	connectOverCDP(ws)
主要语言	JavaScript	Python/Java	JS / Python / Java
自动等待	手写 waitForSelector	手写 WebDriverWait	内置，默认开启
网络拦截	有限	不支持	强大（route/intercept）
多标签页	支持	支持	原生优秀
推荐场景	快速脚本/JS 开发者	已有代码库/Python 爬虫	全场景，新项目首选

---

TIP

🎉 恭喜！您已掌握候鸟 + Playwright 的完整自动化技术栈。

如需深入了解 Playwright 官方文档，请访问 playwright.dev。

如果您想在候鸟客户端内管理和调试您的脚本，请参阅 自动化脚本管理器使用。