Python 爬虫多账号隔离：从原理到工程化实践

原创

用户12890632

发布于 2026-11-14 21:50:48

200

做需要登录的采集时，最忌讳的不是“账号多”，而是“账号之间互相串味”。多个账号共用同一个 Session、同一个代理 IP、同一套 Cookie、同一套浏览器指纹，平台很容易判定为关联账号，轻则限流，重则封号。

下面从 隔离维度、代码实现、账号池、Scrapy 场景、浏览器自动化 五个部分展开。

一、多账号隔离到底要隔离什么

隔离维度	风险点	做法
会话隔离	多个账号共用 requests.Session，Cookie 互相污染	每个账号一个独立 Session
Cookie 隔离	登录态混用、过期后无法定位	按账号持久化存储 Cookie
IP 隔离	同一 IP 登录大量账号	每个账号绑定独立代理 IP
指纹隔离	UA、时区、分辨率、浏览器缓存一致	为每个账号固定一套请求特征
行为隔离	多个账号同时高频访问	随机间隔、错峰调用、控制频率

核心原则可以概括为：

一号一 Session、一号一 Cookie、一号一代理、一号一指纹。

二、基础版：每个账号独立 Session

这是最轻量、也最实用的方式。每个账号拥有自己的 Session、代理和 Cookie 文件，避免状态串扰。

import json
import os
import random
import time
from pathlib import Path
from requests import Session

COOKIES_DIR = Path("./cookies")
COOKIES_DIR.mkdir(exist_ok=True)

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/131.0.0.0 Safari/537.36",
]


def _cookie_path(account_id: str) -> Path:
    return COOKIES_DIR / f"{account_id}.json"


def save_cookies(session: Session, account_id: str) -> None:
    """把当前 Session 的 Cookie 持久化到本地。"""
    path = _cookie_path(account_id)
    cookies = session.cookies.get_dict()
    with open(path, "w", encoding="utf-8") as f:
        json.dump(cookies, f, ensure_ascii=False, indent=2)


def load_cookies(session: Session, account_id: str) -> bool:
    """从本地加载 Cookie，不存在则返回 False。"""
    path = _cookie_path(account_id)
    if not path.exists():
        return False

    with open(path, "r", encoding="utf-8") as f:
        cookies = json.load(f)

    for name, value in cookies.items():
        session.cookies.set(name, value)
    return True


class IsolatedAccount:
    """
    账号隔离会话封装。

    每个账号拥有：
    - 独立 Session
    - 独立代理
    - 独立 Cookie 文件
    - 独立 User-Agent
    """

    def __init__(
        self,
        account_id: str,
        proxy: dict | None = None,
        user_agent: str | None = None,
        timeout: int = 15,
    ):
        self.account_id = account_id
        self.session = Session()
        self.timeout = timeout

        # 代理隔离
        if proxy:
            self.session.proxies = {
                "http": proxy.get("http") or proxy.get("https"),
                "https": proxy.get("https") or proxy.get("http"),
            }

        # 指纹隔离：每个账号固定一个 UA
        self.user_agent = user_agent or random.choice(USER_AGENTS)
        self.session.headers.update(
            {
                "User-Agent": self.user_agent,
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            }
        )

        # 恢复历史 Cookie
        load_cookies(self.session, self.account_id)

    def get(self, url: str, **kwargs) -> None | object:
        kwargs.setdefault("timeout", self.timeout)
        try:
            resp = self.session.get(url, **kwargs)
            save_cookies(self.session, self.account_id)
            return resp
        except Exception as exc:
            print(f"[{self.account_id}] 请求失败: {exc}")
            return None

    def post(self, url: str, **kwargs) -> None | object:
        kwargs.setdefault("timeout", self.timeout)
        try:
            resp = self.session.post(url, **kwargs)
            save_cookies(self.session, self.account_id)
            return resp
        except Exception as exc:
            print(f"[{self.account_id}] 请求失败: {exc}")
            return None


def demo():
    accounts = [
        {
            "account_id": "acc_01",
            "proxy": {
                "http": "http://user:pass@proxy1.example.com:8080",
                "https": "http://user:pass@proxy1.example.com:8080",
            },
        },
        {
            "account_id": "acc_02",
            "proxy": {
                "http": "http://user:pass@proxy2.example.com:8080",
                "https": "http://user:pass@proxy2.example.com:8080",
            },
        },
    ]

    clients: list[IsolatedAccount] = []
    for cfg in accounts:
        clients.append(IsolatedAccount(**cfg))

    for client in clients:
        resp = client.get("https://httpbin.org/headers")
        if resp is not None:
            print(f"[{client.account_id}] status={resp.status_code}")
        time.sleep(random.uniform(2, 5))


if __name__ == "__main__":
    demo()

这段代码的关键点：

禁止共享 Session：每个账号新建 Session()。
Cookie 按账号落盘：避免重启后重新登录。
代理按账号绑定：避免多个账号共用同一出口 IP。
UA 固定到账号：同一个账号尽量保持稳定，不要每次请求都随机换 UA，否则更像异常流量。

三、进阶版：简单账号池

当账号数量变多时，需要统一管理账号状态、可用性和轮换策略。

import random
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class AccountStatus(str, Enum):
    ACTIVE = "active"
    COOLING = "cooling"
    DISABLED = "disabled"


@dataclass
class AccountProfile:
    account_id: str
    proxy: dict
    user_agent: Optional[str] = None
    status: AccountStatus = AccountStatus.ACTIVE
    last_used_at: float = 0.0
    fail_count: int = 0
    success_count: int = 0
    client: Optional[IsolatedAccount] = None


class AccountPool:
    """
    轻量账号池。

    负责：
    - 创建隔离账号客户端
    - 选择可用账号
    - 记录成功/失败
    - 冷却异常账号
    """

    def __init__(
        self,
        max_fail: int = 3,
        cool_seconds: int = 600,
    ):
        self.max_fail = max_fail
        self.cool_seconds = cool_seconds
        self.accounts: dict[str, AccountProfile] = {}

    def add_account(self, profile: AccountProfile) -> None:
        profile.client = IsolatedAccount(
            account_id=profile.account_id,
            proxy=profile.proxy,
            user_agent=profile.user_agent,
        )
        self.accounts[profile.account_id] = profile

    def _is_available(self, profile: AccountProfile) -> bool:
        if profile.status == AccountStatus.DISABLED:
            return False

        if profile.status == AccountStatus.COOLING:
            if time.time() - profile.last_used_at < self.cool_seconds:
                return False
            profile.status = AccountStatus.ACTIVE

        return True

    def pick(self) -> Optional[IsolatedAccount]:
        available = [
            p for p in self.accounts.values() if self._is_available(p)
        ]
        if not available:
            return None

        profile = random.choice(available)
        profile.last_used_at = time.time()
        return profile.client

    def mark_success(self, account_id: str) -> None:
        profile = self.accounts.get(account_id)
        if profile:
            profile.success_count += 1
            profile.fail_count = 0
            profile.status = AccountStatus.ACTIVE

    def mark_failure(self, account_id: str) -> None:
        profile = self.accounts.get(account_id)
        if not profile:
            return

        profile.fail_count += 1

        if profile.fail_count >= self.max_fail:
            profile.status = AccountStatus.COOLING
            profile.last_used_at = time.time()


# ---------- 使用示例 ----------

def pool_demo():
    pool = AccountPool(max_fail=3, cool_seconds=300)

    pool.add_account(
        AccountProfile(
            account_id="acc_01",
            proxy={
                "http": "http://user:pass@proxy1.example.com:8080",
                "https": "http://user:pass@proxy1.example.com:8080",
            },
        )
    )
    pool.add_account(
        AccountProfile(
            account_id="acc_02",
            proxy={
                "http": "http://user:pass@proxy2.example.com:8080",
                "https": "http://user:pass@proxy2.example.com:8080",
            },
        )
    )

    url = "https://httpbin.org/headers"

    for _ in range(6):
        client = pool.pick()
        if client is None:
            print("暂无可用账号，等待冷却")
            time.sleep(5)
            continue

        resp = client.get(url)
        if resp is not None and resp.status_code == 200:
            pool.mark_success(client.account_id)
        else:
            pool.mark_failure(client.account_id)

        time.sleep(random.uniform(2, 4))

生产环境里，账号池通常会接 Redis 或数据库，用来记录账号状态、Cookie、最后登录时间、失败次数等信息。

四、Scrapy 场景：用 `cookiejar` 隔离账号

Scrapy 原生支持通过 request.meta["cookiejar"] 为不同账号维护独立 Cookie 容器，适合多账号轮换采集。

import scrapy


class MultiAccountSpider(scrapy.Spider):
    name = "multi_account"

    accounts = [
        {
            "account_id": "acc_01",
            "username": "user1",
            "password": "pass1",
            "proxy": "http://user:pass@proxy1.example.com:8080",
        },
        {
            "account_id": "acc_02",
            "username": "user2",
            "password": "pass2",
            "proxy": "http://user:pass@proxy2.example.com:8080",
        },
    ]

    login_url = "https://wlt.rubailusm.com/login"
    target_url = "https://wlt.rubailusm.com/dashboard"

    def start_requests(self):
        for idx, account in enumerate(self.accounts):
            yield scrapy.FormRequest(
                url=self.login_url,
                formdata={
                    "username": account["username"],
                    "password": account["password"],
                },
                meta={
                    "cookiejar": idx,
                    "proxy": account["proxy"],
                    "account_id": account["account_id"],
                },
                callback=self.after_login,
                dont_filter=True,
            )

    def after_login(self, response):
        account_id = response.meta["account_id"]
        proxy = response.meta["proxy"]

        # 这里可以加入登录成功校验，例如检查是否跳转到个人中心
        yield scrapy.Request(
            url=self.target_url,
            meta={
                "cookiejar": response.meta["cookiejar"],
                "proxy": proxy,
                "account_id": account_id,
            },
            callback=self.parse_data,
        )

    def parse_data(self, response):
        account_id = response.meta["account_id"]
        self.logger.info("[%s] 采集页面: %s", account_id, response.url)
        # 解析数据...

如果账号数量较多，建议把账号池放到 Redis 中，由中间件按需取出可用账号和 Cookie。

五、浏览器自动化：用独立上下文隔离

如果是 Selenium、Playwright、Pyppeteer 这类浏览器自动化场景，隔离粒度要更细。除了 Cookie，还要隔离缓存、LocalStorage、指纹和代理。

以 Playwright 为例：

from playwright.sync_api import sync_playwright

accounts = [
    {
        "account_id": "acc_01",
        "username": "user1",
        "password": "pass1",
        "proxy": {
            "server": "http://proxy1.example.com:8080",
            "username": "user",
            "password": "pass",
        },
        "user_data_dir": "./browser_profiles/acc_01",
    },
    {
        "account_id": "acc_02",
        "username": "user2",
        "password": "pass2",
        "proxy": {
            "server": "http://proxy2.example.com:8080",
            "username": "user",
            "password": "pass",
        },
        "user_data_dir": "./browser_profiles/acc_02",
    },
]


def run_account(account: dict) -> None:
    with sync_playwright() as p:
        browser = p.chromium.launch_persistent_context(
            user_data_dir=account["user_data_dir"],
            proxy=account["proxy"],
            headless=False,
        )

        page = browser.new_page()
        page.goto("https://example.com/login")

        page.fill("[name=username]", account["username"])
        page.fill("[name=password]", account["password"])
        page.click("button[type=submit]")

        page.wait_for_load_state("networkidle")
        print(f"[{account['account_id']}] 登录完成")

        browser.close()


if __name__ == "__main__":
    for acc in accounts:
        run_account(acc)

这里要注意：

每个账号使用独立 user_data_dir，避免 Cookie、缓存、LocalStorage 互相污染。
每个账号绑定独立代理。
不要在同一浏览器里只开多个 Tab 就认为已经隔离，Tab 之间仍可能共享部分状态。

六、几个容易踩坑的地方

不要全局共享 requests.Session 多线程、多账号场景下，共用 Session 会导致 Cookie 串扰。
IP 隔离不等于完全账号隔离 即使换了代理，如果 Cookie、UA、浏览器缓存、操作行为高度一致，仍可能被关联。
Cookie 需要定期校验和刷新 可以访问用户中心、订单页等需要登录的页面，判断是否失效。
账号失败后要有冷却机制 连续失败应暂停使用，避免进一步触发风控。
请求频率要控制 多账号不是无限加速的理由，仍需控制并发、间隔和总量。
遵守目标网站规则 只采集公开或已授权数据，控制频率，避免影响服务稳定。

七、推荐架构

小规模项目可以这样落地：

账号配置 → 独立 Session → 独立代理 → Cookie 持久化 → 账号池调度 → 日志与告警

大规模项目建议升级为：

Redis 账号池
  ├─ Cookie 存储
  ├─ 代理绑定
  ├─ 账号状态
  ├─ 失败计数
  └─ 冷却时间

多账号隔离本质上不是“多登录几次”，而是把 会话、网络、指纹、行为 四个层面拆开管理。做得好，能显著降低关联风险；但任何隔离都只能降低概率，不能保证完全绕过平台风控。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python