New API 批量删除所有用户

最近对我的自用 New API 进行升级，由老版升级到新版 UI，升级后过了一会儿发现了几百个机器人用户。到设置一看果然开放了注册，赶紧关闭并且开始着手清理账号。由于现在 NewAPI 的 UI 上没有批量删除用户的功能，于是我让 AI 根据官方 API 写了一个 python 脚本，该脚本会删除所有用户，仅保留管理员账号。

建议修改配置后，让 AI 来运行。

#!/usr/bin/env python3
"""
New API 批量删除用户脚本
删除所有普通用户，仅保留管理员账号（role >= 100）

使用方法:
  python delete_users.py

首次运行前请修改下方配置项。
"""

import requests
import time
import sys

# ============ 配置项 ============
BASE_URL = "http://lcoalhost:3000"  # New API 地址
ACCESS_TOKEN = ""                   # 管理员 access_token
ADMIN_USER_ID = ""                  # 留空则自动从 API 获取
PAGE_SIZE = 100                     # 每页用户数
DRY_RUN = False                     # True = 仅预览不删除，False = 实际删除
# ================================

HEADERS = {
    "Authorization": ACCESS_TOKEN,
    "New-Api-User": ADMIN_USER_ID,
    "Content-Type": "application/json",
}


def resolve_admin_id():
    """尝试用 access_token 配合不同 user_id，自动找到可用的管理员 ID"""
    for uid in range(1, 20):
        headers = {
            "Authorization": ACCESS_TOKEN,
            "New-Api-User": str(uid),
            "Content-Type": "application/json",
        }
        try:
            resp = requests.get(
                f"{BASE_URL}/api/user/",
                headers=headers,
                params={"p": 1, "page_size": 1},
                timeout=10,
            )
            if resp.status_code == 200:
                body = resp.json()
                if body.get("success"):
                    print(f"检测到有效凭据: uid={uid}")
                    return str(uid)
        except Exception:
            continue
    return None


def get_all_users():
    """分页获取所有用户"""
    users = []
    page = 1
    while True:
        resp = requests.get(
            f"{BASE_URL}/api/user/",
            headers=HEADERS,
            params={"p": page, "page_size": PAGE_SIZE},
        )
        resp.raise_for_status()
        body = resp.json()

        if not body.get("success"):
            print(f"获取用户列表失败: {body.get('message', '未知错误')}")
            sys.exit(1)

        data = body.get("data", {})
        items = data.get("items", [])
        users.extend(items)

        total = data.get("total", 0)
        if len(users) >= total or len(items) < PAGE_SIZE:
            break
        page += 1

    return users


def delete_user(user_id, username):
    """删除单个用户"""
    resp = requests.delete(
        f"{BASE_URL}/api/user/{user_id}",
        headers=HEADERS,
    )
    # 已知 bug: 成功删除返回空 body，status 200 即视为成功
    if resp.status_code == 200:
        return True
    try:
        body = resp.json()
        print(f"  删除失败 [{username}]: {body.get('message', resp.text)}")
    except Exception:
        print(f"  删除失败 [{username}]: HTTP {resp.status_code} - {resp.text}")
    return False


def main():
    if not ACCESS_TOKEN:
        print("请先配置 ACCESS_TOKEN")
        sys.exit(1)

    # 自动检测管理员 ID
    admin_user_id = ADMIN_USER_ID
    if not admin_user_id:
        print("ADMIN_USER_ID 未填写，尝试自动检测...\n")
        detected = resolve_admin_id()
        if detected:
            admin_user_id = detected
            HEADERS["New-Api-User"] = admin_user_id
            print(f"已自动设置 ADMIN_USER_ID = {admin_user_id}\n")
        else:
            print("无法自动检测管理员 ID，请手动填写 ADMIN_USER_ID")
            sys.exit(1)

    print(f"连接 {BASE_URL} ...")
    users = get_all_users()
    print(f"共获取 {len(users)} 个用户\n")

    to_delete = []
    protected = []

    for u in users:
        username = u.get("username", "")
        uid = u.get("id")
        role = u.get("role", 0)
        if role >= 100:  # 管理员(100) + 超级管理员(1000) 都保留
            protected.append(u)
        else:
            to_delete.append(u)

    print(f"受保护（跳过）: {len(protected)} 个")
    for p in protected:
        print(f"  - {p.get('username')} (id={p.get('id')}, role={p.get('role')})")

    print(f"\n待删除: {len(to_delete)} 个")
    for u in to_delete:
        print(f"  - {u.get('username')} (id={u.get('id')}, role={u.get('role')})")

    if not to_delete:
        print("\n没有需要删除的用户。")
        return

    if DRY_RUN:
        print(f"\n[预览模式] 设置 DRY_RUN = False 后执行实际删除。")
        return

    print(f"\n确认删除以上 {len(to_delete)} 个用户？(y/N): ", end="")
    confirm = "y"
    if confirm != "y":
        print("已取消。")
        return

    success = 0
    failed = 0
    for i, u in enumerate(to_delete, 1):
        username = u.get("username", "")
        uid = u.get("id")
        print(f"[{i}/{len(to_delete)}] 删除 {username} (id={uid}) ... ", end="", flush=True)
        if delete_user(uid, username):
            print("OK")
            success += 1
        else:
            failed += 1
        time.sleep(0.2)  # 避免请求过快

    print(f"\n完成: 成功 {success}, 失败 {failed}")


if __name__ == "__main__":
    main()

1 个帖子 - 1 位参与者

阅读完整话题