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.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.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())

WebClient 导出

WebClient 的导出方式与 Client 相同：

import asyncio
import igapi

async def main():
    web = igapi.WebClient()
    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.AccountInfo.parse(saved_string, platform="android")

    # 传入 account 参数，无需再调用 login()
    client = igapi.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.AccountInfo.parse(saved_string, platform="android")

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

WebClient 从字符串恢复

import igapi

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

account = igapi.AccountInfo.parse(saved_string, platform="web")
web = igapi.WebClient(account=account)

注意：Android 和 Web 的 session 字符串格式不同，不能混用。 AccountInfo.parse() 的 platform 参数必须与字符串实际来源平台一致， 且 AccountInfo 的平台类型必须与客户端类型匹配（Android AccountInfo 不能传给 WebClient，反之亦然）。

---

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 方法

方法	返回值	说明
AccountInfo.parse(s, platform)	AccountInfo	从字符串解析，platform 为 "android" 或 "web"
account.to_account_string()	str	将 AccountInfo 序列化为字符串

---

Session 字符串格式

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

Android 格式

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

示例：

alice:MyPassword123||android-a1b2c3d4e5f6a7b8;550e8400-e29b-41d4-a716-446655440000;6ba7b810-9dad-11d1-80b4-00c04fd430c8;6ba7b814-9dad-11d1-80b4-00c04fd430c8|sessionid=ABC123; ds_user_id=123456789; csrftoken=XYZ789; mid=DEF456||

字段说明：

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

Web 格式

用户名:密码||csrf;;;|cookies||

示例：

alice:MyPassword123||XYZ789CSRFTOKEN;;;|sessionid=ABC123; ds_user_id=123456789; csrftoken=XYZ789; mid=DEF456||

字段说明：

字段位置	内容	说明
用户名:密码	明文凭证	以 : 分隔
csrf	CSRF Token 字符串	Web 请求验证使用的 Token
;;;	三个空字段	Web 无需设备信息，占位符
cookies	key=value; key=value	多个 cookie 以 ; 分隔

必填 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.Client:
    """获取已登录的客户端，优先从 session 文件恢复"""
    # 尝试从文件恢复
    if os.path.exists(SESSION_FILE):
        with open(SESSION_FILE, "r") as f:
            saved = f.read().strip()

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

    # 全新登录
    client = igapi.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) -> igapi.AccountInfo | None:
    """从数据库加载 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:
            return igapi.AccountInfo.parse(row[0], platform=platform)
        except Exception:
            return None
    return None

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

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

    # 重新登录
    client = igapi.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.Client:
    """从环境变量恢复 session，适用于容器化部署"""
    session_str = os.environ.get("IG_SESSION_STRING")

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

    # 重新登录
    client = igapi.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.Client(account=account)

    # 通过实际 API 调用验证 session 有效性
    try:
        user = await client.user().info_by_name("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.Client:
    """全新登录并保存 session"""
    client = igapi.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.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.AccountInfo.parse(saved, platform="android")
            client = igapi.Client(account=account)
            return client
        except Exception:
            pass  # 解析失败，重新登录

    return await login_fresh(username, password)

async def with_auto_relogin(client: igapi.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: c.user().info_by_name("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: 是的，必须指定。igapi 无法从字符串内容自动判断平台类型，platform 参数接受 "android" 或 "web" 两个值。传入错误的平台值会导致 session 字段解析错误，进而在 API 调用时出现认证失败。

Q: has_session 为 True 但 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 官方客户端中退出对应设备的登录状态。