"""crypto_okx_hub — 期权多机中控.""" from __future__ import annotations import os import sys from functools import wraps from pathlib import Path from dotenv import load_dotenv from flask import ( Flask, flash, jsonify, redirect, render_template, request, session, url_for, ) ROOT = Path(__file__).resolve().parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) ENV_PATH = ROOT / ".env" load_dotenv(ENV_PATH) from lib.bridge_client_lib import pull_all_enabled from lib.db_lib import init_db from lib.env_lib import apply_env_updates from lib.instances_lib import ( create_instance, delete_instance, get_instance, list_instances, update_instance, ) from lib.sso_lib import build_embed_open_url APP_HOST = (os.getenv("APP_HOST") or "0.0.0.0").strip() or "0.0.0.0" APP_PORT = int(os.getenv("APP_PORT") or 5100) SECRET = (os.getenv("FLASK_SECRET_KEY") or "hub-dev-secret").strip() # 可变登录态:系统设置页可改,并立刻生效 _auth = { "username": (os.getenv("HUB_USERNAME") or "admin").strip() or "admin", "password": (os.getenv("HUB_PASSWORD") or "admin123").strip() or "admin123", } app = Flask(__name__) app.secret_key = SECRET app.config["TEMPLATES_AUTO_RELOAD"] = True init_db() def hub_username() -> str: return _auth["username"] def hub_password() -> str: return _auth["password"] def set_hub_credentials(username: str, password: str) -> None: u = (username or "").strip() or "admin" p = password if password is not None else "" _auth["username"] = u _auth["password"] = p os.environ["HUB_USERNAME"] = u os.environ["HUB_PASSWORD"] = p def login_required(f): @wraps(f) def wrapped(*args, **kwargs): if session.get("logged_in"): return f(*args, **kwargs) if request.path.startswith("/api/"): return jsonify({"ok": False, "msg": "未登录"}), 401 return redirect(url_for("login", next=request.path)) return wrapped @app.route("/login", methods=["GET", "POST"]) def login(): if request.method == "POST": u = (request.form.get("username") or "").strip() p = request.form.get("password") or "" if u == hub_username() and p == hub_password(): session["logged_in"] = True session.permanent = True nxt = (request.args.get("next") or request.form.get("next") or "/").strip() or "/" if not nxt.startswith("/"): nxt = "/" return redirect(nxt) flash("用户名或密码错误") if session.get("logged_in"): return redirect("/") return render_template("login.html") @app.route("/logout") def logout(): session.clear() return redirect(url_for("login")) @app.route("/") @login_required def dashboard(): return render_template("dashboard.html") @app.route("/instances", methods=["GET", "POST"]) @login_required def instances_page(): if request.method == "POST": action = (request.form.get("action") or "create").strip() if action == "create": _, err = create_instance( name=request.form.get("name") or "", base_url=request.form.get("base_url") or "", bridge_token=request.form.get("bridge_token") or "", embed_path=request.form.get("embed_path") or "/embed", tags=request.form.get("tags") or "", enabled=bool(request.form.get("enabled")), ) flash(err or "已添加") elif action == "update": iid = int(request.form.get("id") or 0) _, err = update_instance( iid, name=request.form.get("name") or "", base_url=request.form.get("base_url") or "", bridge_token=request.form.get("bridge_token") or "", embed_path=request.form.get("embed_path") or "/embed", tags=request.form.get("tags") or "", enabled=bool(request.form.get("enabled")), ) flash(err or "已保存") elif action == "delete": iid = int(request.form.get("id") or 0) ok = delete_instance(iid) flash("已删除" if ok else "删除失败") return redirect(url_for("instances_page")) return render_template("instances.html", instances=list_instances()) @app.route("/settings", methods=["GET", "POST"]) @login_required def settings_page(): if request.method == "POST": current_pw = request.form.get("current_password") or "" new_username = (request.form.get("new_username") or "").strip() new_password = request.form.get("new_password") or "" confirm = request.form.get("confirm_password") or "" if current_pw != hub_password(): flash("当前密码错误") return redirect(url_for("settings_page")) if len(new_username) < 2 or len(new_username) > 64: flash("用户名长度需为 2–64") return redirect(url_for("settings_page")) if any(ch.isspace() for ch in new_username): flash("用户名不能包含空格") return redirect(url_for("settings_page")) updates: dict[str, str] = {"HUB_USERNAME": new_username} final_password = hub_password() if new_password or confirm: if len(new_password) < 6: flash("新密码至少 6 位") return redirect(url_for("settings_page")) if new_password != confirm: flash("两次输入的新密码不一致") return redirect(url_for("settings_page")) updates["HUB_PASSWORD"] = new_password final_password = new_password try: apply_env_updates(ENV_PATH, updates) except OSError as exc: flash(f"写入 .env 失败: {exc}") return redirect(url_for("settings_page")) set_hub_credentials(new_username, final_password) flash("账号已更新,立即生效") return redirect(url_for("settings_page")) return render_template("settings.html", current_username=hub_username()) @app.route("/desk") @login_required def desk(): iid = request.args.get("id") inst = None open_url = "" if iid and str(iid).isdigit(): inst = get_instance(int(iid)) if inst: open_url = build_embed_open_url( inst["base_url"], bridge_token=inst["bridge_token"], embed_path=inst.get("embed_path") or "/embed", ) return render_template( "desk.html", instances=list_instances(enabled_only=True), current=inst, open_url=open_url, ) @app.route("/api/overview") @login_required def api_overview(): rows = pull_all_enabled() return jsonify({"ok": True, "instances": rows}) @app.route("/api/instances//open_url") @login_required def api_open_url(iid: int): inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "不存在"}), 404 url = build_embed_open_url( inst["base_url"], bridge_token=inst["bridge_token"], embed_path=inst.get("embed_path") or "/embed", ) return jsonify({"ok": True, "url": url, "base_url": inst["base_url"]}) if __name__ == "__main__": app.run(host=APP_HOST, port=APP_PORT, debug=True, threaded=True)