实战案例库

本章提供四个可以直接套用到真实项目中的业务脚本模板，覆盖最常见的多账号自动化场景。

---

案例一：自动填写账号密码并登录

适用场景：自动登录各类平台（电商后台、社媒、广告账户等）

import requests
import time
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

API_URL     = "http://127.0.0.1:8186"
SESSION_ID  = "替换为您的环境ID"
DRIVER_PATH = r"C:\Program Files (x86)\Mbbrowser_v7.10.20.219\chromedriver.exe"


def get_driver_from_mbbrowser(session_id: str) -> webdriver.Chrome | None:
    """公共函数：开启候鸟环境并返回 Selenium driver"""
    resp = requests.post(
        f"{API_URL}/api/v1/browser/start",
        json={"Session_ID": session_id},
        timeout=30
    )
    data = resp.json()
    if data.get("code") != 0:
        print(f"❌ 环境开启失败: {data.get('message')}")
        return None

    debugger_address = data["data"]["http"]
    chrome_options = Options()
    chrome_options.add_experimental_option("debuggerAddress", debugger_address)
    service = Service(executable_path=DRIVER_PATH)
    driver = webdriver.Chrome(service=service, options=chrome_options)
    return driver


def auto_login():
    driver = get_driver_from_mbbrowser(SESSION_ID)
    if not driver:
        return

    wait = WebDriverWait(driver, 20)

    try:
        # ── 1. 打开登录页 ──────────────────────────────────
        driver.get("https://example.com/login")

        # ── 2. 等待账号输入框出现，然后输入账号 ───────────
        username_input = wait.until(
            EC.visibility_of_element_located((By.ID, "username"))
        )
        username_input.clear()
        username_input.send_keys("my_account@email.com")
        time.sleep(0.5)  # 短暂停顿，模拟人的节奏

        # ── 3. 输入密码 ────────────────────────────────────
        password_input = driver.find_element(By.ID, "password")
        password_input.clear()
        password_input.send_keys("my_password_123")
        time.sleep(0.3)

        # ── 4. 点击登录按钮 ────────────────────────────────
        login_btn = wait.until(
            EC.element_to_be_clickable((By.CSS_SELECTOR, "button[type='submit']"))
        )
        login_btn.click()

        # ── 5. 等待登录成功的标志（URL 跳转到 dashboard）──
        wait.until(EC.url_contains("/dashboard"))
        print(f"✅ 登录成功！当前页面：{driver.current_url}")

        # ── 6. 登录成功后，候鸟会自动保存本次 Cookie ──────
        # 下次手动或脚本开启该环境，仍然是登录状态
        print("🔒 Cookie 已自动保存到候鸟环境，下次无需重新登录")

    except Exception as e:
        print(f"❌ 登录出现异常: {e}")
        driver.save_screenshot("login_error.png")  # 出错时截图存档
    finally:
        driver.quit()


if __name__ == "__main__":
    auto_login()

---

案例二：注入 Cookie 直接进入登录状态（跳过登录步骤）

适用场景：您已经有了账号的 Cookie 数据（从浏览器导出，或之前登录已保存），想直接跳过登录页面进入账号内部。

def inject_cookies_and_enter():
    driver = get_driver_from_mbbrowser(SESSION_ID)
    if not driver:
        return

    # ── 1. 必须先打开目标域名，才能设置该域的 Cookie ────
    driver.get("https://www.amazon.com")

    # ── 2. 准备您的 Cookie 数据 ───────────────────────
    cookies = [
        {
            "name": "session-id",
            "value": "123-4567890-1234567",
            "domain": ".amazon.com",
            "path": "/",
            "secure": True
        },
        {
            "name": "ubid-main",
            "value": "130-1234567-7654321",
            "domain": ".amazon.com",
            "path": "/"
        }
    ]

    # ── 3. 逐一注入 Cookie ────────────────────────────
    driver.delete_all_cookies()  # 先清空旧 Cookie（可选）
    for cookie in cookies:
        driver.add_cookie(cookie)
    print(f"✅ 已注入 {len(cookies)} 个 Cookie")

    # ── 4. 刷新页面，让 Cookie 生效 ──────────────────
    driver.refresh()
    time.sleep(2)

    # ── 5. 验证是否登录成功 ──────────────────────────
    if "Hello" in driver.page_source or "Account" in driver.title:
        print("✅ Cookie 注入成功，已进入登录状态！")
    else:
        print("⚠️ Cookie 可能已过期，请重新获取")

    driver.quit()

NOTE

候鸟小秘密：候鸟本身会自动持久化每个环境的 Cookie。每次在候鸟环境中完成登录，Cookie 都会被自动保存。下次开启该环境时无需重新登录，直接就是上次的登录状态。

---

案例三：批量抓取数据（商品列表）

适用场景：登录状态下抓取受保护的页面数据，如受众分析报告、竞品价格等。

def scrape_product_list():
    driver = get_driver_from_mbbrowser(SESSION_ID)
    if not driver:
        return

    wait = WebDriverWait(driver, 20)
    all_products = []

    try:
        # 打开商品列表页
        driver.get("https://example.com/products?page=1")

        # ── 翻页抓取逻辑 ──────────────────────────────────
        page = 1
        while True:
            print(f"📄 正在抓取第 {page} 页...")

            # 等待商品列表容器加载完成
            wait.until(EC.presence_of_element_located(
                (By.CSS_SELECTOR, ".product-list")
            ))

            # 抓取本页所有商品
            items = driver.find_elements(By.CSS_SELECTOR, ".product-item")
            for item in items:
                try:
                    name  = item.find_element(By.CSS_SELECTOR, ".name").text.strip()
                    price = item.find_element(By.CSS_SELECTOR, ".price").text.strip()
                    all_products.append({"name": name, "price": price, "page": page})
                except Exception:
                    pass  # 跳过解析失败的单个商品

            print(f"  本页抓取 {len(items)} 条，累计 {len(all_products)} 条")

            # 查找"下一页"按钮，不存在则停止
            next_btns = driver.find_elements(
                By.CSS_SELECTOR, "a.pagination-next:not([aria-disabled='true'])"
            )
            if not next_btns:
                print("✅ 已到最后一页，抓取完成！")
                break

            next_btns[0].click()
            page += 1
            time.sleep(1.5)  # 翻页后稍作等待，避免频率过高

    finally:
        driver.quit()

    # 输出结果（也可以写入 CSV 或数据库）
    print(f"\n共抓取到 {len(all_products)} 个商品：")
    for p in all_products[:5]:
        print(f"  {p['name']} - {p['price']}")
    return all_products

---

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

适用场景：同时操作多个候鸟环境，实现真正的并行效率。

import threading
import requests
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service

API_URL     = "http://127.0.0.1:8186"
DRIVER_PATH = r"C:\Program Files (x86)\Mbbrowser_v7.10.20.219\chromedriver.exe"

# 定义多个要并发操作的环境 ID
SESSION_LIST = [
    "环境ID_1",
    "环境ID_2",
    "环境ID_3",
    "环境ID_4",
    "环境ID_5",
]


def operate_one_account(session_id: str, thread_id: int):
    """单个账号的操作逻辑（每个线程执行一次）"""
    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.get("code") != 0:
            print(f"[线程 {thread_id}] ❌ 失败: {data.get('message')}")
            return

        debugger_address = data["data"]["http"]
        chrome_options = Options()
        chrome_options.add_experimental_option("debuggerAddress", debugger_address)
        service = Service(executable_path=DRIVER_PATH)
        driver = webdriver.Chrome(service=service, options=chrome_options)

        # ── 在这里写您的账号操作逻辑 ──
        driver.get("https://example.com")
        print(f"[线程 {thread_id}] ✅ 操作完成：{driver.title}")

        driver.quit()
    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 = []
    for i, session_id in enumerate(SESSION_LIST):
        t = threading.Thread(
            target=operate_one_account,
            args=(session_id, i + 1)
        )
        threads.append(t)
        t.start()

    # 等待所有线程完成
    for t in threads:
        t.join()

    print("\n🎉 所有账号操作完毕！")


if __name__ == "__main__":
    batch_operate()

IMPORTANT

并发数量建议：每个候鸟环境需要消耗一定内存（约 300-600MB）。建议根据您的机器配置控制并发数量：

• 16GB 内存 → 同时开 10-15 个环境
• 32GB 内存 → 同时开 20-30 个环境

过多并发会导致系统内存不足，浏览器响应变慢甚至崩溃。

---

案例五：完整的生产级脚本框架

以下是一个更贴近真实生产环境的脚本框架，包含完整错误处理、重试机制和日志：

import requests
import time
import logging
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.common.exceptions import TimeoutException, WebDriverException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By

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

API_URL     = "http://127.0.0.1:8186"
DRIVER_PATH = r"C:\Program Files (x86)\Mbbrowser_v7.10.20.219\chromedriver.exe"


class MBBrowserSession:
    """候鸟浏览器 Selenium 会话管理器（支持重试、上下文管理）"""

    def __init__(self, session_id: str, retry: int = 3):
        self.session_id = session_id
        self.retry = retry
        self.driver = None

    def start(self) -> webdriver.Chrome:
        """开启环境并返回 driver（失败自动重试）"""
        for attempt in range(1, self.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.get("code") == 0:
                    debugger_address = data["data"]["http"]
                    chrome_options = Options()
                    chrome_options.add_experimental_option("debuggerAddress", debugger_address)
                    service = Service(executable_path=DRIVER_PATH)
                    self.driver = webdriver.Chrome(service=service, options=chrome_options)
                    logger.info(f"[{self.session_id[:8]}] 环境开启成功 ✅")
                    return self.driver
                else:
                    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)  # 指数退避重试

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

    def stop(self):
        """关闭候鸟环境"""
        if self.driver:
            try:
                self.driver.quit()
            except Exception:
                pass
        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, exc_type, exc_val, exc_tb):
        self.stop()
        return False  # 不吞掉异常


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

with MBBrowserSession(SESSION_ID) as driver:
    wait = WebDriverWait(driver, 15)
    try:
        driver.get("https://www.mbbrowser.com")
        logger.info(f"当前页面标题: {driver.title}")
        driver.save_screenshot("result.png")
    except TimeoutException:
        logger.error("页面加载超时，请检查网络或代理配置")
    except WebDriverException as e:
        logger.error(f"WebDriver 出错: {e}")

---

最佳实践小结

实践建议	说明
✅ 总是用显式等待	用 WebDriverWait 替代 time.sleep()，更稳定
✅ 出错时截图	在 except 中调用 driver.save_screenshot()，方便复盘
✅ 用 with 语句管理资源	配合上下文管理器，确保浏览器一定会被关闭
✅ 控制操作节奏	关键操作之间加 time.sleep(0.3~1.0)，模拟自然人节奏
✅ 按环境日志区分	多线程时每条日志打印环境ID，方便溯源
❌ 不要在循环里硬 sleep	会大幅降低脚本效率，用等待条件代替
❌ 不要共用一个 driver	每个线程/每个账号要开独立的 Session_ID 和 driver

---

TIP

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

如果遇到问题，可以回顾 核心 API 详解 查阅具体 API 用法，或参考 自动化脚本管理器使用 了解如何在候鸟客户端内管理和调试您的 Python 脚本。