dekun
239 lines
8.1 KiB
Python
239 lines
8.1 KiB
Python
"""中控 iframe:壳常驻 + tab 内容 API(/embed,/api/embed/page/<tab>)."""
|
|
from __future__ import annotations
|
|
|
|
from lib.paths import embed_templates_dir
|
|
|
|
import os
|
|
from typing import Callable
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit
|
|
|
|
from flask import Flask, Response, jsonify, make_response, redirect, request, session
|
|
from jinja2 import ChoiceLoader, FileSystemLoader
|
|
|
|
EMBED_TABS: tuple[str, ...] = (
|
|
"dashboard",
|
|
"account_ledger",
|
|
"key_monitor",
|
|
"trade",
|
|
"strategy",
|
|
"strategy_records",
|
|
"options",
|
|
"options_review",
|
|
"hedge_plan",
|
|
"records",
|
|
"stats",
|
|
"risk_policy",
|
|
"system_guide",
|
|
"env_config",
|
|
"settings",
|
|
)
|
|
|
|
PATH_TO_EMBED_TAB: dict[str, str] = {
|
|
"/": "trade",
|
|
"/trade": "trade",
|
|
"/dashboard": "dashboard",
|
|
"/account_ledger": "account_ledger",
|
|
"/key_monitor": "key_monitor",
|
|
"/strategy": "strategy",
|
|
"/strategy/trend": "strategy",
|
|
"/strategy/roll": "strategy",
|
|
"/strategy/records": "strategy_records",
|
|
"/options": "options",
|
|
"/options/review": "options_review",
|
|
"/hedge-plan": "hedge_plan",
|
|
"/records": "records",
|
|
"/stats": "stats",
|
|
"/risk_policy": "risk_policy",
|
|
"/system_guide": "system_guide",
|
|
"/env_config": "env_config",
|
|
"/settings": "settings",
|
|
}
|
|
|
|
ORDER_RULE_TIPS_BY_EXCHANGE: dict[str, str] = {
|
|
"gate": "order_monitor_rule_tips_gate.html",
|
|
"binance": "order_monitor_rule_tips_binance.html",
|
|
"okx": "order_monitor_rule_tips_okx.html",
|
|
}
|
|
|
|
|
|
def order_rule_tips_template(exchange_key: str) -> str:
|
|
ex = (exchange_key or "").strip().lower()
|
|
return ORDER_RULE_TIPS_BY_EXCHANGE.get(ex, "order_monitor_rule_tips_gate.html")
|
|
|
|
|
|
def include_transfer_block(exchange_key: str) -> bool:
|
|
"""三所 standalone / embed 壳均在顶栏展示划转区块."""
|
|
return (exchange_key or "").strip().lower() in ORDER_RULE_TIPS_BY_EXCHANGE
|
|
|
|
|
|
def ui_open_guard_enabled(exchange_key: str) -> bool:
|
|
return (exchange_key or "").strip().lower() == "okx"
|
|
|
|
|
|
def ui_orphan_recovery_enabled(exchange_key: str) -> bool:
|
|
return (exchange_key or "").strip().lower() == "binance"
|
|
|
|
|
|
def path_to_embed_tab(path: str) -> str | None:
|
|
p = (path or "/").strip()
|
|
if not p.startswith("/"):
|
|
p = "/" + p
|
|
base = urlsplit(p).path.rstrip("/") or "/"
|
|
return PATH_TO_EMBED_TAB.get(base)
|
|
|
|
|
|
def embed_shell_enabled() -> bool:
|
|
return (os.getenv("HUB_EMBED_SHELL") or "1").strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
_SETTINGS_SUB_TABS = frozenset(
|
|
{"nav", "password", "transfer", "export", "options_swap", "options_transfer"}
|
|
)
|
|
|
|
|
|
def redirect_to_embed_shell_if_enabled(page: str):
|
|
"""直连 /trade 等整页路由时,重定向到 embed 壳(顶栏常驻,tab 软切换)."""
|
|
if not embed_shell_enabled():
|
|
return None
|
|
if (request.args.get("embed") or "").strip() == "1":
|
|
return None
|
|
if (request.path or "").rstrip("/") == "/embed":
|
|
return None
|
|
q = {k: v for k, v in request.args.items()}
|
|
# embed 的 tab=页面名;系统设置内页签用 settings_tab,避免 /settings?tab=transfer 被覆盖成 tab=settings
|
|
if (page or "").strip() == "settings":
|
|
sub = (q.get("settings_tab") or "").strip()
|
|
legacy = (q.get("tab") or "").strip()
|
|
if not sub and legacy in _SETTINGS_SUB_TABS:
|
|
q["settings_tab"] = legacy
|
|
q["tab"] = page
|
|
q["embed"] = "1"
|
|
return redirect("/embed?" + urlencode(q))
|
|
|
|
|
|
def rewrite_embed_dest(path: str, hub_theme: str | None = None) -> str:
|
|
"""embed=1 打开时:/trade → /embed?tab=trade&embed=1"""
|
|
if not embed_shell_enabled():
|
|
split = urlsplit(path or "/")
|
|
q = dict(parse_qsl(split.query, keep_blank_values=True))
|
|
q["embed"] = "1"
|
|
ht = (hub_theme or q.get("hub_theme") or "").strip().lower()
|
|
if ht in ("light", "dark"):
|
|
q["hub_theme"] = ht
|
|
dest = split.path or "/"
|
|
if q:
|
|
return f"{dest}?{urlencode(q)}"
|
|
return dest + "?embed=1"
|
|
split = urlsplit(path or "/")
|
|
tab = path_to_embed_tab(split.path)
|
|
q = dict(parse_qsl(split.query, keep_blank_values=True))
|
|
if tab:
|
|
if tab == "settings":
|
|
sub = (q.get("settings_tab") or "").strip()
|
|
legacy = (q.get("tab") or "").strip()
|
|
if not sub and legacy in _SETTINGS_SUB_TABS:
|
|
q["settings_tab"] = legacy
|
|
q["tab"] = tab
|
|
q["embed"] = "1"
|
|
ht = (hub_theme or q.get("hub_theme") or "").strip().lower()
|
|
if ht in ("light", "dark"):
|
|
q["hub_theme"] = ht
|
|
return f"/embed?{urlencode(q)}"
|
|
q["embed"] = "1"
|
|
ht = (hub_theme or q.get("hub_theme") or "").strip().lower()
|
|
if ht in ("light", "dark"):
|
|
q["hub_theme"] = ht
|
|
dest = split.path or "/"
|
|
if split.query:
|
|
dest += "?" + split.query
|
|
if "embed=1" not in dest:
|
|
sep = "&" if "?" in dest else "?"
|
|
dest += f"{sep}embed=1"
|
|
if ht in ("light", "dark") and "hub_theme=" not in dest:
|
|
sep = "&" if "?" in dest else "?"
|
|
dest += f"{sep}hub_theme={ht}"
|
|
return dest
|
|
|
|
|
|
def attach_embed_templates(app: Flask, repo_root: str) -> None:
|
|
embed_dir = embed_templates_dir(repo_root)
|
|
if not os.path.isdir(embed_dir):
|
|
return
|
|
existing = app.jinja_loader
|
|
loaders = [FileSystemLoader(embed_dir)]
|
|
if existing is not None:
|
|
if isinstance(existing, ChoiceLoader):
|
|
loaders = list(existing.loaders) + loaders
|
|
else:
|
|
loaders.insert(0, existing)
|
|
app.jinja_loader = ChoiceLoader(loaders)
|
|
|
|
|
|
def register_embed_routes(
|
|
app: Flask,
|
|
login_required: Callable,
|
|
render_main_page_fn: Callable,
|
|
) -> None:
|
|
from lib.instance.instance_live_push_lib import register_instance_live_routes
|
|
|
|
app.config["RENDER_MAIN_PAGE_FN"] = render_main_page_fn
|
|
register_instance_live_routes(app, login_required)
|
|
|
|
@login_required
|
|
@app.route("/embed")
|
|
def embed_shell_page():
|
|
tab = (request.args.get("tab") or "trade").strip()
|
|
if tab not in EMBED_TABS:
|
|
tab = "trade"
|
|
session["hub_embed_shell"] = True
|
|
resp = make_response(render_main_page_fn(tab, embed_mode="shell"))
|
|
resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
|
|
resp.headers["Pragma"] = "no-cache"
|
|
return resp
|
|
|
|
@login_required
|
|
@app.route("/api/embed/page/<tab>")
|
|
def api_embed_page(tab: str):
|
|
tab = (tab or "").strip()
|
|
if tab not in EMBED_TABS:
|
|
return jsonify({"ok": False, "msg": "unknown tab"}), 404
|
|
allowed_fn = app.config.get("INSTANCE_TAB_ALLOWED_FN")
|
|
if callable(allowed_fn) and not allowed_fn(tab):
|
|
return jsonify({"ok": False, "msg": "tab disabled"}), 403
|
|
html = render_main_page_fn(tab, embed_mode="fragment")
|
|
if isinstance(html, Response):
|
|
html = html.get_data(as_text=True)
|
|
resp = jsonify({"ok": True, "page": tab, "html": html})
|
|
resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
|
|
resp.headers["Pragma"] = "no-cache"
|
|
return resp
|
|
|
|
|
|
def pwa_app_name(exchange_key: str) -> str:
|
|
"""安装 App / 主屏幕显示名(各所独立标识)."""
|
|
ex = (exchange_key or "").strip().lower()
|
|
return {
|
|
"binance": "Binance 交易系统",
|
|
"okx": "OKX 交易系统",
|
|
"gate": "Gate 交易系统",
|
|
}.get(ex, "交易系统")
|
|
|
|
|
|
def embed_context_extras(exchange_key: str) -> dict:
|
|
# 顶栏共享模板调用 trading_account_label / options_funding_label;
|
|
# 须注入三所,否则 Gate/Binance 渲染会 UndefinedError → HTTP 500.
|
|
from lib.instance.instance_embed_context_lib import (
|
|
options_funding_label,
|
|
trading_account_label,
|
|
)
|
|
|
|
return {
|
|
"order_rule_tips_tpl": order_rule_tips_template(exchange_key),
|
|
"include_transfer_block": include_transfer_block(exchange_key),
|
|
"ui_open_guard_enabled": ui_open_guard_enabled(exchange_key),
|
|
"ui_orphan_recovery_enabled": ui_orphan_recovery_enabled(exchange_key),
|
|
"pwa_app_name": pwa_app_name(exchange_key),
|
|
"options_funding_label": options_funding_label,
|
|
"trading_account_label": trading_account_label,
|
|
}
|