跳转至

Session 管理

概述

igapi 的 Session 管理围绕 AccountInfo 对象展开。登录成功后,客户端将认证状态(cookies、设备信息、用户 ID 等)存储在内存中,通过 export_account()export_account_string() 可将其序列化为可持久化的对象或字符串。下次启动时,只需将 AccountInfo 对象传入客户端构造函数,即可跳过登录流程直接恢复已认证状态。

Session 管理解决的核心问题:

  • 避免每次运行程序都重新登录,减少对账号风控配额的消耗
  • 在进程重启、服务器迁移等场景下保持登录状态连续
  • 支持将 session 持久化到文件、数据库或环境变量中

导出 Session

登录成功后,可通过以下两种方式导出当前 session:

方式一:导出为 AccountInfo 对象

import asyncio
import igapi

async def main():
    client = igapi.android.Client()
    await client.login("your_username", "your_password")

    # 导出为 AccountInfo 对象,可直接传入客户端构造函数
    account = client.export_account()

    print(account.username)    # 用户名
    print(account.platform)    # "android"
    print(account.has_session) # True(已登录)
    print(account.user_id)     # Instagram 用户 ID(int)
    print(account.session_id)  # sessionid cookie 值

asyncio.run(main())

方式二:导出为字符串

import asyncio
import igapi

async def main():
    client = igapi.android.Client()
    await client.login("your_username", "your_password")

    # 导出为字符串,便于持久化存储
    session_str = client.export_account_string()
    print(session_str)
    # 输出示例: your_username:your_password||android_id;phone_id;uuid;device_id|sessionid=xxx; ds_user_id=yyy; ...||

asyncio.run(main())

Web 平台导出

igapi.web.Client 的导出方式与 Android 客户端相同:

import asyncio
import igapi

async def main():
    web = igapi.web.Client()
    await web.login("your_username", "your_password")

    account = web.export_account()
    print(account.platform)    # "web"
    print(account.csrf_token)  # Web 平台独有,CSRF Token

    session_str = web.export_account_string()

asyncio.run(main())

恢复 Session

从 AccountInfo 对象恢复

import asyncio
import igapi

async def main():
    # 假设已从某处获取到 session 字符串
    account = igapi.android.AccountInfo.parse(saved_string)

    # 传入 account 参数,无需再调用 login()
    client = igapi.android.Client(account=account)

    # 验证 session 是否有效(仅检查内存状态,不发起网络请求)
    print(await client.is_logged_in())  # True

asyncio.run(main())

从字符串解析并恢复

import igapi

saved_string = "your_username:your_password||android_id;phone_id;uuid;device_id|cookies||"

# 解析字符串为 AccountInfo 对象
account = igapi.android.AccountInfo.parse(saved_string)

# 恢复客户端
client = igapi.android.Client(account=account)

Web 平台从字符串恢复

import igapi

saved_string = "..."  # web 平台导出的字符串

account = igapi.web.AccountInfo.parse(saved_string)
web = igapi.web.Client(account=account)

注意:Android 和 Web 的 session 字符串格式不同,不能混用。 必须使用与字符串来源一致的平台命名空间解析: Android 字符串用 igapi.android.AccountInfo.parse(), Web 字符串用 igapi.web.AccountInfo.parse()AccountInfo 的平台类型必须与客户端类型匹配(Android AccountInfo 不能传给 Web 客户端,反之亦然)。


文件持久化 Session(logout / save_session / restore_session)

除了 AccountInfo 字符串导出/解析,igapi.android.Clientigapi.ios.Clientigapi.web.Client 三个客户端还提供更轻量的文件持久化方式:save_session(path) / restore_session(path) 直接读写本地 JSON 文件,无需手动拼接账号字符串;logout() 用于登出并清理内存中的登录态。

三个方法均为异步方法,签名一致:

方法 说明
await client.logout() 调用 Instagram 登出端点,清除内存中的 user_idcsrf_token 等登录态。登出后客户端仍可复用(如重新调用 login()
await client.save_session(path: str) 将当前会话状态(cookies、设备信息、用户 ID 等)序列化为 JSON 写入 path
await client.restore_session(path: str) path 读取 save_session() 写入的 JSON 文件,恢复到当前客户端的内存状态

注意save_session / restore_session 只操作内存中的会话状态SessionState),与 export_account() / AccountInfo.parse() 的账号字符串格式不同,两者不能混用读写。选择哪种持久化方式取决于场景:需要跨语言/跨进程传递完整账号信息(含密码)时用 AccountInfo 字符串;只需要在同一 Python 环境内保存/恢复登录态、不关心密码明文时,用 save_session / restore_session 更简洁。

保存与恢复示例

import asyncio
import igapi

SESSION_PATH = "session.json"

async def main():
    client = igapi.android.Client()
    await client.login("your_username", "your_password")

    # 保存当前会话到文件
    await client.save_session(SESSION_PATH)

    # ... 进程重启后,创建新客户端并恢复会话 ...
    new_client = igapi.android.Client()
    await new_client.restore_session(SESSION_PATH)
    print(await new_client.is_logged_in())  # True

asyncio.run(main())

登出示例

import asyncio
import igapi

async def main():
    client = igapi.web.Client()
    await client.login("your_username", "your_password")

    # 使用完毕后登出,清理内存登录态
    await client.logout()
    print(await client.is_logged_in())  # False

asyncio.run(main())

Android / iOS / Web 三个客户端的 logout() / save_session() / restore_session() 用法完全一致,仅需替换 igapi.android.Clientigapi.ios.Clientigapi.web.Client


AccountInfo 属性参考

AccountInfo 对象包含认证所需的全部信息。

通用属性(Android 和 Web 均有)

属性 类型 说明
username str Instagram 用户名
password str 账号密码(明文存储在对象中)
platform str 平台标识,"android""web"
has_session bool 是否包含有效的 session 信息(即是否已登录)
user_id int \| None Instagram 用户 ID,登录前为 None
session_id str \| None sessionid cookie 值,未登录时为 None
mid str \| None Machine ID(mid cookie),设备标识符
ds_user_id int \| None ds_user_id cookie 值,与 user_id 一致

Android 专属属性

属性 类型 说明
android_id str \| None Android 设备 ID,格式 android-xxxxxxxxxxxxxxxx
phone_id str \| None 手机 ID(UUID 格式)
uuid str \| None 设备 UUID
device_id str \| None 设备 ID(UUID 格式)

Web 专属属性

属性 类型 说明
csrf_token str \| None CSRF Token,Web 请求验证必需

AccountInfo 方法

方法 返回值 说明
igapi.android.AccountInfo.parse(s) igapi.android.AccountInfo 从 Android 账号字符串解析
igapi.web.AccountInfo.parse(s) igapi.web.AccountInfo 从 Web 账号字符串解析
account.to_account_string() str 将 AccountInfo 序列化为字符串

Session 字符串格式

Session 字符串是 igapi 定义的序列化格式,包含恢复客户端所需的全部信息。

Android 格式

用户名:密码||android_id;phone_id;uuid;device_id|cookies||

示例:

<username>:<password>||android-a1b2c3d4e5f6a7b8;<phone_id>;<uuid>;<device_id>|sessionid=<sessionid>; ds_user_id=<user_id>; csrftoken=<csrftoken>; mid=<mid>||

字段说明:

字段位置 内容 说明
用户名:密码 明文凭证 : 分隔
android_id android-xxxxxxxxxxxxxxxx 16 个十六进制字符
phone_id UUID 手机 ID
uuid UUID 设备 UUID
device_id UUID 设备 ID
cookies key=value; key=value 多个 cookie 以 ; 分隔

Web 格式

用户名:密码||csrf_token|cookies||

示例:

<username>:<password>||<csrf_token>|sessionid=<sessionid>; ds_user_id=<user_id>; csrftoken=<csrftoken>; mid=<mid>||

字段说明:

字段位置 内容 说明
用户名:密码 明文凭证 : 分隔
csrf_token CSRF Token 字符串 Web 请求验证使用的 Token
cookies key=value; key=value 多个 cookie 以 ; 分隔

无论 Android 还是 Web,以下 cookie 字段在恢复 session 时是必需的:

Cookie 说明
sessionid 主要身份认证 cookie,需 URL decode
ds_user_id 用户 ID,与账号 pk 一致

以下字段为可选,但建议包含以提高兼容性:

Cookie 说明
csrftoken CSRF 保护 Token
mid Machine ID,设备标识
rur 路由信息(Regional URL Routing)

持久化存储示例

文件存储

最简单的持久化方案,适用于单机单账号场景:

import asyncio
import igapi
import os

SESSION_FILE = "session.txt"

async def get_client() -> igapi.android.Client:
    """获取已登录的客户端,优先从 session 文件恢复"""
    # 尝试从文件恢复
    if os.path.exists(SESSION_FILE):
        with open(SESSION_FILE, "r") as f:
            saved = f.read().strip()

        if saved:
            try:
                account = igapi.android.AccountInfo.parse(saved)
                client = igapi.android.Client(account=account)
                if await client.is_logged_in():
                    return client
            except Exception as e:
                print(f"session 恢复失败,将重新登录: {e}")

    # 全新登录
    client = igapi.android.Client()
    await client.login(
        os.environ["IG_USERNAME"],
        os.environ["IG_PASSWORD"],
    )

    # 保存 session 到文件
    with open(SESSION_FILE, "w") as f:
        f.write(client.export_account_string())

    return client

async def main():
    client = await get_client()
    # 使用 client 进行 API 调用...

asyncio.run(main())

数据库存储

适用于多账号管理或多进程共享场景:

import asyncio
import igapi
import sqlite3
import os

DB_FILE = "sessions.db"

def init_db(db_path: str) -> None:
    """初始化数据库,创建 sessions 表"""
    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS sessions (
                username TEXT PRIMARY KEY,
                platform TEXT NOT NULL,
                session_str TEXT NOT NULL,
                updated_at TEXT DEFAULT (datetime('now'))
            )
        """)

def save_session(db_path: str, username: str, platform: str, session_str: str) -> None:
    """保存或更新 session"""
    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            INSERT INTO sessions (username, platform, session_str, updated_at)
            VALUES (?, ?, ?, datetime('now'))
            ON CONFLICT(username) DO UPDATE SET
                session_str = excluded.session_str,
                updated_at = excluded.updated_at
        """, (username, platform, session_str))

def load_session(db_path: str, username: str, platform: str):
    """从数据库加载 session"""
    with sqlite3.connect(db_path) as conn:
        row = conn.execute(
            "SELECT session_str FROM sessions WHERE username = ? AND platform = ?",
            (username, platform)
        ).fetchone()
    if row:
        try:
            if platform == "web":
                return igapi.web.AccountInfo.parse(row[0])
            return igapi.android.AccountInfo.parse(row[0])
        except Exception:
            return None
    return None

async def get_client(username: str, password: str) -> igapi.android.Client:
    """从数据库恢复 session,失败则重新登录"""
    init_db(DB_FILE)

    account = load_session(DB_FILE, username, "android")
    if account:
        client = igapi.android.Client(account=account)
        if await client.is_logged_in():
            return client

    # 重新登录
    client = igapi.android.Client()
    await client.login(username, password)
    save_session(DB_FILE, username, "android", client.export_account_string())
    return client

async def main():
    client = await get_client(
        os.environ["IG_USERNAME"],
        os.environ["IG_PASSWORD"],
    )

asyncio.run(main())

环境变量存储

适用于容器化部署(Docker、Kubernetes)或无状态服务:

import asyncio
import igapi
import os

async def get_client() -> igapi.android.Client:
    """从环境变量恢复 session,适用于容器化部署"""
    session_str = os.environ.get("IG_SESSION_STRING")

    if session_str:
        try:
            account = igapi.android.AccountInfo.parse(session_str)
            client = igapi.android.Client(account=account)
            if await client.is_logged_in():
                return client
        except Exception as e:
            print(f"环境变量中的 session 无效,将重新登录: {e}")

    # 重新登录
    client = igapi.android.Client()
    await client.login(
        os.environ["IG_USERNAME"],
        os.environ["IG_PASSWORD"],
    )

    # 打印新 session 供用户更新环境变量
    new_session = client.export_account_string()
    print(f"新 session(请更新 IG_SESSION_STRING 环境变量):")
    print(new_session)

    return client

async def main():
    client = await get_client()

asyncio.run(main())

Session 有效期和刷新

Session 有效期

Instagram session 没有固定的过期时间,但以下情况会导致 session 失效:

原因 说明
密码被修改 修改密码后,所有现有 session 立即失效
账号在其他设备强制退出 「退出所有设备」操作会使全部 session 失效
长时间未使用 通常数周至数月后自动失效(具体时间不固定)
账号被封禁或受限 Instagram 安全机制触发时

检测 Session 失效

is_logged_in() 仅检查内存中是否存在 session 数据,不会发起网络请求。 若需验证 session 在服务端是否仍然有效,可尝试调用一个实际 API:

import asyncio
import igapi

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

    # 通过实际 API 调用验证 session 有效性
    try:
        user_api = igapi.UserApi(client)
        user = await user_api.info_by_username("instagram")
        print(f"session 有效,当前用户粉丝数: {user.follower_count}")
    except PermissionError as e:
        print(f"session 已失效,需重新登录: {e}")
        # 重新登录逻辑...

asyncio.run(main())

处理 Session 失效的推荐模式

import asyncio
import igapi
import os

SESSION_FILE = "session.txt"

async def login_fresh(username: str, password: str) -> igapi.android.Client:
    """全新登录并保存 session"""
    client = igapi.android.Client()
    await client.login(username, password)
    with open(SESSION_FILE, "w") as f:
        f.write(client.export_account_string())
    return client

async def get_client() -> igapi.android.Client:
    """获取有效客户端,session 失效时自动重新登录"""
    username = os.environ["IG_USERNAME"]
    password = os.environ["IG_PASSWORD"]

    if os.path.exists(SESSION_FILE):
        with open(SESSION_FILE) as f:
            saved = f.read().strip()
        try:
            account = igapi.android.AccountInfo.parse(saved)
            client = igapi.android.Client(account=account)
            return client
        except Exception:
            pass  # 解析失败,重新登录

    return await login_fresh(username, password)

async def with_auto_relogin(client: igapi.android.Client, func):
    """执行 API 调用,遇到 session 失效时自动重新登录并重试"""
    try:
        return await func(client)
    except PermissionError:
        print("session 失效,正在重新登录...")
        new_client = await login_fresh(
            os.environ["IG_USERNAME"],
            os.environ["IG_PASSWORD"],
        )
        return await func(new_client)

async def main():
    client = await get_client()

    # 使用自动重登录包装器
    result = await with_auto_relogin(
        client,
        lambda c: igapi.UserApi(c).info_by_username("instagram")
    )

asyncio.run(main())

Session 管理流程图

下图展示了 Session 完整的生命周期,从首次登录到失效处理:

flowchart TD
    A([程序启动]) --> B{是否存在已保存的 session}

    B -->|是| C[加载 session 字符串]
    B -->|否| D[创建 Client 并调用 login]

    C --> E[AccountInfo.parse 解析字符串]
    E -->|解析失败| F[session 格式无效]
    F --> D

    E -->|解析成功| G["Client(account=account) 恢复"]
    G --> H{client.is_logged_in}

    H -->|False| D
    H -->|True| I([客户端就绪,调用 API])

    D -->|登录成功| J[export_account_string 导出 session]
    J --> K[持久化存储]
    K --> I

    D -->|TwoFactorRequired| L[处理 2FA 验证]
    L --> J

    D -->|其他异常| M([登录失败,处理错误])

    I --> N{API 调用结果}
    N -->|成功| O([继续使用])
    N -->|"PermissionError(session 失效)"| P[清除本地 session]
    P --> D

常见问题

Q: AccountInfo.parse() 还需要传 platform 参数吗?

A: Python 用户层不传 platform 参数,而是通过平台命名空间表达:Android 使用 igapi.android.AccountInfo.parse(saved_string),Web 使用 igapi.web.AccountInfo.parse(saved_string)。两种账号字符串的设备字段不同,不能混用。

Q: has_sessionTrue 但 API 调用报 PermissionError,是什么原因?

A: has_session 仅表示内存中存在 session 数据(session_id 不为空),不代表服务端仍然认可该 session。session 可能在以下情况下已在服务端失效:密码被修改、账号在其他设备退出、长时间未使用等。遇到此情况应重新调用 login()

Q: session 字符串中包含明文密码,安全吗?

A: 出于恢复登录能力的需要(session 失效后自动重新登录),session 字符串中确实包含明文密码。请务必:不要将 session 字符串提交到 Git 仓库;存储到文件时设置适当的文件权限(如 chmod 600);存储到数据库时考虑加密;不要在日志中打印完整的 session 字符串。

Q: Android 和 Web 的 session 可以互换吗?

A: 不可以。两种 session 的设备信息字段完全不同(Android 有 android_id/phone_id/uuid/device_id,Web 有 csrf_token),且各自的 AccountInfo 只能传给对应类型的客户端。混用会在初始化时抛出 ValueError

Q: 多个进程同时使用同一个 session 字符串,会有问题吗?

A: 读取同一 session 字符串的多个客户端实例通常可以并行使用,因为 Instagram 允许同一账号同时保持多个活跃 session。但应避免多个进程同时写入同一个 session 文件,否则可能导致文件内容损坏。建议使用数据库存储来支持多进程场景。

Q: 如何安全地删除一个已保存的 session?

A: 直接删除保存 session 的文件或数据库记录即可。下次需要使用时,程序会自动重新登录。如果需要同时使服务端 session 失效(防止 session 泄露后被滥用),可以在 Instagram 官方客户端中退出对应设备的登录状态。