dekun
80 lines
2.8 KiB
Python
80 lines
2.8 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)
|
|
sort_order = db.Column(db.Integer, nullable=False, default=0)
|
|
|
|
services = db.relationship(
|
|
"Service",
|
|
backref="group",
|
|
lazy="dynamic",
|
|
cascade="all, delete-orphan",
|
|
)
|
|
|
|
|
|
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"
|
|
return f"{proto}://{self.host}:{self.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 首次打开的地址（中控走 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":
|
|
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"
|