"""中控 iframe:壳常驻 + tab 内容 API(/embed、/api/embed/page/)。""" 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, redirect, request, session from jinja2 import ChoiceLoader, FileSystemLoader EMBED_TABS: tuple[str, ...] = ( "key_monitor", "trade", "strategy", "strategy_records", "options", "records", "stats", "risk_policy", "settings", ) PATH_TO_EMBED_TAB: dict[str, str] = { "/": "trade", "/trade": "trade", "/key_monitor": "key_monitor", "/strategy": "strategy", "/strategy/trend": "strategy", "/strategy/roll": "strategy", "/strategy/records": "strategy_records", "/options": "options", "/records": "records", "/stats": "stats", "/risk_policy": "risk_policy", "/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") 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()} 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: 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 return render_main_page_fn(tab, embed_mode="shell") @login_required @app.route("/api/embed/page/") def api_embed_page(tab: str): tab = (tab or "").strip() if tab not in EMBED_TABS: return jsonify({"ok": False, "msg": "unknown tab"}), 404 html = render_main_page_fn(tab, embed_mode="fragment") if isinstance(html, Response): html = html.get_data(as_text=True) return jsonify({"ok": True, "page": tab, "html": html}) def embed_context_extras(exchange_key: str) -> dict: 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), }