mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 23:44:16 +08:00
150 lines
4.2 KiB
Python
150 lines
4.2 KiB
Python
"""Key / 凭据本地存储(§14)。
|
|
|
|
存储位置:`data/user_data/secrets.json`,权限 0600。
|
|
优先级:secrets.json > .env > 空(Free 模式)。
|
|
|
|
UI 改 Key 时只动这个文件,不动 .env。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _path() -> Path:
|
|
from app.config import settings
|
|
p = settings.data_dir / "user_data" / "secrets.json"
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
return p
|
|
|
|
|
|
def load() -> dict:
|
|
p = _path()
|
|
if p.exists():
|
|
try:
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("secrets.json malformed: %s", e)
|
|
return {}
|
|
|
|
|
|
def save(updates: dict) -> dict:
|
|
"""合并写入(不会清掉未提及的字段)。返回新内容。"""
|
|
current = load()
|
|
current.update({k: v for k, v in updates.items() if v is not None})
|
|
p = _path()
|
|
p.write_text(json.dumps(current, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
try:
|
|
os.chmod(p, 0o600)
|
|
except OSError:
|
|
pass
|
|
return current
|
|
|
|
|
|
def clear(*keys: str) -> dict:
|
|
"""清掉指定字段(留空清全部)。"""
|
|
p = _path()
|
|
if not p.exists():
|
|
return {}
|
|
if not keys:
|
|
p.unlink()
|
|
return {}
|
|
current = load()
|
|
for k in keys:
|
|
current.pop(k, None)
|
|
p.write_text(json.dumps(current, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return current
|
|
|
|
|
|
def get_tickflow_key() -> str:
|
|
"""取当前 TickFlow Key:secrets.json 优先,否则 .env。"""
|
|
val = load().get("tickflow_api_key")
|
|
if val:
|
|
return val
|
|
from app.config import settings
|
|
return settings.tickflow_api_key or ""
|
|
|
|
|
|
def get_ai_key() -> str:
|
|
"""取当前 AI Key:secrets.json 优先,否则 .env。"""
|
|
val = load().get("ai_api_key")
|
|
if val:
|
|
return val
|
|
from app.config import settings
|
|
return settings.ai_api_key or ""
|
|
|
|
|
|
def get_ai_config(key: str, default: str = "") -> str:
|
|
"""取 AI 配置项:secrets.json 优先,否则 config。"""
|
|
val = load().get(key)
|
|
if val:
|
|
return val
|
|
from app.config import settings
|
|
return getattr(settings, key, default) or default
|
|
|
|
|
|
def get_ai_config_int(key: str, default: int) -> int:
|
|
"""取 AI 数值配置项 (如 ai_max_output_tokens): secrets.json 优先,否则 config。"""
|
|
val = load().get(key)
|
|
if val is not None:
|
|
try:
|
|
return int(val)
|
|
except (TypeError, ValueError):
|
|
logger.warning("ai config %s is not an int: %r", key, val)
|
|
from app.config import settings
|
|
return int(getattr(settings, key, default) or default)
|
|
|
|
|
|
def get_custom_webhook_secret() -> str:
|
|
"""Return the optional HMAC secret for the generic outbound webhook."""
|
|
return str(load().get("custom_webhook_secret") or "")
|
|
|
|
|
|
def set_custom_webhook_secret(secret: str) -> str:
|
|
"""Persist or clear the generic outbound webhook HMAC secret."""
|
|
value = (secret or "").strip()
|
|
if value:
|
|
save({"custom_webhook_secret": value})
|
|
else:
|
|
clear("custom_webhook_secret")
|
|
return value
|
|
|
|
|
|
def get_email_smtp_password() -> str:
|
|
"""Return the SMTP password used by the email notification channel."""
|
|
return str(load().get("email_smtp_password") or "")
|
|
|
|
|
|
def set_email_smtp_password(password: str) -> str:
|
|
"""Persist or clear the SMTP password used by email notifications."""
|
|
value = password or ""
|
|
if value:
|
|
save({"email_smtp_password": value})
|
|
else:
|
|
clear("email_smtp_password")
|
|
return value
|
|
|
|
|
|
def get_env_backed_secret(field: str, env_name: str) -> str:
|
|
"""取环境变量后备的密钥(插件 API Key 等):secrets.json 优先,否则环境变量。
|
|
|
|
与 get_tickflow_key 同优先级语义:UI 写入 secrets.json 后即覆盖 .env。
|
|
"""
|
|
val = load().get(field)
|
|
if val:
|
|
return str(val).strip()
|
|
return os.environ.get(env_name, "").strip()
|
|
|
|
|
|
def mask(key: str, prefix: int = 4, suffix: int = 4) -> str:
|
|
"""脱敏显示。"""
|
|
if not key:
|
|
return ""
|
|
if len(key) <= prefix + suffix:
|
|
return "•" * len(key)
|
|
return f"{key[:prefix]}{'•' * 6}{key[-suffix:]}"
|