最佳实践

本文档总结使用 igapi 时经过验证的最佳实践模式，帮助你写出更健壮、更安全、更易维护的代码。

---

错误处理模式

完整异常处理模板

igapi 将 Instagram API 的各类错误映射为 Python 内建异常和库自定义异常。以下模板覆盖了登录场景中所有常见的异常：

import asyncio
import igapi


async def safe_login(
    username: str,
    password: str,
    two_fa_secret: str | None = None,
    platform: str = "web",
) -> igapi.WebClient | igapi.Client | None:
    """
    安全登录函数，返回已登录的客户端，失败返回 None。

    参数：
        username      - Instagram 用户名
        password      - 账号密码
        two_fa_secret - TOTP 密钥（可选，用于自动处理双因素认证）
        platform      - 平台类型："web" 或 "android"
    """
    if platform == "web":
        client = igapi.WebClient()
    else:
        client = igapi.Client()

    try:
        await client.login(username, password, two_fa_secret=two_fa_secret)
        return client

    except igapi.TwoFactorRequired:
        # 提供了 two_fa_secret 时不应走到这里，走到这里说明密钥有误或未提供
        code = input(f"[{username}] 请输入双因素验证码：")
        try:
            await client.verify_two_factor(code)
            return client
        except ValueError as e:
            print(f"[{username}] 验证码错误：{e}")
            return None

    except igapi.ChallengeRequired as e:
        # Instagram 要求完成安全验证挑战（如邮件/短信验证）
        print(f"[{username}] 需要通过安全验证：{e}")
        print("请手动完成验证后，重新使用已通过验证的 Session 登录")
        return None

    except ValueError as e:
        # 密码错误、账号不存在、账号被封等
        print(f"[{username}] 登录失败：{e}")
        return None

    except ConnectionError as e:
        # 网络不通、代理配置错误、DNS 解析失败等
        print(f"[{username}] 网络连接失败：{e}")
        return None

    except Exception as e:
        # 未预期的异常，记录完整信息以便排查
        print(f"[{username}] 未知错误：{type(e).__name__}: {e}")
        return None

API 调用的错误处理

登录后的 API 调用同样需要完整的错误处理：

import asyncio
import igapi
import time


async def get_user_info_with_retry(
    client: igapi.Client,
    user_id: int,
    max_retries: int = 3,
) -> igapi.User | None:
    """
    获取用户信息，自动重试（指数退避）。

    返回：
        成功时返回 igapi.User，失败时返回 None
    """
    for attempt in range(max_retries):
        try:
            return await client.user().info(user_id=user_id)

        except PermissionError:
            # Session 失效或私密账号，不重试
            print(f"用户 {user_id}：无访问权限（Session 失效或私密账号）")
            return None

        except KeyError:
            # 用户不存在，不重试
            print(f"用户 {user_id} 不存在")
            return None

        except ConnectionError as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 指数退避：1s, 2s, 4s
                print(f"网络错误，{wait} 秒后重试（{attempt + 1}/{max_retries}）：{e}")
                time.sleep(wait)
            else:
                print(f"网络错误，已达最大重试次数：{e}")
                return None

        except Exception as e:
            print(f"未知错误：{type(e).__name__}: {e}")
            return None

    return None

异常类型速查表

异常类型	含义	建议处理方式
igapi.TwoFactorRequired	账号启用了双因素认证	提示用户输入验证码，调用 verify_two_factor()
igapi.ChallengeRequired	需要完成安全验证挑战	提示用户手动完成，不适合自动化处理
ValueError	密码错误、账号不存在、参数非法等	记录错误信息，不重试
PermissionError	未登录、Session 失效、私密账号限制	检查 Session 状态，必要时重新登录
KeyError	资源不存在（用户 ID、媒体 ID 无效）	记录并跳过，不重试
ConnectionError	网络问题、代理故障、超时	指数退避重试

---

Session 管理

核心原则：登录一次，持久化复用

Instagram 对登录行为有严格的风控策略，频繁登录同一账号（尤其是在不同设备信息下）会触发安全验证甚至封号。正确的做法是首次登录后立即保存 Session，后续直接复用。

import asyncio
import igapi
import os


SESSION_FILE = "session.txt"


async def get_client() -> igapi.WebClient:
    """
    获取已登录的客户端。
    优先从文件恢复 Session，文件不存在时才执行登录。
    """
    if os.path.exists(SESSION_FILE):
        with open(SESSION_FILE, "r", encoding="utf-8") as f:
            session_str = f.read().strip()

        if session_str:
            account = igapi.AccountInfo.parse(session_str, platform="web")
            client = igapi.WebClient(account=account)

            # 简单验证 Session 是否有效
            if client.is_logged_in():
                print("已从文件恢复 Session")
                return client

    # Session 不存在或无效，执行登录
    username = os.environ["IG_USERNAME"]
    password = os.environ["IG_PASSWORD"]

    client = igapi.WebClient()
    await client.login(username, password)

    # 立即保存 Session
    with open(SESSION_FILE, "w", encoding="utf-8") as f:
        f.write(client.export_account_string())
    print("登录成功，Session 已保存")

    return client


# 使用示例
async def main():
    client = await get_client()
    user = await client.user().info(user_id=25025320)
    print(f"@{user.username}")

asyncio.run(main())

Session 过期的处理

is_logged_in() 只检查本地状态，无法检测服务端 Session 是否已失效。对于长期运行的程序，建议在 PermissionError 时触发重新登录：

import asyncio
import igapi
import os


async def with_auto_relogin(client_factory, action):
    """
    执行操作，若 Session 失效则自动重新登录后重试。

    参数：
        client_factory - 无参异步函数，返回新的已登录客户端
        action         - 接受 client 参数的异步函数
    """
    client = await client_factory()

    try:
        return await action(client)
    except PermissionError:
        print("Session 已失效，正在重新登录...")
        # 删除旧 Session 文件，强制重新登录
        if os.path.exists("session.txt"):
            os.remove("session.txt")
        client = await client_factory()
        return await action(client)


# 使用示例
async def main():
    result = await with_auto_relogin(
        client_factory=get_client,
        action=lambda c: c.user().info(user_id=25025320),
    )

asyncio.run(main())

---

速率限制应对

请求间隔策略

Instagram 对 API 调用频率有严格限制，超出阈值会触发临时封禁（PermissionError）或返回错误。建议在每次请求后加入随机延迟，模拟真实用户行为：

import asyncio
import igapi
import time
import random


def human_delay(min_seconds: float = 1.0, max_seconds: float = 3.0):
    """模拟人工操作间隔，加入随机抖动"""
    delay = random.uniform(min_seconds, max_seconds)
    time.sleep(delay)


async def main():
    account = igapi.AccountInfo.parse("用户名:密码||设备信息|cookies||")
    client = igapi.Client(account=account)
    user_ids = [25025320, 460563723, 12345678]

    for user_id in user_ids:
        try:
            user = await client.user().info(user_id=user_id)
            print(f"@{user.username}")
        except Exception as e:
            print(f"跳过 {user_id}：{e}")

        human_delay(1.5, 4.0)  # 每次请求后随机等待 1.5-4 秒

asyncio.run(main())

指数退避重试

遇到连接错误或临时性失败时，使用指数退避策略重试，避免立即重试加剧服务器负担：

import asyncio
import igapi
import time


async def with_exponential_backoff(func, max_retries: int = 3, base_delay: float = 2.0):
    """
    带指数退避的重试装饰器（异步版本）。

    重试间隔：2s -> 4s -> 8s（base_delay 的幂次）
    """
    last_exception = None

    for attempt in range(max_retries):
        try:
            return await func()
        except (ConnectionError, TimeoutError) as e:
            last_exception = e
            if attempt < max_retries - 1:
                delay = base_delay ** (attempt + 1)
                print(f"请求失败，{delay:.0f} 秒后重试（{attempt + 1}/{max_retries}）")
                time.sleep(delay)
        except (PermissionError, KeyError, ValueError):
            # 业务逻辑错误，不重试
            raise

    raise last_exception


async def main():
    account = igapi.AccountInfo.parse("用户名:密码||设备信息|cookies||")
    client = igapi.Client(account=account)

    # 使用示例
    result = await with_exponential_backoff(
        func=lambda: client.user().info(user_id=25025320),
        max_retries=3,
        base_delay=2.0,
    )

asyncio.run(main())

请求频率参考

以下为建议的请求间隔，这些值基于保守估计，实际限制因账号状态和操作类型而异：

操作类型	建议间隔	说明
获取用户信息 user().info()	1-3 秒	单账号查询限制较宽松
搜索用户 user().search()	2-5 秒	搜索接口限制较严格
获取 Feed feed().timeline()	3-5 秒	模拟刷动态行为
发布内容	10-30 秒	发布操作触发风控的可能性最高
批量操作（不同账号轮换）	1-2 秒	账号间轮换可适当缩短间隔

---

安全建议

使用环境变量存储凭据

禁止将密码和 Session 字符串硬编码在源代码中，应通过环境变量传入：

import asyncio
import igapi
import os

# 正确：从环境变量读取
async def main():
    username = os.environ["IG_USERNAME"]
    password = os.environ["IG_PASSWORD"]

    client = igapi.WebClient()
    await client.login(username, password)

asyncio.run(main())

在终端设置环境变量：

export IG_USERNAME="你的用户名"
export IG_PASSWORD="你的密码"
python your_script.py

或者使用 .env 文件配合 python-dotenv：

# .env 文件（不要提交到 git）
IG_USERNAME=你的用户名
IG_PASSWORD=你的密码

from dotenv import load_dotenv
import os
import igapi

load_dotenv()  # 加载 .env 文件

username = os.environ["IG_USERNAME"]
password = os.environ["IG_PASSWORD"]

确保 .env 文件已添加到 .gitignore：

# .gitignore
.env
session.txt
sessions.txt
accounts.txt
*.session

Session 文件权限

保存账号 Session 的文件包含明文密码，需要严格控制文件权限：

# 设置文件仅所有者可读写
chmod 600 session.txt
chmod 600 sessions.txt
chmod 600 accounts.txt

代理安全

仅在本地调试时使用 danger_accept_invalid_certs=True。在正式环境或处理真实用户凭据时，始终使用受信任的证书。如果使用第三方代理服务，请确认代理提供商的可信度，因为代理节点可以看到所有未加密的请求内容。

---

调试技巧

配合代理抓包

最直接的调试方式是通过抓包代理查看实际请求和响应，详见 代理配置 文档。

快速启动调试客户端的通用模板：

import igapi
import os

# 通过环境变量控制是否开启调试模式
DEBUG = os.environ.get("IGAPI_DEBUG", "0") == "1"

def make_client(**kwargs) -> igapi.WebClient:
    """创建客户端，调试模式下自动配置 mitmproxy"""
    if DEBUG:
        return igapi.WebClient(
            proxy="http://127.0.0.1:8080",
            danger_accept_invalid_certs=True,
            http1_only=True,
            **kwargs,
        )
    return igapi.WebClient(**kwargs)


client = make_client()

启动时：

# 调试模式
IGAPI_DEBUG=1 python your_script.py

# 正常模式
python your_script.py

打印异常完整信息

排查问题时，建议打印完整的异常信息（包含类型和堆栈）：

import asyncio
import traceback
import igapi

async def main():
    account = igapi.AccountInfo.parse("用户名:密码||设备信息|cookies||")
    client = igapi.Client(account=account)

    try:
        await client.user().info(user_id=99999999999)
    except Exception as e:
        # 打印完整堆栈
        traceback.print_exc()
        # 或只打印异常类型和消息
        print(f"异常类型：{type(e).__name__}")
        print(f"异常消息：{e}")

asyncio.run(main())

验证 Session 状态

在执行批量操作前，先验证 Session 有效性，避免因 Session 失效导致大量请求失败：

import asyncio
import igapi

async def check_session(client: igapi.Client | igapi.WebClient, username: str) -> bool:
    """
    验证 Session 是否有效。
    通过一次轻量 API 调用检测，返回 True 表示有效。
    """
    if not client.is_logged_in():
        print(f"[{username}] 本地无 Session")
        return False

    try:
        await client.user().id_from_username(username)
        return True
    except PermissionError:
        print(f"[{username}] Session 已过期")
        return False
    except Exception as e:
        print(f"[{username}] 验证失败：{e}")
        return False

---

常见反模式

不要每次都登录

# 反模式：每次运行都重新登录
async def get_user_info(user_id):
    client = igapi.WebClient()
    await client.login("用户名", "密码")  # 每次都登录
    return await client.user().info(user_id=user_id)

频繁登录会触发 Instagram 的安全验证机制，导致账号被要求完成 Challenge 甚至被封禁。正确做法是登录一次后保存 Session，后续直接复用（参见 Session 管理 章节）。

# 正确：复用已保存的 Session
async def get_user_info(user_id):
    client = await get_client()  # 内部优先从文件恢复 Session
    return await client.user().info(user_id=user_id)

不要忽略异常

# 反模式：忽略所有异常
async def bad_example(client, user_id):
    try:
        user = await client.user().info(user_id=user_id)
    except Exception:
        pass  # 悄悄失败，完全不知道发生了什么

忽略异常会掩盖真正的问题，如 Session 失效、账号被限制、网络故障等。最低限度应该记录异常类型和消息：

# 正确：至少记录异常
async def good_example(client, user_id):
    try:
        user = await client.user().info(user_id=user_id)
    except PermissionError:
        print(f"用户 {user_id}：无访问权限（Session 失效或私密账号）")
        user = None
    except KeyError:
        print(f"用户 {user_id} 不存在")
        user = None
    except Exception as e:
        print(f"用户 {user_id}：未知错误 {type(e).__name__}: {e}")
        user = None

不要硬编码密码

# 反模式：密码写在代码里
await client.login("myaccount", "mypassword123")

密码一旦提交到版本控制系统，即使后来删除，在历史记录中仍然可以被检索到。正确做法是通过环境变量传入（参见 安全建议 章节）：

# 正确：从环境变量读取
import os
await client.login(os.environ["IG_USERNAME"], os.environ["IG_PASSWORD"])

不要在循环中创建新客户端

# 反模式：每次循环都创建新 Client
async def bad_loop(user_ids, account):
    for user_id in user_ids:
        client = igapi.Client(account=account)  # 每次都创建新实例，浪费资源
        user = await client.user().info(user_id=user_id)

每个 Client 实例会初始化独立的 HTTP 连接池。在循环中反复创建新实例会导致资源浪费和连接池无法被复用。应在循环外创建一次实例：

# 正确：在循环外创建客户端
async def good_loop(user_ids, account):
    client = igapi.Client(account=account)

    for user_id in user_ids:
        user = await client.user().info(user_id=user_id)