crypto_monitor/lib/instance/instance_embed_lib.py

"""中控 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, redirect, request, session
from jinja2 import ChoiceLoader, FileSystemLoader

EMBED_TABS: tuple[str, ...] = (
    "key_monitor",
    "trade",
    "strategy",
    "strategy_records",
    "records",
    "stats",
    "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",
    "/records": "records",
    "/stats": "stats",
    "/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 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:
    app.config["RENDER_MAIN_PAGE_FN"] = render_main_page_fn

    @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/<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
        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),
    }