"""实例系统设置 API:导航开关,env 读写,改密,PM2 重启.""" from __future__ import annotations import os from functools import wraps from typing import Any, Callable from flask import jsonify, request, session from lib.env.env_file_lib import apply_env_updates, env_get, read_env_lines from lib.env.env_ui_manifest import ( build_env_ui_payload, filter_updates_for_ui, coerce_hedge_partial_close_with_manual, validate_env_ui_updates, ) from lib.env.env_schema import parse_env_example_schema from lib.instance.instance_display_prefs_lib import ( display_meta_for_ui, get_display_prefs, normalize_display_prefs, save_display_prefs, tab_allowed, ) from lib.instance.instance_pm2_lib import restart_instance_pm2 from lib.instance.runtime_config_lib import apply_env_reload def _api_login_required(hub_token_write_allowed: bool = False): def decorator(f): @wraps(f) def wrapped(*args, **kwargs): from lib.hub.hub_auth import request_allowed as hub_request_allowed logged_in = bool(session.get("logged_in")) auth_disabled = (os.getenv("APP_AUTH_DISABLED") or "").strip().lower() in ( "1", "true", "yes", "on", ) hub_hdr = (request.headers.get("X-Hub-Token") or "").strip() bridge = (os.getenv("HUB_BRIDGE_TOKEN") or "").strip() if hub_hdr and bridge and hub_hdr == bridge and not hub_token_write_allowed: return jsonify({"ok": False, "msg": "Hub Token 不可修改实例设置"}), 403 if hub_request_allowed(logged_in, auth_disabled): return f(*args, **kwargs) return jsonify({"ok": False, "msg": "未登录"}), 401 return wrapped return decorator def register_instance_settings_routes( app, *, get_db: Callable, login_required_fn: Callable, base_dir: str, exchange_key: str, username: str, password: str, ) -> None: env_path = os.path.join(base_dir, ".env") example_path = os.path.join(base_dir, ".env.example") api_auth = _api_login_required() @app.route("/api/settings/display", methods=["GET", "POST"]) @api_auth def api_settings_display(): if request.method == "GET": prefs = get_display_prefs(get_db) return jsonify( { "ok": True, "display": prefs, "meta": display_meta_for_ui(), } ) body = request.get_json(silent=True) or {} raw = body.get("display") if isinstance(body.get("display"), dict) else body saved = save_display_prefs(get_db, raw) return jsonify({"ok": True, "display": saved}) @app.route("/api/settings/env/meta", methods=["GET"]) @api_auth def api_env_meta(): groups = build_env_ui_payload(exchange_key, example_path, env_path) return jsonify({"ok": True, "groups": groups}) @app.route("/api/settings/env", methods=["GET", "POST"]) @api_auth def api_settings_env(): if request.method == "GET": groups = build_env_ui_payload(exchange_key, example_path, env_path) return jsonify({"ok": True, "groups": groups}) body = request.get_json(silent=True) or {} updates = body.get("values") if isinstance(body.get("values"), dict) else body if not isinstance(updates, dict): return jsonify({"ok": False, "msg": "无效请求体"}), 400 updates = filter_updates_for_ui(exchange_key, updates) clean, errors = validate_env_ui_updates(exchange_key, example_path, updates) if errors: return jsonify({"ok": False, "msg": "; ".join(errors)}), 400 clean = coerce_hedge_partial_close_with_manual(clean, env_path=env_path) if not clean: return jsonify({"ok": True, "changed_keys": [], "restart_required": False}) if "OKX_OPTIONS_MARGIN_MODE" in clean: try: from lib.options.options_margin_mode_lib import normalize_options_margin_mode from lib.options.options_spot_bridge_lib import mode_switch_block_msg lines = read_env_lines(env_path) old_mode = normalize_options_margin_mode(env_get(lines, "OKX_OPTIONS_MARGIN_MODE") or "coin") new_mode = normalize_options_margin_mode(clean.get("OKX_OPTIONS_MARGIN_MODE")) if old_mode != new_mode: conn_m = get_db() try: block = mode_switch_block_msg(conn_m, None) if block: return jsonify({"ok": False, "msg": block}), 400 finally: conn_m.close() except Exception as e: return jsonify({"ok": False, "msg": f"本位切换校验失败: {e}"}), 400 changed = apply_env_updates(env_path, clean) groups = parse_env_example_schema(example_path) reload_info = apply_env_reload(env_path, get_db, changed, groups) return jsonify( { "ok": True, "changed_keys": changed, "restart_required": reload_info.get("restart_required", False), } ) @app.route("/api/settings/password", methods=["POST"]) @api_auth def api_change_password(): body = request.get_json(silent=True) or {} old_password = str(body.get("old_password") or "") new_username = str(body.get("new_username") or "").strip() new_password = str(body.get("new_password") or "") confirm = str(body.get("confirm_password") or "") if not old_password or old_password != password: return jsonify({"ok": False, "msg": "当前密码错误"}), 400 if len(new_password) < 6: return jsonify({"ok": False, "msg": "新密码至少 6 位"}), 400 if new_password != confirm: return jsonify({"ok": False, "msg": "两次输入的新密码不一致"}), 400 updates: dict[str, str] = {"APP_PASSWORD": new_password} if new_username: updates["APP_USERNAME"] = new_username changed = apply_env_updates(env_path, updates) groups = parse_env_example_schema(example_path) apply_env_reload(env_path, get_db, changed, groups) return jsonify({"ok": True, "restart_required": True, "changed_keys": changed}) @app.route("/api/admin/restart", methods=["POST"]) @api_auth def api_admin_restart(): result = restart_instance_pm2(exchange_key, defer=True) code = 200 if result.get("ok") else 500 return jsonify({"ok": bool(result.get("ok")), **result}), code @app.route("/api/admin/health", methods=["GET"]) def api_admin_health(): return jsonify({"ok": True, "status": "up"}) def tab_allowed_fn(tab: str) -> bool: prefs = get_display_prefs(get_db) return tab_allowed(tab, prefs) app.config["INSTANCE_GET_DB"] = get_db app.config["INSTANCE_TAB_ALLOWED_FN"] = tab_allowed_fn @app.route("/api/embed/tab_allowed/", methods=["GET"]) @api_auth def api_tab_allowed(tab: str): prefs = get_display_prefs(get_db) return jsonify({"ok": True, "tab": tab, "allowed": tab_allowed(tab, prefs)}) def merge_ui_template_context(page: str, get_db: Callable, **settings_kwargs: Any) -> dict[str, Any]: from lib.instance.instance_settings_lib import settings_page_context prefs = get_display_prefs(get_db) ctx = { "display": prefs, "display_meta": display_meta_for_ui(), **settings_page_context(page, display=prefs, **settings_kwargs), } return ctx