diff --git a/dual-link-proxy/scripts/3xui-api.py b/dual-link-proxy/scripts/3xui-api.py index dde2ffc..acc2f43 100644 --- a/dual-link-proxy/scripts/3xui-api.py +++ b/dual-link-proxy/scripts/3xui-api.py @@ -18,6 +18,7 @@ import re import sys import urllib.parse import urllib.request +import uuid as uuidlib def opener_with_cookies(): @@ -170,6 +171,79 @@ def cmd_add(op, token, base, args): % (obj.get("id"), args.remark, args.port, args.path)) +def fetch_inbounds(op, base): + raw = http_get(op, base + "/panel/api/inbounds/list") + return json.loads(raw).get("obj") or [] + + +def save_inbound(op, token, base, ib): + raw = http_post( + op, + base + "/panel/api/inbounds/update/%s" % ib.get("id"), + js=ib, + headers={"X-CSRF-Token": token, "Origin": base, + "Referer": base + "/panel/inbounds"}, + ) + data = json.loads(raw) + if not data.get("success"): + sys.exit("更新入站 %s 失败:%s" % (ib.get("remark"), data.get("msg"))) + + +def api_post(op, token, base, path, js, referer="/panel/clients"): + raw = http_post( + op, base + path, js=js, + headers={"X-CSRF-Token": token, "Origin": base, "Referer": base + referer}, + ) + try: + return json.loads(raw) + except json.JSONDecodeError: + sys.exit("接口 %s 返回非 JSON:%s" % (path, raw[:200])) + + +def target_inbound_ids(items, only): + ids = [ib.get("id") for ib in items + if not only or ib.get("remark") == only] + if not ids: + sys.exit("没有匹配的入站(检查 --inbound 的备注名)") + return ids + + +def cmd_add_user(op, token, base, args): + items = fetch_inbounds(op, base) + if not items: + sys.exit("没有可用的入站") + ids = target_inbound_ids(items, args.inbound) + new_uuid = args.uuid or str(uuidlib.uuid4()) + payload = { + "client": {"email": args.email, "uuid": new_uuid, "enable": True}, + "inboundIds": ids, + } + data = api_post(op, token, base, "/panel/api/clients/add", payload) + if not data.get("success"): + sys.exit("建号失败:%s" % data.get("msg")) + # v3 的服务端会自行生成 UUID,这里回读真实值,避免打印出来的和实际不一致 + actual = new_uuid + for ib in fetch_inbounds(op, base): + for c in (as_obj(ib.get("settings")).get("clients") or []): + if c.get("email") == args.email: + actual = c.get("id") or actual + break + names = [ib.get("remark") for ib in items if ib.get("id") in ids] + print("账号 %s 已创建,UUID = %s" % (args.email, actual)) + print("已挂到入站:" + "、".join(names)) + + +def cmd_del_user(op, token, base, args): + data = api_post( + op, token, base, + "/panel/api/clients/del/" + urllib.parse.quote(args.email), + {}, + ) + if not data.get("success"): + sys.exit("销号失败:%s" % data.get("msg")) + print("账号 %s 已删除(从所有入站一并摘除)" % args.email) + + def main(): p = argparse.ArgumentParser(description="3x-ui 命令行封装") p.add_argument("--base", default="http://127.0.0.1:2053") @@ -188,6 +262,15 @@ def main(): a.add_argument("--listen", default="127.0.0.1") a.add_argument("--email", default=None) + u = sub.add_parser("add-user", help="给所有入站加同一个用户,一人两条链路") + u.add_argument("--email", required=True) + u.add_argument("--uuid", default=None) + u.add_argument("--inbound", default=None, help="只加到指定备注的入站") + + d = sub.add_parser("del-user", help="从所有入站移除该用户") + d.add_argument("--email", required=True) + d.add_argument("--inbound", default=None) + args = p.parse_args() if not args.cmd: p.error("缺少子命令:ping / list / add") @@ -198,6 +281,10 @@ def main(): cmd_list(op, token, args.base) elif args.cmd == "add": cmd_add(op, token, args.base, args) + elif args.cmd == "add-user": + cmd_add_user(op, token, args.base, args) + elif args.cmd == "del-user": + cmd_del_user(op, token, args.base, args) if __name__ == "__main__":