多账号管理¶

概述¶

igapi 采用实例隔离设计：每个 Client 或 WebClient 对象持有独立的 HTTP 客户端、Cookie 容器和 Session 状态，不同实例之间完全不共享任何状态。这意味着：

可以在同一进程中同时持有任意数量的账号客户端
一个账号的登录状态变化不会影响其他账号
每个实例可以配置不同的代理，实现流量分散
同一账号可以有多个实例，但各自的 Session 变更不会同步

此特性使得多账号批量管理成为可能，典型场景包括：内容矩阵运营、自动化数据采集、账号健康状态批量检测等。

快速开始¶

以下示例演示最简单的多账号加载方式——从文件中读取已保存的 AccountInfo 字符串，批量创建客户端：

import asyncio
import igapi

async def main():
    # 从文件批量加载账号
    clients = {}

    with open("accounts.txt", "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue  # 跳过空行和注释行

            account = igapi.AccountInfo.parse(line, platform="android")
            client = igapi.Client(account=account)
            clients[account.username] = client

    print(f"已加载 {len(clients)} 个账号")

    # 使用指定账号
    client = clients["目标用户名"]
    user = await client.user().info(user_id=25025320)
    print(f"@{user.username}: {user.follower_count} 粉丝")

asyncio.run(main())

账号文件格式¶

账号文件（accounts.txt）每行存储一个账号的完整序列化字符串，通过 client.export_account_string() 生成。

格式规范¶

用户名:密码||android_id;phone_id;uuid;device_id|sessionid=xxx;ds_user_id=yyy;csrftoken=zzz;mid=aaa;rur=bbb||

文件示例¶

# 账号文件示例（# 开头为注释行）
# 每行一个账号，由 export_account_string() 生成

alice:pass123||abc123;def456;ghi789;jkl012|sessionid=abc%3D;ds_user_id=111;csrftoken=xyz;mid=mid1;rur=rur1||
bob:secret456||mno345;pqr678;stu901;vwx234|sessionid=def%3D;ds_user_id=222;csrftoken=uvw;mid=mid2;rur=rur2||

注意事项¶

每行格式由库内部定义，不应手动编辑，始终通过 export_account_string() 生成
# 开头的行作为注释，加载时需手动跳过
文件应使用 UTF-8 编码保存
sessionid 中的特殊字符已进行 URL 编码，无需手动处理

批量操作示例¶

以下示例展示完整的生命周期：加载账号 → 登录（若未登录）→ 执行操作 → 保存 Session。

import asyncio
import igapi
import time

ACCOUNTS_FILE = "accounts.txt"
CREDENTIALS_FILE = "credentials.txt"  # 仅含 用户名:密码 的文件，用于首次登录


async def load_or_login(credentials_file: str, accounts_file: str) -> dict:
    """
    优先从 accounts.txt 恢复 Session，
    若账号不存在则通过 credentials.txt 登录并保存。
    """
    # 读取已保存的 Session
    saved_sessions = {}
    try:
        with open(accounts_file, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                account = igapi.AccountInfo.parse(line, platform="android")
                saved_sessions[account.username] = line
    except FileNotFoundError:
        pass

    clients = {}

    # 读取登录凭据
    with open(credentials_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            username, password = line.split(":", 1)

            if username in saved_sessions:
                # 直接从保存的 Session 恢复
                account = igapi.AccountInfo.parse(
                    saved_sessions[username], platform="android"
                )
                clients[username] = igapi.Client(account=account)
                print(f"[{username}] 已从 Session 恢复")
            else:
                # 首次登录
                client = igapi.Client()
                try:
                    await client.login(username, password)
                    clients[username] = client
                    print(f"[{username}] 登录成功")
                    time.sleep(2)  # 登录间隔，避免触发风控
                except igapi.TwoFactorRequired:
                    code = input(f"[{username}] 请输入双因素验证码：")
                    await client.verify_two_factor(code)
                    clients[username] = client
                except ValueError as e:
                    print(f"[{username}] 登录失败：{e}")

    return clients


def save_all_sessions(clients: dict, accounts_file: str):
    """将所有账号的当前 Session 保存到文件"""
    with open(accounts_file, "w", encoding="utf-8") as f:
        f.write("# 自动生成，请勿手动编辑\n")
        for username, client in clients.items():
            if client.is_logged_in():
                session_str = client.export_account_string()
                f.write(session_str + "\n")
    print(f"已保存 {len(clients)} 个 Session 到 {accounts_file}")


async def batch_get_user_info(clients: dict, target_user_id: int):
    """用所有账号获取同一用户的信息（演示批量操作）"""
    results = {}

    for username, client in clients.items():
        try:
            user = await client.user().info(user_id=target_user_id)
            results[username] = user
            print(f"[{username}] 查询成功：@{user.username}")
            time.sleep(1)  # 请求间隔
        except PermissionError:
            print(f"[{username}] Session 已失效，跳过")
        except Exception as e:
            print(f"[{username}] 查询出错：{e}")

    return results


async def main():
    clients = await load_or_login(CREDENTIALS_FILE, ACCOUNTS_FILE)
    print(f"共 {len(clients)} 个账号就绪")

    # 批量操作
    await batch_get_user_info(clients, target_user_id=25025320)

    # 保存最新 Session
    save_all_sessions(clients, ACCOUNTS_FILE)


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

账号池管理¶

在需要轮换使用账号的场景中（如分散请求频率、避免单账号触发速率限制），可以实现一个简单的账号池：

import igapi
import itertools
import time


class AccountPool:
    """简单的账号池，支持轮询使用"""

    def __init__(self, accounts_file: str, platform: str = "android"):
        self._clients = []
        self._cycle = None
        self._load(accounts_file, platform)

    def _load(self, accounts_file: str, platform: str):
        with open(accounts_file, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                account = igapi.AccountInfo.parse(line, platform=platform)
                client = igapi.Client(account=account)
                self._clients.append(client)

        if not self._clients:
            raise ValueError("账号文件为空，无法初始化账号池")

        # 无限轮询迭代器
        self._cycle = itertools.cycle(self._clients)
        print(f"账号池已加载 {len(self._clients)} 个账号")

    def next(self) -> igapi.Client:
        """获取下一个账号（轮询）"""
        return next(self._cycle)

    def size(self) -> int:
        """返回账号池中的账号数量"""
        return len(self._clients)


async def main():
    pool = AccountPool("accounts.txt")

    user_ids = [25025320, 460563723, 12345678, 98765432, 11223344]

    for user_id in user_ids:
        client = pool.next()  # 轮询取一个账号

        try:
            user = await client.user().info(user_id=user_id)
            print(f"查询 {user_id} 成功：@{user.username}")
        except PermissionError:
            print(f"查询 {user_id} 失败：Session 无效")
        except Exception as e:
            print(f"查询 {user_id} 出错：{e}")

        time.sleep(1)  # 每次请求后等待

asyncio.run(main())

Session 批量保存和恢复¶

保存 Session¶

import igapi

clients: dict[str, igapi.Client] = {}  # username -> client

# ... 登录或加载账号 ...

# 批量保存（is_logged_in 和 export_account_string 均为同步方法）
with open("sessions.txt", "w", encoding="utf-8") as f:
    for username, client in clients.items():
        if client.is_logged_in():
            f.write(client.export_account_string() + "\n")
        else:
            print(f"[{username}] 未登录，跳过保存")

恢复 Session¶

import igapi

clients = {}

with open("sessions.txt", "r", encoding="utf-8") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        account = igapi.AccountInfo.parse(line, platform="android")
        clients[account.username] = igapi.Client(account=account)

print(f"已恢复 {len(clients)} 个账号的 Session")

验证 Session 有效性¶

is_logged_in() 仅检查本地是否有 sessionid，不发起网络请求。若需验证 Session 是否真的有效，需要执行一次实际 API 调用：

import asyncio
import igapi


async def verify_sessions(clients: dict) -> dict:
    """
    验证所有账号的 Session 是否有效。
    返回字典：{username: is_valid}
    """
    results = {}

    for username, client in clients.items():
        if not client.is_logged_in():
            results[username] = False
            print(f"[{username}] 本地无 Session")
            continue

        try:
            # 用一次轻量请求验证真实有效性
            await client.user().id_from_username(username)
            results[username] = True
            print(f"[{username}] Session 有效")
        except PermissionError:
            results[username] = False
            print(f"[{username}] Session 已过期")
        except Exception as e:
            results[username] = False
            print(f"[{username}] 验证失败：{e}")

    return results

注意事项¶

每个实例独立 HTTP 客户端¶

每个 Client 实例持有独立的 reqwest HTTP 客户端，包括独立的连接池。在需要管理大量账号时，这意味着每个实例都会占用一定的系统资源（文件描述符、内存等）。

内存占用评估¶

单个 Client 实例的内存开销主要来自：HTTP 连接池（约 1-5 MB，视并发活跃连接数）、Session 状态（数 KB）、设备信息（数 KB）。管理 100 个账号通常需要约 100-500 MB 内存，需根据实际情况评估。

速率限制是账号维度的¶

Instagram 的速率限制基于账号，而不是 IP。多账号轮询可以分散每个账号的请求频率，但不能绕过总请求量的限制。即使使用账号池，也应在请求之间保持适当间隔（建议每个账号每次请求后等待 1-3 秒）。

线程安全¶

单个 Client 实例是线程安全的，多个线程可以同时调用同一实例的方法。但由于 Python GIL 的存在，多线程在 CPU 密集型任务上无法实现真正的并行。对于 I/O 密集型的 API 调用，多线程可以有效提升吞吐量。

常见问题¶

Q：批量操作时遇到 PermissionError，是所有账号都失效了吗？

A：不一定。PermissionError 可能由以下原因之一触发：(1) 该账号的 Session 已过期，需要重新登录；(2) 目标用户是私密账号且未与该账号互关；(3) 该账号被临时限制了访问权限。建议对抛出异常的账号单独调用验证逻辑，区分具体原因。

Q：能否在多线程中同时使用不同的 Client 实例？

A：可以。不同 Client 实例之间完全独立，在多个线程中同时使用不同实例是安全的。也可以在多个线程中使用同一个 Client 实例，该实例内部状态受锁保护。

Q：export_account_string() 导出的字符串包含明文密码吗？

A：是的，序列化字符串的格式为 用户名:密码||设备信息|cookies||，其中包含明文密码。请妥善保管账号文件，建议设置严格的文件权限（如 chmod 600 sessions.txt），并将账号文件加入 .gitignore，避免提交到版本控制系统。

Q：加载账号后如何确认 Session 是否还有效？

A：is_logged_in() 仅检查本地状态（有 sessionid 就返回 True），不发起网络请求。若要验证 Session 是否真的有效，需要执行一次轻量 API 调用（如 client.user().id_from_username(username)），根据是否抛出 PermissionError 来判断。详见本文档验证 Session 有效性一节。