crypto_monitor/lib/instance/instance_settings_register.py

"""实例系统设置 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,
    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
        if not clean:
            return jsonify({"ok": True, "changed_keys": [], "restart_required": False})
        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)
        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/<tab>", 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