crypto_okx_hub/app.py

"""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))

load_dotenv(ROOT / ".env")

from lib.bridge_client_lib import pull_all_enabled
from lib.db_lib import init_db
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)
HUB_USERNAME = (os.getenv("HUB_USERNAME") or "admin").strip() or "admin"
HUB_PASSWORD = (os.getenv("HUB_PASSWORD") or "change-me").strip() or "change-me"
SECRET = (os.getenv("FLASK_SECRET_KEY") or "hub-dev-secret").strip()

app = Flask(__name__)
app.secret_key = SECRET
app.config["TEMPLATES_AUTO_RELOAD"] = True

init_db()


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("/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/<int:iid>/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)