dekun
157 lines
4.4 KiB
Python
157 lines
4.4 KiB
Python
"""客户端实例 CRUD."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
from lib.db_lib import get_db, row_to_dict
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def normalize_base_url(url: str) -> str:
|
|
u = (url or "").strip().rstrip("/")
|
|
if not u:
|
|
return ""
|
|
if "://" not in u:
|
|
u = "https://" + u
|
|
parsed = urlparse(u)
|
|
if not parsed.netloc:
|
|
return ""
|
|
return f"{parsed.scheme}://{parsed.netloc}"
|
|
|
|
|
|
def list_instances(*, enabled_only: bool = False) -> list[dict[str, Any]]:
|
|
conn = get_db()
|
|
try:
|
|
if enabled_only:
|
|
rows = conn.execute(
|
|
"SELECT * FROM hub_instances WHERE enabled=1 ORDER BY sort_order ASC, id ASC"
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT * FROM hub_instances ORDER BY sort_order ASC, id ASC"
|
|
).fetchall()
|
|
return [row_to_dict(r) for r in rows if r]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_instance(iid: int) -> Optional[dict[str, Any]]:
|
|
conn = get_db()
|
|
try:
|
|
row = conn.execute("SELECT * FROM hub_instances WHERE id=?", (int(iid),)).fetchone()
|
|
return row_to_dict(row)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def create_instance(
|
|
*,
|
|
name: str,
|
|
base_url: str,
|
|
bridge_token: str,
|
|
embed_path: str = "/embed",
|
|
tags: str = "",
|
|
enabled: bool = True,
|
|
sort_order: int = 0,
|
|
) -> tuple[Optional[dict[str, Any]], str]:
|
|
name = (name or "").strip()
|
|
token = (bridge_token or "").strip()
|
|
base = normalize_base_url(base_url)
|
|
path = (embed_path or "/embed").strip() or "/embed"
|
|
if not path.startswith("/"):
|
|
path = "/" + path
|
|
if not name:
|
|
return None, "名称不能为空"
|
|
if not base:
|
|
return None, "公网 URL 无效"
|
|
if not token:
|
|
return None, "Bridge Token 不能为空"
|
|
now = _now()
|
|
conn = get_db()
|
|
try:
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO hub_instances(
|
|
name, base_url, bridge_token, embed_path, tags, enabled, sort_order, created_at, updated_at
|
|
) VALUES (?,?,?,?,?,?,?,?,?)
|
|
""",
|
|
(
|
|
name,
|
|
base,
|
|
token,
|
|
path,
|
|
(tags or "").strip(),
|
|
1 if enabled else 0,
|
|
int(sort_order or 0),
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.commit()
|
|
return get_instance(int(cur.lastrowid)), ""
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def update_instance(iid: int, **fields: Any) -> tuple[Optional[dict[str, Any]], str]:
|
|
existing = get_instance(iid)
|
|
if not existing:
|
|
return None, "实例不存在"
|
|
name = fields.get("name", existing["name"])
|
|
base_url = fields.get("base_url", existing["base_url"])
|
|
token = fields.get("bridge_token", existing["bridge_token"])
|
|
embed_path = fields.get("embed_path", existing["embed_path"])
|
|
tags = fields.get("tags", existing.get("tags") or "")
|
|
enabled = fields.get("enabled", existing["enabled"])
|
|
sort_order = fields.get("sort_order", existing["sort_order"])
|
|
|
|
name = (name or "").strip()
|
|
token = (token or "").strip()
|
|
base = normalize_base_url(str(base_url))
|
|
path = (str(embed_path) or "/embed").strip() or "/embed"
|
|
if not path.startswith("/"):
|
|
path = "/" + path
|
|
if not name or not base or not token:
|
|
return None, "名称 / URL / Token 均必填"
|
|
|
|
conn = get_db()
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
UPDATE hub_instances SET
|
|
name=?, base_url=?, bridge_token=?, embed_path=?, tags=?,
|
|
enabled=?, sort_order=?, updated_at=?
|
|
WHERE id=?
|
|
""",
|
|
(
|
|
name,
|
|
base,
|
|
token,
|
|
path,
|
|
(tags or "").strip(),
|
|
1 if bool(int(enabled)) else 0,
|
|
int(sort_order or 0),
|
|
_now(),
|
|
int(iid),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return get_instance(iid), ""
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def delete_instance(iid: int) -> bool:
|
|
conn = get_db()
|
|
try:
|
|
cur = conn.execute("DELETE FROM hub_instances WHERE id=?", (int(iid),))
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
finally:
|
|
conn.close()
|