跳转至

igapi 异常类型参考

1. 概述

igapi 模块的异常体系分为两层:

标准 Python 异常:将底层 Rust 错误映射到语义最贴近的内置类型,无需额外 import,与 Python 惯用的 try/except 模式无缝集成。

自定义异常igapi.TwoFactorRequiredigapi.ChallengeRequired 继承自 Exception,在抛出时附带结构化的 dict 属性,用于程序化处理双因素验证和安全挑战流程。

异常设计原则:可预期的业务错误使用精确类型,不可预期的系统错误使用 RuntimeErrorKeyError 表示资源不存在,PermissionError 表示权限或状态受限,ValueError 表示输入或响应格式有误,ConnectionError 仅用于网络层故障。


2. 异常分类汇总表

按 Python 异常类型分组,覆盖全部错误映射。

ValueError 组

输入参数无效、服务端响应格式异常、数据解析失败时抛出。

触发场景 错误消息示例
密码与账号不匹配 Bad password
用户名不存在或账号已停用 Invalid user
服务端返回无法解析的响应(状态码异常) Invalid response [400]: ...
响应体不是合法 JSON (serde_json 错误描述)
AccountInfo.parse() 格式错误或其他字符串解析失败 Parse error: ...

PermissionError 组

账号状态受限、访问权限不足、Session 失效时抛出。

触发场景 错误消息示例
调用需要认证的接口前未登录 Login required
账号被 Instagram 临时或永久封禁 Account locked
Cookie 失效,Session 已过期 Session expired
尝试访问私密账号内容(未获授权) Private account
账号需要在官方应用中先同意新条款 Consent required

KeyError 组

请求的资源在服务端不存在时抛出。

触发场景 错误消息示例
通过 user().info(pk) 查询已删除或不存在的用户 User not found
通过 media().info(id) 查询已删除或不存在的媒体 Media not found

RuntimeError 组

初始化失败、速率限制、加密错误、Session 读写失败及其他系统级错误时抛出。

触发场景 错误消息示例
登录前的初始化请求失败(网络或 IP 封禁) Login init failed: ...
请求过于频繁,被服务端限流 Rate limit exceeded, retry after: Some(30s)
密码加密过程出错 (加密错误描述)
请求签名生成失败 (签名错误描述)
Session 文件读取失败 (读取错误描述)
Session 文件写入失败 (写入错误描述)
其他未分类的内部错误 (错误描述)

ConnectionError 组

底层网络层发生故障时抛出,由 reqwest HTTP 客户端错误直接转换而来。

触发场景 错误消息示例
连接超时、DNS 解析失败、代理配置错误、TLS 握手失败 (reqwest 错误描述)

igapi.TwoFactorRequired

账号开启了双因素验证(2FA),服务端要求提供验证码时抛出。

属性 类型 错误消息
.two_factor_info (dict) igapi.TwoFactorRequired Two-factor authentication required

igapi.ChallengeRequired

Instagram 检测到异常登录行为,要求完成安全挑战验证时抛出。

属性 类型 错误消息
.challenge_info (dict) igapi.ChallengeRequired Challenge required

3. 自定义异常详解

igapi.TwoFactorRequired

继承关系Exceptionigapi.TwoFactorRequired

账号主动开启 2FA 保护后,每次登录均会触发此异常。异常实例附带 two_factor_info 属性,包含完成 2FA 流程所需的全部信息。

属性说明:

属性名 类型 说明
two_factor_info dict 服务端返回的 2FA 信息字典

two_factor_info 的常见字段:

字段 类型 说明
two_factor_identifier str 2FA 会话标识符,提交验证码时必须携带
username str 正在登录的用户名
obfuscated_phone_number str 脱敏手机号,如 ***-***-1234
totp_two_factor_on bool 是否启用了 TOTP(如 Google Authenticator)
sms_two_factor_on bool 是否启用了 SMS 验证

处理示例:

import asyncio
import igapi

async def main():
    client = igapi.Client()

    try:
        await client.login("your_user", "your_pass")
    except igapi.TwoFactorRequired as e:
        info = e.two_factor_info
        identifier = info["two_factor_identifier"]
        username = info["username"]

        print(f"账号 {username} 需要 2FA 验证")

        # 提示用户输入验证码(TOTP 或短信)
        code = input("请输入 6 位验证码:").strip()

        # 提交 2FA 验证码完成登录
        await client.login_2fa(
            two_factor_identifier=identifier,
            verification_code=code,
        )
        print("登录成功!")

asyncio.run(main())

igapi.ChallengeRequired

继承关系Exceptionigapi.ChallengeRequired

Instagram 检测到异常登录行为(如新设备、新 IP 地址)后触发,要求完成额外的安全验证。与 2FA 不同,此验证不会在每次登录时触发,通常需要通过官方渠道(邮件或手机短信)完成。

属性说明:

属性名 类型 说明
challenge_info dict 服务端返回的挑战信息字典

challenge_info 的常见字段:

字段 类型 说明
challenge_url str 挑战验证的 API 路径
challenge_context str 挑战上下文标识符
flow_render_type int 验证流程类型编号
bloks_action str 客户端操作标识

处理示例:

import asyncio
import igapi

async def main():
    client = igapi.Client()

    try:
        await client.login("your_user", "your_pass")
    except igapi.ChallengeRequired as e:
        info = e.challenge_info
        print("账号触发安全验证,挑战信息:")
        print(f"  验证路径:{info.get('challenge_url', '未知')}")
        print(f"  流程类型:{info.get('flow_render_type', '未知')}")

        # ChallengeRequired 通常需要人工介入
        # 请在 Instagram 官方应用中完成验证后,使用新 Session 重新初始化客户端
        print("请在 Instagram 应用中完成安全验证后重试")

asyncio.run(main())

4. 错误处理最佳实践

以下示例展示在实际使用场景中处理各类异常的完整模式,涵盖登录、用户查询、媒体查询的全部异常情况。

import asyncio
import igapi
import time


async def safe_login(username: str, password: str) -> igapi.Client:
    """带完整错误处理的登录函数。"""
    client = igapi.Client()

    try:
        await client.login(username, password)
        print(f"登录成功:{username}")
        return client

    except igapi.TwoFactorRequired as e:
        # 账号启用了 2FA,需要用户输入验证码
        info = e.two_factor_info
        identifier = info["two_factor_identifier"]
        print(f"账号 {username} 需要 2FA 验证")
        code = input("请输入验证码(6位):").strip()
        await client.login_2fa(
            two_factor_identifier=identifier,
            verification_code=code,
        )
        print("2FA 验证通过,登录成功!")
        return client

    except igapi.ChallengeRequired as e:
        # 触发安全挑战,通常需要人工介入后重新登录
        print("账号触发安全验证,请在 Instagram 应用中完成验证后重试")
        print(f"挑战信息:{e.challenge_info}")
        raise

    except ValueError as e:
        # 密码错误、用户无效、响应格式异常等
        print(f"登录失败(凭证无效或响应异常):{e}")
        raise

    except PermissionError as e:
        # 账号锁定、需要同意条款等账号状态问题
        print(f"账号受限:{e}")
        raise

    except RuntimeError as e:
        # 速率限制时等待后重试,其他运行时错误直接上抛
        if "Rate limit" in str(e):
            print(f"触发速率限制:{e}")
            print("等待 60 秒后重试...")
            time.sleep(60)
            return await safe_login(username, password)
        raise

    except ConnectionError as e:
        # 网络层故障,检查代理配置
        print(f"网络连接失败,请检查代理配置:{e}")
        raise


async def safe_get_user(client: igapi.Client, user_pk: int) -> "igapi.User | None":
    """安全获取用户信息,不存在或私密时返回 None。"""
    try:
        return await client.user().info(user_pk)
    except KeyError:
        # 用户不存在或已删除
        print(f"用户 {user_pk} 不存在")
        return None
    except PermissionError as e:
        if "Private account" in str(e):
            # 私密账号,无法访问
            print(f"用户 {user_pk} 是私密账号")
            return None
        if "Session expired" in str(e):
            # Session 过期,需要重新登录后再试
            raise
        raise


async def safe_get_media(client: igapi.Client, media_id: str) -> "igapi.Media | None":
    """安全获取媒体信息,不存在时返回 None。"""
    try:
        return await client.media().info(media_id)
    except KeyError:
        # 媒体不存在或已删除
        print(f"媒体 {media_id} 不存在或已删除")
        return None


# 使用示例
async def main():
    try:
        client = await safe_login("john_doe", "mypassword")

        user = await safe_get_user(client, 123456789)
        if user:
            print(f"找到用户:{user.username}{user.follower_count} 粉丝)")

        media = await safe_get_media(client, "3456789012345678901_123456")
        if media:
            print(f"找到媒体:类型={media.media_type},点赞={media.like_count}")

    except Exception as e:
        print(f"未处理的异常:{type(e).__name__}: {e}")


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

5. 常见问题

Q:TwoFactorRequiredChallengeRequired 的区别是什么?

TwoFactorRequired 是账号主动开启的 2FA 保护,每次登录都会触发,属于预期流程,程序应捕获并引导用户输入验证码。ChallengeRequired 是 Instagram 检测到可疑行为后触发的额外安全校验,不会在每次登录时出现,通常需要通过官方渠道(邮件、手机)手动完成,无法完全程序化处理。

Q:如何判断 PermissionError 的具体原因?

检查异常消息字符串的内容:

except PermissionError as e:
    msg = str(e)
    if "Session expired" in msg:
        # Session 过期,重新登录
        pass
    elif "Account locked" in msg:
        # 账号被封
        pass
    elif "Login required" in msg:
        # 尚未登录
        pass
    elif "Private account" in msg:
        # 私密账号
        pass
    elif "Consent required" in msg:
        # 需要同意条款
        pass

Q:RuntimeError: Rate limit exceeded 时应等待多久?

错误消息中通常包含 retry after: Some(30s) 格式的提示,可通过字符串解析提取等待时间。若无明确提示,建议至少等待 60 秒后再重试。频繁触发限流可能导致账号被临时封禁,建议在业务逻辑层控制请求频率。

Q:ValueError: Parse error 在什么情况下出现?

最常见于 AccountInfo.parse() 传入了格式不正确的字符串。常见原因包括:缺少 || 分隔符、cookies 段缺少 sessionidds_user_id 字段、sessionid 未进行 URL decode。请参考 类型参考 - AccountInfo 中的格式说明。

Q:ConnectionError 一定是网络问题吗?

是的,ConnectionError 仅映射自 reqwest 的网络层错误,包括连接超时、DNS 解析失败、代理不可达、TLS 握手失败等。若使用了代理,请确认代理地址格式正确(如 http://127.0.0.1:7890)且代理服务正在运行。

Q:捕获异常时应使用哪个顺序?

自定义异常(igapi.TwoFactorRequiredigapi.ChallengeRequired)继承自 Exception,应排在内置异常之前捕获,避免被 Exception 基类先行捕获。标准异常之间无继承关系,顺序可根据业务需要调整,建议将最具体的错误类型排在前面。


相关文档