dekun/crypto_key
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加多账户

Browse Source
This commit is contained in:
dekun
2026-05-27 07:34:34 +08:00
parent a5383d8ca1
commit 1f695799bf
8 changed files with 1371 additions and 770 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+5
View File
@@ -0,0 +1,5 @@
# 复制为 .env 后修改（勿提交 .env 到 Git）
SECRET_KEY=请替换为随机长字符串
AUTH_USERNAME=admin
AUTH_PASSWORD=请设置强密码
DATA_ENCRYPTION_KEY=请运行 python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
.gitignore
+6 -4
View File
@@ -1,4 +1,6 @@
venv/ venv/
logs/*.log logs/*.log
data.json data.json
__pycache__/ settings.json
.env
__pycache__/
README.md
+130 -116
View File
@@ -1,116 +1,130 @@
# crypto_key — 币圈 API 密钥管理工具 # crypto_key — 凭证保险库
本地 Web 工具，管理多个交易所/平台 API 账户（账户名称、API Key、API Secret）。数据保存在本地 `data.json`，无登录。 本地 Web 工具，管理交易所 API、网站、邮箱、机器人、微信/QQ 及自定义类型凭证。数据 **AES 加密** 存入 `data.json`，登录凭据与加密主密钥在 `.env`。
| 项目 | 说明 | | 项目 | 说明 |
|------|------| |------|------|
| 仓库 | [git.bz121.com/dekun/crypto_key](https://git.bz121.com/dekun/crypto_key.git) | | 仓库 | [git.bz121.com/dekun/crypto_key](https://git.bz121.com/dekun/crypto_key.git) |
| Ubuntu 部署路径 | `/opt/crypto_key` | | Ubuntu 部署路径 | `/opt/crypto_key` |
| 服务端口 | `5200` | | 服务端口 | `5200` |
| 进程守护 | PM2 | | 进程守护 | PM2 |
--- ---
## 功能 ## 功能
- 交易所：`binance` / `okx` / `gate`（OKX 额外保存密码 Passphrase） - **登录**：用户名 + 密码（`.env`），浏览器不缓存明文
- 账户数量不限，每项含 `exchange` / `username` / `api_key` / `api_secret` - **内置类型**：交易所 API、网站、邮箱、企业微信/钉钉机器人、微信、QQ
- 添加后默认不展示列表，需选择交易所点击「确认」查询 - **自定义类型**：系统设置中添加（如小红书、抖音、快手）
- 黑色专业界面，列表展示 + 每项 3 个复制按钮（**复制始终为明文**） - **查询**：类型搜索 + 关键词 + 确认后显示；网站/邮箱可点击跳转
- 可选界面打码显示，不影响复制内容 - **旧 data.json**：首次启动自动迁移为加密格式
- 数据持久化至 `data.json` - 黑色专业界面，列表展示 + 每项 3 个复制按钮（**复制始终为明文**）
- 可选界面打码显示，不影响复制内容
--- - 数据持久化至 `data.json`
## Ubuntu 服务器部署（/opt） ---
完整步骤见 **[DEPLOY.md](./DEPLOY.md)**。 ## 首次配置
```bash ```bash
sudo git clone https://git.bz121.com/dekun/crypto_key.git /opt/crypto_key cp .env.example .env
cd /opt/crypto_key # 编辑 .env：SECRET_KEY、AUTH_USERNAME、AUTH_PASSWORD、DATA_ENCRYPTION_KEY
sudo bash scripts/install-ubuntu.sh pip install -r requirements.txt
``` python app.py
```
访问：
默认首次运行会自动生成 `.env`（用户名 `admin` / 密码 `admin123`，请立即在「系统设置」中修改）。
- 本机：**http://127.0.0.1:5200**
- 局域网：**http://<服务器局域网IP>:5200**（如 `http://192.168.1.100:5200`） ## Ubuntu 服务器部署（/opt）
查看本机 IP：`hostname -I` 或 `ip addr` 完整步骤见 **[DEPLOY.md](./DEPLOY.md)**。
```bash ```bash
pm2 status sudo git clone https://git.bz121.com/dekun/crypto_key.git /opt/crypto_key
pm2 logs api-key-manager cd /opt/crypto_key
``` cp .env.example .env # 并编辑
sudo bash scripts/install-ubuntu.sh
--- ```
## 本地开发 访问：
```bash - 本机：**http://127.0.0.1:5200**
python3 -m venv venv - 局域网：**http://<服务器局域网IP>:5200**（如 `http://192.168.1.100:5200`）
source venv/bin/activate # Windows: venv\Scripts\activate
pip install -r requirements.txt 查看本机 IP：`hostname -I` 或 `ip addr`
python app.py
``` ```bash
pm2 status
浏览器打开 http://127.0.0.1:5200（局域网内其他设备用 `http://<本机IP>:5200`） pm2 logs api-key-manager
```
PM2 守护：`./pm2-start.sh`（Linux）或 `.\pm2-start.ps1`（Windows）
---
---
## 本地开发
## 文件说明
```bash
| 文件 | 说明 | python3 -m venv venv
|------|------| source venv/bin/activate # Windows: venv\Scripts\activate
| `app.py` | Flask 后端 | pip install -r requirements.txt
| `index.html` | 前端页面 | python app.py
| `ecosystem.config.cjs` | PM2 配置 | ```
| `scripts/install-ubuntu.sh` | Ubuntu 一键安装 |
| `DEPLOY.md` | 完整部署文档 | 浏览器打开 http://127.0.0.1:5200（局域网内其他设备用 `http://<本机IP>:5200`）
--- PM2 守护：`./pm2-start.sh`（Linux）或 `.\pm2-start.ps1`（Windows）
## 数据格式 ---
```json ## 文件说明
{
"id": "uuid", | 文件 | 说明 |
"exchange": "binance", |------|------|
"username": "账户名称", | `app.py` | Flask 后端 |
"api_key": "API Key", | `index.html` | 前端页面 |
"api_secret": "API Secret", | `ecosystem.config.cjs` | PM2 配置 |
"password": "仅 OKX 需要" | `scripts/install-ubuntu.sh` | Ubuntu 一键安装 |
} | `DEPLOY.md` | 完整部署文档 |
```
---
---
## 数据格式
## API
```json
| 方法 | 路径 | 说明 | {
|------|------|------| "id": "uuid",
| GET | `/` | 前端页面 | "exchange": "binance",
| GET | `/api/accounts` | 获取全部账户 | "username": "账户名称",
| GET | `/api/accounts?exchange=binance` | 按交易所筛选 | "api_key": "API Key",
| POST | `/api/accounts` | 新增账户 | "api_secret": "API Secret",
| DELETE | `/api/accounts/<id>` | 删除账户 | "password": "仅 OKX 需要"
}
--- ```
## 安全提示 ---
- 无认证，监听 `0.0.0.0`（局域网可访问），请勿暴露到公网；建议仅在可信局域网使用 ## API
- `data.json` 为明文密钥，已 `.gitignore`，请定期备份
| 方法 | 路径 | 说明 |
--- |------|------|------|
| GET | `/` | 前端页面 |
## 依赖 | GET | `/api/accounts` | 获取全部账户 |
| GET | `/api/accounts?exchange=binance` | 按交易所筛选 |
``` | POST | `/api/accounts` | 新增账户 |
flask>=3.0.0,<4.0.0 | DELETE | `/api/accounts/<id>` | 删除账户 |
```
---
服务器另需：Python 3.8+、Node.js（PM2）
## 安全提示
- 无认证，监听 `0.0.0.0`（局域网可访问），请勿暴露到公网；建议仅在可信局域网使用
- `data.json` 为明文密钥，已 `.gitignore`，请定期备份
---
## 依赖
```
flask>=3.0.0,<4.0.0
```
服务器另需：Python 3.8+、Node.js（PM2）
app.py
+226 -108
View File
@@ -1,108 +1,226 @@
""" """
多账户 API 密钥管理工具 — Flask 后端 凭证保险库 — Flask 后端
端口: 5200 (0.0.0.0 局域网可访问) | 数据: data.json 登录凭据与加密主密钥在 .env | 数据加密存 data.json
""" """
import json import re
import uuid import uuid
from pathlib import Path from functools import wraps
from flask import Flask, jsonify, request, send_from_directory from flask import Flask, jsonify, request, send_from_directory, session
app = Flask(__name__) import env_config
from storage import (
BASE_DIR = Path(__file__).resolve().parent BUILTIN_TYPES,
DATA_FILE = BASE_DIR / "data.json" all_type_defs,
VALID_EXCHANGES = frozenset({"binance", "okx", "gate"}) filter_records,
get_type_def,
load_records,
def load_accounts(): load_settings,
if not DATA_FILE.exists(): save_records,
return [] save_settings,
try: validate_record_payload,
with open(DATA_FILE, "r", encoding="utf-8") as f: )
data = json.load(f)
if isinstance(data, list): env_config.ensure_env()
return data
return [] app = Flask(__name__)
except (json.JSONDecodeError, OSError): app.secret_key = env_config.get_secret_key()
return [] app.config.update(
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SAMESITE="Lax",
def save_accounts(accounts): PERMANENT_SESSION_LIFETIME=1800,
with open(DATA_FILE, "w", encoding="utf-8") as f: )
json.dump(accounts, f, ensure_ascii=False, indent=2)
BASE_DIR = env_config.BASE_DIR
PUBLIC_API = {"/api/auth/login", "/api/auth/status"}
def validate_account_payload(payload):
exchange = (payload.get("exchange") or "").strip().lower()
username = (payload.get("username") or "").strip() def login_required(fn):
api_key = (payload.get("api_key") or "").strip() @wraps(fn)
api_secret = (payload.get("api_secret") or "").strip() def wrapper(*args, **kwargs):
if not session.get("authenticated"):
if exchange not in VALID_EXCHANGES: return jsonify({"error": "未登录"}), 401
return None, "请选择交易所：binance / okx / gate" return fn(*args, **kwargs)
if not username:
return None, "账户名称不能为空" return wrapper
if not api_key:
return None, "API Key 不能为空"
if not api_secret: @app.before_request
return None, "API Secret 不能为空" def guard_api():
if not request.path.startswith("/api/"):
account = { return None
"exchange": exchange, if request.path in PUBLIC_API:
"username": username, return None
"api_key": api_key, if not session.get("authenticated"):
"api_secret": api_secret, return jsonify({"error": "未登录"}), 401
} return None
if exchange == "okx":
password = (payload.get("password") or "").strip() @app.route("/")
if not password: def index():
return None, "OKX 密码（Passphrase）不能为空" return send_from_directory(BASE_DIR, "index.html")
account["password"] = password
return account, None @app.route("/api/auth/status", methods=["GET"])
def auth_status():
return jsonify({"logged_in": bool(session.get("authenticated"))})
@app.route("/")
def index():
return send_from_directory(BASE_DIR, "index.html") @app.route("/api/auth/login", methods=["POST"])
def auth_login():
body = request.get_json(silent=True) or {}
@app.route("/api/accounts", methods=["GET"]) username = (body.get("username") or "").strip()
def list_accounts(): password = body.get("password") or ""
accounts = load_accounts() if username != env_config.get_auth_username() or not env_config.verify_password(password):
exchange = (request.args.get("exchange") or "").strip().lower() return jsonify({"error": "用户名或密码错误"}), 401
if exchange: session["authenticated"] = True
if exchange not in VALID_EXCHANGES: session.permanent = True
return jsonify({"error": "无效交易所"}), 400 return jsonify({"ok": True})
accounts = [a for a in accounts if a.get("exchange") == exchange]
return jsonify(accounts)
@app.route("/api/auth/logout", methods=["POST"])
@login_required
@app.route("/api/accounts", methods=["POST"]) def auth_logout():
def create_account(): session.clear()
body = request.get_json(silent=True) or {} return jsonify({"ok": True})
account, err = validate_account_payload(body)
if err:
return jsonify({"error": err}), 400 @app.route("/api/settings", methods=["GET"])
@login_required
accounts = load_accounts() def get_settings():
account["id"] = str(uuid.uuid4()) s = load_settings()
accounts.append(account) return jsonify(
save_accounts(accounts) {
return jsonify(account), 201 "builtin_types": BUILTIN_TYPES,
"custom_types": s.get("custom_types", []),
"username": env_config.get_auth_username(),
@app.route("/api/accounts/<account_id>", methods=["DELETE"]) }
def delete_account(account_id): )
accounts = load_accounts()
new_accounts = [a for a in accounts if a.get("id") != account_id]
if len(new_accounts) == len(accounts): @app.route("/api/settings/auth", methods=["PUT"])
return jsonify({"error": "账户不存在"}), 404 @login_required
save_accounts(new_accounts) def update_settings_auth():
return jsonify({"ok": True}) body = request.get_json(silent=True) or {}
current = body.get("current_password") or ""
if not env_config.verify_password(current):
if __name__ == "__main__": return jsonify({"error": "当前密码错误"}), 400
app.run(host="0.0.0.0", port=5200, debug=False) username = (body.get("username") or "").strip()
new_pass = body.get("new_password") or ""
if not username:
return jsonify({"error": "用户名不能为空"}), 400
if len(new_pass) < 6:
return jsonify({"error": "新密码至少 6 位"}), 400
env_config.update_auth(username, new_pass)
return jsonify({"ok": True, "username": username})
@app.route("/api/settings/types", methods=["POST"])
@login_required
def add_custom_type():
body = request.get_json(silent=True) or {}
type_id = (body.get("id") or "").strip().lower()
label = (body.get("label") or "").strip()
fields = body.get("fields") or []
if not re.match(r"^[a-z][a-z0-9_]{1,31}$", type_id):
return jsonify({"error": "类型 ID 需为小写字母开头，仅含 a-z0-9_"}), 400
if not label:
return jsonify({"error": "显示名称不能为空"}), 400
if get_type_def(type_id):
return jsonify({"error": "类型 ID 已存在"}), 400
if not fields:
return jsonify({"error": "至少添加一个字段"}), 400
norm_fields = []
for f in fields:
key = (f.get("key") or "").strip().lower()
if not re.match(r"^[a-z][a-z0-9_]{0,31}$", key):
return jsonify({"error": f"无效字段 key: {key}"}), 400
norm_fields.append(
{
"key": key,
"label": (f.get("label") or key).strip(),
"type": f.get("type") or "text",
"required": bool(f.get("required")),
}
)
s = load_settings()
entry = {"id": type_id, "label": label, "builtin": False, "fields": norm_fields}
s["custom_types"].append(entry)
save_settings(s)
return jsonify(entry), 201
@app.route("/api/settings/types/<type_id>", methods=["DELETE"])
@login_required
def delete_custom_type(type_id):
s = load_settings()
before = len(s["custom_types"])
s["custom_types"] = [t for t in s["custom_types"] if t["id"] != type_id]
if len(s["custom_types"]) == before:
return jsonify({"error": "自定义类型不存在"}), 404
save_settings(s)
records = [r for r in load_records() if r.get("type_id") != type_id]
save_records(records)
return jsonify({"ok": True})
@app.route("/api/credentials", methods=["GET"])
@login_required
def list_credentials():
records = load_records()
type_id = (request.args.get("type_id") or "").strip()
q = (request.args.get("q") or "").strip()
if type_id == "exchange" and request.args.get("exchange"):
ex = request.args.get("exchange").strip().lower()
records = [
r
for r in records
if r.get("type_id") == "exchange" and r.get("fields", {}).get("exchange") == ex
]
else:
records = filter_records(records, type_id=type_id or None, q=q or None)
return jsonify(records)
@app.route("/api/credentials", methods=["POST"])
@login_required
def create_credential():
body = request.get_json(silent=True) or {}
record, err = validate_record_payload(body)
if err:
return jsonify({"error": err}), 400
record["id"] = str(uuid.uuid4())
records = load_records()
records.append(record)
save_records(records)
return jsonify(record), 201
@app.route("/api/credentials/<record_id>", methods=["DELETE"])
@login_required
def delete_credential(record_id):
records = load_records()
new_records = [r for r in records if r.get("id") != record_id]
if len(new_records) == len(records):
return jsonify({"error": "记录不存在"}), 404
save_records(new_records)
return jsonify({"ok": True})
# 兼容旧前端路径
@app.route("/api/accounts", methods=["GET", "POST"])
@login_required
def accounts_compat():
if request.method == "GET":
return list_credentials()
return create_credential()
@app.route("/api/accounts/<account_id>", methods=["DELETE"])
@login_required
def accounts_delete_compat(account_id):
return delete_credential(account_id)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5200, debug=False)
env_config.py
+81
View File
@@ -0,0 +1,81 @@
"""读取与更新 .env 配置"""
import os
import re
from pathlib import Path
from cryptography.fernet import Fernet
from dotenv import load_dotenv, set_key
from werkzeug.security import check_password_hash, generate_password_hash
BASE_DIR = Path(__file__).resolve().parent
ENV_FILE = BASE_DIR / ".env"
def _valid_fernet_key(value: str) -> bool:
if not value or "请" in value or len(value) < 40:
return False
try:
Fernet(value.encode())
return True
except (ValueError, Exception):
return False
def ensure_env():
if not ENV_FILE.exists():
example = BASE_DIR / ".env.example"
if example.exists():
ENV_FILE.write_text(example.read_text(encoding="utf-8"), encoding="utf-8")
else:
ENV_FILE.write_text("", encoding="utf-8")
load_dotenv(ENV_FILE, override=True)
changed = False
if not os.getenv("SECRET_KEY") or "请" in os.getenv("SECRET_KEY", ""):
set_key(str(ENV_FILE), "SECRET_KEY", os.urandom(32).hex())
changed = True
if not _valid_fernet_key(os.getenv("DATA_ENCRYPTION_KEY", "")):
set_key(str(ENV_FILE), "DATA_ENCRYPTION_KEY", Fernet.generate_key().decode())
changed = True
if not os.getenv("AUTH_USERNAME"):
set_key(str(ENV_FILE), "AUTH_USERNAME", "admin")
changed = True
if not os.getenv("AUTH_PASSWORD"):
set_key(str(ENV_FILE), "AUTH_PASSWORD", generate_password_hash("admin123"))
changed = True
if changed:
load_dotenv(ENV_FILE, override=True)
def reload_env():
load_dotenv(ENV_FILE, override=True)
def get_secret_key():
return os.getenv("SECRET_KEY", "dev-insecure-key")
def get_encryption_key():
raw = os.getenv("DATA_ENCRYPTION_KEY", "")
if not raw:
raise ValueError("缺少 DATA_ENCRYPTION_KEY，请检查 .env")
return raw.encode() if isinstance(raw, str) else raw
def get_auth_username():
return os.getenv("AUTH_USERNAME", "admin")
def verify_password(plain: str) -> bool:
stored = os.getenv("AUTH_PASSWORD", "")
if not stored:
return False
if stored.startswith("pbkdf2:") or stored.startswith("scrypt:"):
return check_password_hash(stored, plain)
return stored == plain
def update_auth(username: str, password: str):
ensure_env()
set_key(str(ENV_FILE), "AUTH_USERNAME", username)
set_key(str(ENV_FILE), "AUTH_PASSWORD", generate_password_hash(password))
reload_env()
index.html
+639 -541
View File
File diff suppressed because it is too large Load Diff
requirements.txt
+4 -1
View File
@@ -1 +1,4 @@
flask>=3.0.0,<4.0.0 flask>=3.0.0,<4.0.0
python-dotenv>=1.0.0,<2.0.0
cryptography>=42.0.0,<44.0.0
werkzeug>=3.0.0,<4.0.0
storage.py
+280
View File
@@ -0,0 +1,280 @@
"""加密存储、旧版 data.json 迁移"""
import json
import uuid
from pathlib import Path
from cryptography.fernet import Fernet, InvalidToken
from env_config import get_encryption_key
BASE_DIR = Path(__file__).resolve().parent
DATA_FILE = BASE_DIR / "data.json"
SETTINGS_FILE = BASE_DIR / "settings.json"
VALID_EXCHANGES = frozenset({"binance", "okx", "gate"})
BUILTIN_TYPES = [
{
"id": "exchange",
"label": "交易所 API",
"builtin": True,
"fields": [
{"key": "exchange", "label": "交易所", "type": "select", "options": ["binance", "okx", "gate"]},
{"key": "username", "label": "账户名称", "type": "text", "required": True},
{"key": "api_key", "label": "API Key", "type": "secret", "required": True},
{"key": "api_secret", "label": "API Secret", "type": "secret", "required": True},
{"key": "password", "label": "OKX 密码", "type": "secret", "required": True, "when": {"exchange": "okx"}},
],
},
{
"id": "website",
"label": "网站账号",
"builtin": True,
"fields": [
{"key": "title", "label": "名称", "type": "text", "required": True},
{"key": "url", "label": "网址", "type": "url", "required": True},
{"key": "username", "label": "用户名", "type": "text", "required": True},
{"key": "password", "label": "密码", "type": "secret", "required": True},
{"key": "note", "label": "备注", "type": "text"},
],
},
{
"id": "email",
"label": "邮箱账户",
"builtin": True,
"fields": [
{"key": "title", "label": "名称", "type": "text", "required": True},
{"key": "email", "label": "邮箱地址", "type": "email", "required": True},
{"key": "username", "label": "用户名", "type": "text"},
{"key": "password", "label": "密码", "type": "secret", "required": True},
{"key": "web_url", "label": "网页登录地址", "type": "url"},
{"key": "note", "label": "备注", "type": "text"},
],
},
{
"id": "wecom_bot",
"label": "企业微信机器人",
"builtin": True,
"fields": [
{"key": "title", "label": "名称", "type": "text", "required": True},
{"key": "webhook", "label": "Webhook", "type": "secret", "required": True},
{"key": "note", "label": "备注", "type": "text"},
],
},
{
"id": "dingtalk_bot",
"label": "钉钉机器人",
"builtin": True,
"fields": [
{"key": "title", "label": "名称", "type": "text", "required": True},
{"key": "webhook", "label": "Webhook", "type": "secret", "required": True},
{"key": "note", "label": "备注", "type": "text"},
],
},
{
"id": "wechat",
"label": "微信",
"builtin": True,
"fields": [
{"key": "title", "label": "名称", "type": "text", "required": True},
{"key": "wechat_id", "label": "微信号", "type": "text", "required": True},
{"key": "phone", "label": "手机号", "type": "phone", "required": True},
{"key": "password", "label": "密码", "type": "secret"},
{"key": "note", "label": "备注", "type": "text"},
],
},
{
"id": "qq",
"label": "QQ",
"builtin": True,
"fields": [
{"key": "title", "label": "名称", "type": "text", "required": True},
{"key": "qq_number", "label": "QQ号", "type": "text", "required": True},
{"key": "phone", "label": "手机号", "type": "phone", "required": True},
{"key": "password", "label": "密码", "type": "secret"},
{"key": "note", "label": "备注", "type": "text"},
],
},
]
def _fernet():
return Fernet(get_encryption_key())
def default_settings():
return {"custom_types": []}
def load_settings():
if not SETTINGS_FILE.exists():
s = default_settings()
save_settings(s)
return s
try:
with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
if "custom_types" not in data:
data["custom_types"] = []
return data
except (json.JSONDecodeError, OSError):
return default_settings()
def save_settings(data):
with open(SETTINGS_FILE, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def all_type_defs():
settings = load_settings()
return BUILTIN_TYPES + settings.get("custom_types", [])
def get_type_def(type_id: str):
for t in all_type_defs():
if t["id"] == type_id:
return t
return None
def _legacy_to_record(item: dict) -> dict:
exchange = (item.get("exchange") or "binance").lower()
fields = {
"exchange": exchange,
"username": item.get("username", ""),
"api_key": item.get("api_key", ""),
"api_secret": item.get("api_secret", ""),
}
if item.get("password"):
fields["password"] = item["password"]
title = fields.get("username") or exchange
return {
"id": item.get("id") or str(uuid.uuid4()),
"type_id": "exchange",
"title": title,
"fields": fields,
}
def _read_raw_file():
if not DATA_FILE.exists():
return None
with open(DATA_FILE, "r", encoding="utf-8") as f:
return json.load(f)
def _migrate_plain_list(items: list) -> list:
records = []
for item in items:
if isinstance(item, dict) and item.get("type_id"):
records.append(item)
elif isinstance(item, dict) and item.get("exchange"):
records.append(_legacy_to_record(item))
elif isinstance(item, dict):
records.append(item)
return records
def load_records():
raw = _read_raw_file()
if raw is None:
return []
if isinstance(raw, list):
records = _migrate_plain_list(raw)
save_records(records)
return records
if isinstance(raw, dict) and raw.get("v") == 2:
token = raw.get("payload", "")
if not token:
return []
try:
plain = _fernet().decrypt(token.encode()).decode()
data = json.loads(plain)
return data if isinstance(data, list) else []
except (InvalidToken, json.JSONDecodeError):
return []
return []
def save_records(records: list):
plain = json.dumps(records, ensure_ascii=False).encode()
token = _fernet().encrypt(plain).decode()
payload = {"v": 2, "payload": token}
with open(DATA_FILE, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
def record_title(type_id: str, fields: dict) -> str:
for key in ("title", "username", "email", "wechat_id", "qq_number"):
if fields.get(key):
return str(fields[key])
if type_id == "exchange":
ex = fields.get("exchange", "")
un = fields.get("username", "")
return f"{ex} {un}".strip() or "未命名"
return "未命名"
def _field_applies(fd: dict, fields_in: dict) -> bool:
when = fd.get("when")
if not when:
return True
return all(str(fields_in.get(k, "")).strip() == str(v) for k, v in when.items())
def validate_record_payload(payload: dict):
type_id = (payload.get("type_id") or "").strip()
type_def = get_type_def(type_id)
if not type_def:
return None, "无效的凭证类型"
fields_in = payload.get("fields") or {}
if not isinstance(fields_in, dict):
return None, "字段格式错误"
fields = {}
for fd in type_def.get("fields", []):
if not _field_applies(fd, fields_in):
continue
key = fd["key"]
val = fields_in.get(key)
if val is None:
val = ""
val = str(val).strip() if not isinstance(val, str) else val.strip()
if fd.get("required") and not val:
return None, f"{fd.get('label', key)} 不能为空"
fields[key] = val
if type_id == "exchange":
ex = fields.get("exchange", "").lower()
if ex not in VALID_EXCHANGES:
return None, "无效交易所"
for req in ("username", "api_key", "api_secret"):
if not fields.get(req):
return None, f"{req} 不能为空"
if ex == "okx" and not fields.get("password"):
return None, "OKX 密码不能为空"
return {
"type_id": type_id,
"title": record_title(type_id, fields),
"fields": fields,
}, None
def filter_records(records, type_id=None, q=None):
out = records
if type_id:
out = [r for r in out if r.get("type_id") == type_id]
if q:
q = q.lower()
filtered = []
for r in out:
blob = json.dumps(r, ensure_ascii=False).lower()
if q in blob:
filtered.append(r)
out = filtered
return out