dekun
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""批量写入 .env 并刷新 Settings 缓存。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from .config import get_settings
|
|
from .credentials import upsert_env_file
|
|
|
|
|
|
def upsert_env_keys(updates: dict[str, str]) -> Path | None:
|
|
"""写入多项；空 value 跳过。返回最后写入的 .env 路径。"""
|
|
target: Path | None = None
|
|
for key, value in updates.items():
|
|
if value is None:
|
|
continue
|
|
# 允许显式清空密钥（传空串以外的 sentinel 由调用方决定）；空串表示跳过
|
|
if value == "":
|
|
continue
|
|
target = upsert_env_file(key, value)
|
|
get_settings.cache_clear()
|
|
return target
|
|
|
|
|
|
def mask_secret(raw: str | None, *, keep: int = 0) -> str | None:
|
|
"""脱敏：仅返回是否已配置（不再泄露末尾字符）。"""
|
|
s = (raw or "").strip()
|
|
if not s:
|
|
return None
|
|
return "********"
|
|
|
|
|
|
def okx_keys_configured(s=None) -> bool:
|
|
st = s or get_settings()
|
|
return bool(
|
|
(st.okx_api_key or "").strip()
|
|
and (st.okx_api_secret or "").strip()
|
|
and (st.okx_api_passphrase or "").strip()
|
|
)
|
|
|
|
|
|
def binance_keys_configured(s=None) -> bool:
|
|
st = s or get_settings()
|
|
return bool(
|
|
(st.binance_api_key or "").strip() and (st.binance_api_secret or "").strip()
|
|
)
|
|
|
|
|
|
def live_ready(*, exchange: str | None = None) -> tuple[bool, str]:
|
|
"""LIVE 是否可下单。返回 (ok, reason)。"""
|
|
from .exchange.runtime import load_runtime_settings, normalize_exchange_name
|
|
|
|
st = get_settings()
|
|
if st.is_sim:
|
|
return True, "sim"
|
|
ex = normalize_exchange_name(exchange or load_runtime_settings().exchange)
|
|
if ex == "binance":
|
|
if not binance_keys_configured(st):
|
|
return False, "币安 API Key/Secret 未配置"
|
|
return True, "ok"
|
|
if ex == "okx":
|
|
if not okx_keys_configured(st):
|
|
return False, "OKX API Key/Secret/Passphrase 未配置"
|
|
return True, "ok"
|
|
return False, f"未知交易所: {ex}"
|