dekun
f7ce6f1058
- 分组支持 icon 字段，可按名称自动匹配或手动选择 - 左侧导航与总览卡片显示彩色 SVG 图标 - 优化侧栏链接圆角与选中态样式 Co-authored-by: Cursor <cursoragent@cursor.com>
119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_login import UserMixin
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
|
|
db = SQLAlchemy()
|
|
|
|
|
|
class User(UserMixin, db.Model):
|
|
__tablename__ = "users"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(80), unique=True, nullable=False, index=True)
|
|
password_hash = db.Column(db.String(256), nullable=False)
|
|
|
|
def set_password(self, password: str) -> None:
|
|
self.password_hash = generate_password_hash(password)
|
|
|
|
def check_password(self, password: str) -> bool:
|
|
return check_password_hash(self.password_hash, password)
|
|
|
|
|
|
class ServiceGroup(db.Model):
|
|
__tablename__ = "service_groups"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(120), nullable=False)
|
|
icon = db.Column(db.String(16), nullable=False, default="", server_default="")
|
|
sort_order = db.Column(db.Integer, nullable=False, default=0)
|
|
|
|
services = db.relationship(
|
|
"Service",
|
|
backref="group",
|
|
lazy="dynamic",
|
|
cascade="all, delete-orphan",
|
|
)
|
|
|
|
def resolve_icon_key(self) -> str:
|
|
key = (self.icon or "").strip().lower()
|
|
if key:
|
|
return key
|
|
name = self.name or ""
|
|
lower = name.lower()
|
|
if "下单" in name:
|
|
return "order"
|
|
if "交易" in name:
|
|
return "trade"
|
|
if "复盘" in name:
|
|
return "review"
|
|
if "api" in lower:
|
|
return "api"
|
|
if "gate" in lower or "扫单" in name:
|
|
return "gate"
|
|
if "k线" in lower or "k 线" in name or "chart" in lower:
|
|
return "chart"
|
|
return "folder"
|
|
|
|
|
|
class Service(db.Model):
|
|
__tablename__ = "services"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(120), nullable=False)
|
|
scheme = db.Column(db.String(8), nullable=False, default="http")
|
|
host = db.Column(db.String(255), nullable=False)
|
|
port = db.Column(db.Integer, nullable=False)
|
|
path = db.Column(db.String(512), nullable=False, default="/")
|
|
sort_order = db.Column(db.Integer, nullable=False, default=0)
|
|
group_id = db.Column(
|
|
db.Integer, db.ForeignKey("service_groups.id"), nullable=False, index=True
|
|
)
|
|
# hub=复盘中控（iframe 嵌入需 /embed-auth）；留空=普通内嵌
|
|
embed_kind = db.Column(db.String(16), nullable=False, default="", server_default="")
|
|
|
|
def build_origin(self) -> str:
|
|
proto = (self.scheme or "http").strip().lower()
|
|
if proto not in ("http", "https"):
|
|
proto = "http"
|
|
port = int(self.port or (443 if proto == "https" else 80))
|
|
if (proto == "https" and port == 443) or (proto == "http" and port == 80):
|
|
return f"{proto}://{self.host}"
|
|
return f"{proto}://{self.host}:{port}"
|
|
|
|
def build_url(self) -> str:
|
|
p = (self.path or "/").strip()
|
|
if not p.startswith("/"):
|
|
p = "/" + p
|
|
return f"{self.build_origin()}{p}"
|
|
|
|
def build_open_url(self) -> str:
|
|
"""导航 iframe 首次打开的地址（中控 / Gate 扫单走 login?embed=1 以便写入会话）。"""
|
|
kind = (self.embed_kind or "").strip().lower()
|
|
next_path = (self.path or "/monitor").strip() or "/monitor"
|
|
if not next_path.startswith("/"):
|
|
next_path = "/" + next_path
|
|
if kind == "hub" or kind in ("gate_scout", "gate_exec", "scout", "exec"):
|
|
from urllib.parse import urlencode
|
|
|
|
q = urlencode({"embed": "1", "next": next_path})
|
|
return f"{self.build_origin()}/login?{q}"
|
|
return self.build_url()
|
|
|
|
def is_hub_embed(self) -> bool:
|
|
return (self.embed_kind or "").strip().lower() == "hub"
|
|
|
|
def is_gate_scout_embed(self) -> bool:
|
|
return (self.embed_kind or "").strip().lower() in (
|
|
"gate_scout",
|
|
"gate_exec",
|
|
"scout",
|
|
"exec",
|
|
)
|
|
|
|
def gate_scout_api_login_path(self) -> str:
|
|
"""服务端代登录使用的 API 路径。"""
|
|
kind = (self.embed_kind or "").strip().lower()
|
|
if kind in ("gate_exec", "exec"):
|
|
return "/login"
|
|
return "/api/auth/login"
|