1434 lines
66 KiB
Python
1434 lines
66 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""企微客服助手的内置配置后台。
|
||
|
||
仅依赖 Python 标准库，提供：
|
||
- 用户名/密码登录和强制首次改密
|
||
- admin / operator / viewer 三种角色
|
||
- AI 与 MCP 配置的网页管理
|
||
- 供桌面端认证并拉取配置的 JSON API
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import getpass
|
||
import hashlib
|
||
import hmac
|
||
import html
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import secrets
|
||
import sqlite3
|
||
import sys
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
from collections import defaultdict, deque
|
||
from datetime import datetime
|
||
from http import HTTPStatus
|
||
from http.cookies import SimpleCookie
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
DEFAULT_DB = SCRIPT_DIR / "backend.db"
|
||
DEFAULT_RUNTIME_FILE = SCRIPT_DIR / "backend_runtime.json"
|
||
DEFAULT_HOST = "127.0.0.1"
|
||
DEFAULT_PORT = 8765
|
||
DEFAULT_PORT_ATTEMPTS = 100
|
||
DEFAULT_ADMIN_PASSWORD = "Admin@123456"
|
||
ROLES = ("admin", "operator", "viewer")
|
||
ROLE_LABELS = {"admin": "管理员", "operator": "配置员", "viewer": "只读用户"}
|
||
PBKDF2_ITERATIONS = 310_000
|
||
CONFIG_KEYS = (
|
||
"AI_ENABLED",
|
||
"AI_API_BASE",
|
||
"AI_API_KEY",
|
||
"AI_MODEL",
|
||
"GROK_CUSTOMER_SERVICE_ENABLED",
|
||
"GROK_CUSTOMER_SERVICE_TIMEOUT",
|
||
"GROK_CUSTOMER_SERVICE_MAX_TURNS",
|
||
"GROK_CUSTOMER_SERVICE_EFFORT",
|
||
"GROK_MODEL_ENABLED",
|
||
"GROK_API_BASE",
|
||
"GROK_API_KEY",
|
||
"GROK_MODEL",
|
||
"GROK_API_BACKEND",
|
||
"GROK_AUTH_SCHEME",
|
||
"GROK_DIFY_INPUTS",
|
||
"GROK_CONTEXT_WINDOW",
|
||
"GROK_MAX_TOKENS",
|
||
"GROK_TEMPERATURE",
|
||
"AI_USE_VISION",
|
||
"AI_CONTEXT_ENABLED",
|
||
"AI_CONTEXT_MAX_ROUNDS",
|
||
"AI_COUNTER_INSULT_ENABLED",
|
||
"AI_AGENT_NAME",
|
||
"AI_HOSPITAL_NAME",
|
||
"AI_MAX_TOKENS",
|
||
"AI_TEMPERATURE",
|
||
"AI_TIMEOUT",
|
||
"AI_MCP_ENABLED",
|
||
"AI_MCP_MAX_ROUNDS",
|
||
"AI_MCP_SERVERS",
|
||
)
|
||
BOOL_KEYS = {
|
||
"AI_ENABLED",
|
||
"AI_USE_VISION",
|
||
"AI_CONTEXT_ENABLED",
|
||
"AI_COUNTER_INSULT_ENABLED",
|
||
"AI_MCP_ENABLED",
|
||
"GROK_MODEL_ENABLED",
|
||
"GROK_CUSTOMER_SERVICE_ENABLED",
|
||
}
|
||
|
||
|
||
def now_text() -> str:
|
||
return datetime.now().astimezone().isoformat(timespec="seconds")
|
||
|
||
|
||
def password_hash(password: str, salt: bytes | None = None) -> tuple[str, str]:
|
||
salt = salt or secrets.token_bytes(16)
|
||
digest = hashlib.pbkdf2_hmac(
|
||
"sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
|
||
)
|
||
return base64.b64encode(salt).decode("ascii"), base64.b64encode(digest).decode("ascii")
|
||
|
||
|
||
def verify_password(password: str, salt_text: str, digest_text: str) -> bool:
|
||
try:
|
||
salt = base64.b64decode(salt_text)
|
||
expected = base64.b64decode(digest_text)
|
||
except Exception:
|
||
return False
|
||
actual = hashlib.pbkdf2_hmac(
|
||
"sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
|
||
)
|
||
return hmac.compare_digest(actual, expected)
|
||
|
||
|
||
def token_hash(token: str) -> str:
|
||
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def load_initial_config() -> dict[str, Any]:
|
||
saved: Any = {}
|
||
for path in (
|
||
SCRIPT_DIR / "ai_settings.local.json",
|
||
SCRIPT_DIR / "ai_settings.json",
|
||
):
|
||
try:
|
||
saved = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError, TypeError):
|
||
continue
|
||
if isinstance(saved, dict):
|
||
break
|
||
defaults: dict[str, Any] = {
|
||
"AI_ENABLED": True,
|
||
"AI_API_BASE": "",
|
||
"AI_API_KEY": "",
|
||
"AI_MODEL": "",
|
||
"GROK_CUSTOMER_SERVICE_ENABLED": True,
|
||
"GROK_CUSTOMER_SERVICE_TIMEOUT": 180,
|
||
"GROK_CUSTOMER_SERVICE_MAX_TURNS": 8,
|
||
"GROK_CUSTOMER_SERVICE_EFFORT": "low",
|
||
"GROK_MODEL_ENABLED": False,
|
||
"GROK_API_BASE": "",
|
||
"GROK_API_KEY": "",
|
||
"GROK_MODEL": "",
|
||
"GROK_API_BACKEND": "chat_completions",
|
||
"GROK_AUTH_SCHEME": "auto",
|
||
"GROK_DIFY_INPUTS": {},
|
||
"GROK_CONTEXT_WINDOW": 128000,
|
||
"GROK_MAX_TOKENS": 8192,
|
||
"GROK_TEMPERATURE": 0.3,
|
||
"AI_USE_VISION": False,
|
||
"AI_CONTEXT_ENABLED": True,
|
||
"AI_CONTEXT_MAX_ROUNDS": 5,
|
||
"AI_COUNTER_INSULT_ENABLED": False,
|
||
"AI_AGENT_NAME": "客服",
|
||
"AI_HOSPITAL_NAME": "",
|
||
"AI_MAX_TOKENS": 500,
|
||
"AI_TEMPERATURE": 0.35,
|
||
"AI_TIMEOUT": 120,
|
||
"AI_MCP_ENABLED": False,
|
||
"AI_MCP_MAX_ROUNDS": 5,
|
||
"AI_MCP_SERVERS": [],
|
||
}
|
||
if isinstance(saved, dict):
|
||
defaults.update({key: saved[key] for key in CONFIG_KEYS if key in saved})
|
||
return defaults
|
||
|
||
|
||
class ClosingConnection(sqlite3.Connection):
|
||
"""提交/回滚后立即关闭，避免 Windows 上数据库文件长期被占用。"""
|
||
|
||
def __exit__(self, exc_type, exc_value, traceback):
|
||
try:
|
||
return super().__exit__(exc_type, exc_value, traceback)
|
||
finally:
|
||
self.close()
|
||
|
||
|
||
class Database:
|
||
def __init__(self, path: Path):
|
||
self.path = path
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
def connect(self) -> sqlite3.Connection:
|
||
connection = sqlite3.connect(
|
||
self.path, timeout=10, factory=ClosingConnection
|
||
)
|
||
connection.row_factory = sqlite3.Row
|
||
connection.execute("PRAGMA foreign_keys = ON")
|
||
return connection
|
||
|
||
def initialize(self, initial_password: str) -> bool:
|
||
with self.connect() as db:
|
||
db.executescript(
|
||
"""
|
||
PRAGMA journal_mode = WAL;
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
||
password_salt TEXT NOT NULL,
|
||
password_digest TEXT NOT NULL,
|
||
role TEXT NOT NULL CHECK(role IN ('admin','operator','viewer')),
|
||
active INTEGER NOT NULL DEFAULT 1,
|
||
must_change_password INTEGER NOT NULL DEFAULT 1,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS auth_tokens (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||
token_digest TEXT NOT NULL UNIQUE,
|
||
kind TEXT NOT NULL,
|
||
csrf_token TEXT NOT NULL,
|
||
device_name TEXT NOT NULL DEFAULT '',
|
||
created_at TEXT NOT NULL,
|
||
expires_at INTEGER NOT NULL,
|
||
last_used_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS model_config (
|
||
id INTEGER PRIMARY KEY CHECK(id = 1),
|
||
config_json TEXT NOT NULL,
|
||
version INTEGER NOT NULL DEFAULT 1,
|
||
updated_at TEXT NOT NULL,
|
||
updated_by INTEGER REFERENCES users(id)
|
||
);
|
||
CREATE TABLE IF NOT EXISTS audit_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id INTEGER REFERENCES users(id),
|
||
action TEXT NOT NULL,
|
||
detail TEXT NOT NULL DEFAULT '',
|
||
ip_address TEXT NOT NULL DEFAULT '',
|
||
created_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_tokens_digest ON auth_tokens(token_digest);
|
||
CREATE INDEX IF NOT EXISTS idx_audit_created ON audit_log(created_at DESC);
|
||
"""
|
||
)
|
||
created = db.execute("SELECT COUNT(*) FROM users").fetchone()[0] == 0
|
||
if created:
|
||
salt, digest = password_hash(initial_password)
|
||
stamp = now_text()
|
||
db.execute(
|
||
"""INSERT INTO users
|
||
(username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
|
||
VALUES (?,?,?,?,1,1,?,?)""",
|
||
("admin", salt, digest, "admin", stamp, stamp),
|
||
)
|
||
if db.execute("SELECT COUNT(*) FROM model_config").fetchone()[0] == 0:
|
||
admin = db.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
|
||
db.execute(
|
||
"""INSERT INTO model_config
|
||
(id,config_json,version,updated_at,updated_by) VALUES (1,?,1,?,?)""",
|
||
(
|
||
json.dumps(load_initial_config(), ensure_ascii=False),
|
||
now_text(),
|
||
admin["id"] if admin else None,
|
||
),
|
||
)
|
||
else:
|
||
row = db.execute(
|
||
"SELECT config_json,version FROM model_config WHERE id=1"
|
||
).fetchone()
|
||
try:
|
||
existing = json.loads(row["config_json"]) if row else {}
|
||
except (ValueError, TypeError):
|
||
existing = {}
|
||
if not isinstance(existing, dict):
|
||
existing = {}
|
||
migrated = load_initial_config()
|
||
migrated.update(
|
||
{key: existing[key] for key in CONFIG_KEYS if key in existing}
|
||
)
|
||
if migrated != existing:
|
||
db.execute(
|
||
"""UPDATE model_config
|
||
SET config_json=?,version=?,updated_at=? WHERE id=1""",
|
||
(
|
||
json.dumps(migrated, ensure_ascii=False),
|
||
int(row["version"] if row else 0) + 1,
|
||
now_text(),
|
||
),
|
||
)
|
||
db.commit()
|
||
return created
|
||
|
||
def authenticate(self, username: str, password: str) -> sqlite3.Row | None:
|
||
with self.connect() as db:
|
||
user = db.execute(
|
||
"SELECT * FROM users WHERE username = ? AND active = 1", (username,)
|
||
).fetchone()
|
||
if user and verify_password(password, user["password_salt"], user["password_digest"]):
|
||
return user
|
||
return None
|
||
|
||
def create_token(
|
||
self, user_id: int, kind: str, device_name: str, lifetime: int
|
||
) -> tuple[str, str]:
|
||
token = secrets.token_urlsafe(36)
|
||
csrf = secrets.token_urlsafe(24)
|
||
stamp = now_text()
|
||
with self.connect() as db:
|
||
db.execute(
|
||
"""INSERT INTO auth_tokens
|
||
(user_id,token_digest,kind,csrf_token,device_name,created_at,expires_at,last_used_at)
|
||
VALUES (?,?,?,?,?,?,?,?)""",
|
||
(
|
||
user_id,
|
||
token_hash(token),
|
||
kind,
|
||
csrf,
|
||
str(device_name or "")[:120],
|
||
stamp,
|
||
int(time.time()) + lifetime,
|
||
stamp,
|
||
),
|
||
)
|
||
db.execute("DELETE FROM auth_tokens WHERE expires_at < ?", (int(time.time()),))
|
||
db.commit()
|
||
return token, csrf
|
||
|
||
def session(self, token: str) -> sqlite3.Row | None:
|
||
if not token:
|
||
return None
|
||
with self.connect() as db:
|
||
row = db.execute(
|
||
"""SELECT u.*, t.id AS token_id, t.kind AS token_kind,
|
||
t.csrf_token, t.expires_at
|
||
FROM auth_tokens t JOIN users u ON u.id=t.user_id
|
||
WHERE t.token_digest=? AND t.expires_at>=? AND u.active=1""",
|
||
(token_hash(token), int(time.time())),
|
||
).fetchone()
|
||
if row:
|
||
db.execute(
|
||
"UPDATE auth_tokens SET last_used_at=? WHERE id=?",
|
||
(now_text(), row["token_id"]),
|
||
)
|
||
db.commit()
|
||
return row
|
||
|
||
def revoke(self, token: str) -> None:
|
||
if not token:
|
||
return
|
||
with self.connect() as db:
|
||
db.execute("DELETE FROM auth_tokens WHERE token_digest=?", (token_hash(token),))
|
||
db.commit()
|
||
|
||
def config(self) -> sqlite3.Row:
|
||
with self.connect() as db:
|
||
return db.execute(
|
||
"""SELECT c.*, u.username AS updated_by_name
|
||
FROM model_config c LEFT JOIN users u ON u.id=c.updated_by WHERE c.id=1"""
|
||
).fetchone()
|
||
|
||
def save_config(self, config: dict[str, Any], user_id: int, ip: str) -> int:
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
current = db.execute("SELECT version FROM model_config WHERE id=1").fetchone()
|
||
version = int(current["version"]) + 1
|
||
db.execute(
|
||
"""UPDATE model_config SET config_json=?,version=?,updated_at=?,updated_by=?
|
||
WHERE id=1""",
|
||
(json.dumps(config, ensure_ascii=False), version, now_text(), user_id),
|
||
)
|
||
self._audit(db, user_id, "config.update", f"version={version}", ip)
|
||
db.commit()
|
||
return version
|
||
|
||
@staticmethod
|
||
def _audit(
|
||
db: sqlite3.Connection, user_id: int | None, action: str, detail: str, ip: str
|
||
) -> None:
|
||
db.execute(
|
||
"INSERT INTO audit_log(user_id,action,detail,ip_address,created_at) VALUES(?,?,?,?,?)",
|
||
(user_id, action, detail[:500], ip[:80], now_text()),
|
||
)
|
||
|
||
def audit(self, user_id: int | None, action: str, detail: str, ip: str) -> None:
|
||
with self.connect() as db:
|
||
self._audit(db, user_id, action, detail, ip)
|
||
db.commit()
|
||
|
||
def list_users(self) -> list[sqlite3.Row]:
|
||
with self.connect() as db:
|
||
return list(db.execute("SELECT * FROM users ORDER BY id"))
|
||
|
||
def recent_audit(self, limit: int = 12) -> list[sqlite3.Row]:
|
||
with self.connect() as db:
|
||
return list(
|
||
db.execute(
|
||
"""SELECT a.*, u.username FROM audit_log a
|
||
LEFT JOIN users u ON u.id=a.user_id ORDER BY a.id DESC LIMIT ?""",
|
||
(limit,),
|
||
)
|
||
)
|
||
|
||
def create_user(
|
||
self, username: str, password: str, role: str, actor_id: int, ip: str
|
||
) -> None:
|
||
salt, digest = password_hash(password)
|
||
stamp = now_text()
|
||
with self.connect() as db:
|
||
db.execute(
|
||
"""INSERT INTO users
|
||
(username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
|
||
VALUES(?,?,?,?,1,1,?,?)""",
|
||
(username, salt, digest, role, stamp, stamp),
|
||
)
|
||
self._audit(db, actor_id, "user.create", f"username={username}, role={role}", ip)
|
||
db.commit()
|
||
|
||
def update_user(
|
||
self,
|
||
user_id: int,
|
||
role: str,
|
||
active: bool,
|
||
new_password: str,
|
||
actor_id: int,
|
||
ip: str,
|
||
) -> None:
|
||
with self.connect() as db:
|
||
target = db.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
|
||
if not target:
|
||
raise ValueError("用户不存在")
|
||
if user_id == actor_id and not active:
|
||
raise ValueError("不能停用当前登录账号")
|
||
if target["role"] == "admin" and (role != "admin" or not active):
|
||
count = db.execute(
|
||
"SELECT COUNT(*) FROM users WHERE role='admin' AND active=1"
|
||
).fetchone()[0]
|
||
if count <= 1:
|
||
raise ValueError("系统至少需要保留一个启用的管理员")
|
||
values: list[Any] = [role, int(active), now_text()]
|
||
sql = "UPDATE users SET role=?,active=?,updated_at=?"
|
||
if new_password:
|
||
salt, digest = password_hash(new_password)
|
||
sql += ",password_salt=?,password_digest=?,must_change_password=1"
|
||
values.extend([salt, digest])
|
||
sql += " WHERE id=?"
|
||
values.append(user_id)
|
||
db.execute(sql, values)
|
||
if not active or new_password:
|
||
db.execute("DELETE FROM auth_tokens WHERE user_id=?", (user_id,))
|
||
self._audit(
|
||
db,
|
||
actor_id,
|
||
"user.update",
|
||
f"username={target['username']}, role={role}, active={int(active)}, password_reset={bool(new_password)}",
|
||
ip,
|
||
)
|
||
db.commit()
|
||
|
||
def change_password(self, user_id: int, current: str, new_password: str, ip: str) -> None:
|
||
with self.connect() as db:
|
||
user = db.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
|
||
if not user or not verify_password(
|
||
current, user["password_salt"], user["password_digest"]
|
||
):
|
||
raise ValueError("当前密码不正确")
|
||
salt, digest = password_hash(new_password)
|
||
db.execute(
|
||
"""UPDATE users SET password_salt=?,password_digest=?,must_change_password=0,
|
||
updated_at=? WHERE id=?""",
|
||
(salt, digest, now_text(), user_id),
|
||
)
|
||
self._audit(db, user_id, "password.change", "", ip)
|
||
db.commit()
|
||
|
||
def reset_admin_password(self, password: str) -> None:
|
||
salt, digest = password_hash(password)
|
||
with self.connect() as db:
|
||
result = db.execute(
|
||
"""UPDATE users SET password_salt=?,password_digest=?,active=1,
|
||
must_change_password=1,updated_at=? WHERE username='admin'""",
|
||
(salt, digest, now_text()),
|
||
)
|
||
if result.rowcount == 0:
|
||
stamp = now_text()
|
||
db.execute(
|
||
"""INSERT INTO users
|
||
(username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
|
||
VALUES('admin',?,?,'admin',1,1,?,?)""",
|
||
(salt, digest, stamp, stamp),
|
||
)
|
||
db.execute(
|
||
"DELETE FROM auth_tokens WHERE user_id=(SELECT id FROM users WHERE username='admin')"
|
||
)
|
||
db.commit()
|
||
|
||
|
||
class LoginLimiter:
|
||
def __init__(self):
|
||
self._attempts: dict[str, deque[float]] = defaultdict(deque)
|
||
self._lock = threading.Lock()
|
||
|
||
def allowed(self, key: str) -> bool:
|
||
cutoff = time.time() - 300
|
||
with self._lock:
|
||
items = self._attempts[key]
|
||
while items and items[0] < cutoff:
|
||
items.popleft()
|
||
return len(items) < 8
|
||
|
||
def failure(self, key: str) -> None:
|
||
with self._lock:
|
||
self._attempts[key].append(time.time())
|
||
|
||
def success(self, key: str) -> None:
|
||
with self._lock:
|
||
self._attempts.pop(key, None)
|
||
|
||
|
||
LOGIN_LIMITER = LoginLimiter()
|
||
|
||
|
||
BASE_CSS = """
|
||
:root{--bg:#f3f7f5;--surface:#fff;--ink:#17251f;--muted:#67786f;--line:#dce7e1;
|
||
--accent:#0d9871;--deep:#102b21;--danger:#c94352;--soft:#e1f4ec}
|
||
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font-family:"Microsoft YaHei UI","Segoe UI",sans-serif}
|
||
a{color:var(--accent);text-decoration:none}.shell{min-height:100vh;display:grid;grid-template-columns:250px 1fr}
|
||
aside{background:var(--deep);color:white;padding:30px 24px;position:sticky;top:0;height:100vh}.brand{font-size:22px;font-weight:800;letter-spacing:.04em}
|
||
.brand small{display:block;margin-top:8px;color:#9fc0b3;font-size:12px;font-weight:500}.userbox{margin-top:44px;padding:16px;background:#17392c;border:1px solid #2b5143;border-radius:14px}
|
||
.role{display:inline-block;margin-top:8px;padding:4px 9px;background:#275643;border-radius:999px;color:#cce9de;font-size:12px}
|
||
nav{margin-top:30px;display:grid;gap:8px}nav a{color:#c6d8d0;padding:10px 12px;border-radius:9px}nav a:hover{background:#1c4435;color:white}
|
||
main{padding:34px;max-width:1450px;width:100%;margin:0 auto}.top{display:flex;justify-content:space-between;align-items:end;margin-bottom:24px}.top h1{font-size:30px;margin:0 0 7px}.muted{color:var(--muted);font-size:13px}
|
||
.grid{display:grid;grid-template-columns:1.35fr .9fr;gap:20px;align-items:start}.card{background:var(--surface);border:1px solid var(--line);border-radius:17px;padding:22px;margin-bottom:20px;box-shadow:0 10px 30px rgba(17,51,38,.035)}
|
||
.card h2{font-size:18px;margin:0 0 5px}.cardhead{display:flex;justify-content:space-between;gap:15px;align-items:start;margin-bottom:18px}.version{font:600 12px Consolas,monospace;color:var(--accent);background:var(--soft);padding:6px 9px;border-radius:8px}
|
||
.formgrid{display:grid;grid-template-columns:1fr 1fr;gap:14px}.full{grid-column:1/-1}label{display:block;color:#53675d;font-size:13px;margin-bottom:6px}input,select,textarea{width:100%;border:1px solid #ceddd5;background:#fbfcfb;border-radius:9px;padding:10px 12px;color:var(--ink);font:inherit;outline:none}
|
||
input:focus,select:focus,textarea:focus{border-color:var(--accent);box-shadow:0 0 0 3px #dff3eb}textarea{min-height:150px;font:13px Consolas,monospace;resize:vertical}.switches{display:grid;grid-template-columns:1fr 1fr;gap:10px 18px}.check{display:flex;align-items:center;gap:9px;color:var(--ink)}.check input{width:18px;height:18px;accent-color:var(--accent)}
|
||
.actions{display:flex;justify-content:flex-end;margin-top:18px;gap:10px}button,.button{border:0;border-radius:9px;padding:10px 17px;font:600 14px inherit;cursor:pointer;background:var(--accent);color:white}button:hover{filter:brightness(.93)}button.secondary{background:#edf3f0;color:#28483b}.danger{color:var(--danger)}
|
||
.flash{padding:12px 15px;border-radius:10px;margin-bottom:18px;background:var(--soft);color:#087656;border:1px solid #c9eadc}.flash.error{background:#fbeaec;color:var(--danger);border-color:#f2cbd1}
|
||
.user{display:grid;grid-template-columns:1.1fr .8fr .6fr 1.1fr auto;gap:8px;align-items:end;padding:12px 0;border-top:1px solid #edf2ef}.user:first-of-type{border-top:0}.username{font-weight:700;padding:11px 0}.user input,.user select{padding:8px 9px}.tiny{font-size:11px;color:#829189}
|
||
.audit{display:grid;gap:12px}.event{border-left:3px solid #b8ddce;padding-left:11px}.event b{font-size:13px}.event div{font-size:11px;color:#829189;margin-top:3px}
|
||
.loginwrap{min-height:100vh;display:grid;place-items:center;padding:24px}.login{width:min(430px,100%);background:white;border:1px solid var(--line);border-radius:22px;padding:34px;box-shadow:0 25px 70px rgba(12,48,34,.12)}.login h1{margin:0 0 8px;font-size:27px}.login form{display:grid;gap:15px;margin-top:26px}.login button{margin-top:5px;padding:12px}.notice{background:#fff7e8;border:1px solid #f0d6a5;color:#8a5a0a;padding:13px;border-radius:10px;font-size:13px;margin-bottom:18px}
|
||
@media(max-width:980px){.shell{grid-template-columns:1fr}aside{height:auto;position:static}.grid{grid-template-columns:1fr}.user{grid-template-columns:1fr 1fr}.user .actions{grid-column:1/-1}.formgrid{grid-template-columns:1fr}.full{grid-column:auto}main{padding:20px}}
|
||
"""
|
||
|
||
|
||
def page(title: str, body: str) -> str:
|
||
return (
|
||
"<!doctype html><html lang='zh-CN'><head><meta charset='utf-8'>"
|
||
"<meta name='viewport' content='width=device-width,initial-scale=1'>"
|
||
f"<title>{html.escape(title)}</title><style>{BASE_CSS}</style></head><body>{body}</body></html>"
|
||
)
|
||
|
||
|
||
class AdminServer(ThreadingHTTPServer):
|
||
daemon_threads = True
|
||
|
||
def __init__(self, address: tuple[str, int], database: Database):
|
||
self.local_sync_token = secrets.token_urlsafe(36)
|
||
super().__init__(address, AdminHandler)
|
||
self.database = database
|
||
|
||
|
||
def _address_in_use(exc: OSError) -> bool:
|
||
"""兼容 Windows、Linux 和 macOS 的端口占用错误码。"""
|
||
# Windows 在独占地址被另一个进程监听时，可能返回 10013 而不是 10048。
|
||
return exc.errno in (13, 48, 98, 10013, 10048) or getattr(
|
||
exc, "winerror", None
|
||
) in (10013, 10048)
|
||
|
||
|
||
def create_server(
|
||
host: str,
|
||
preferred_port: int,
|
||
database: Database,
|
||
*,
|
||
max_attempts: int = DEFAULT_PORT_ATTEMPTS,
|
||
) -> tuple[AdminServer, int]:
|
||
"""创建服务；首选端口被占用时自动向后尝试可用端口。"""
|
||
if not 0 <= preferred_port <= 65535:
|
||
raise ValueError("端口必须在 0-65535 之间")
|
||
if max_attempts < 1:
|
||
raise ValueError("端口尝试次数必须大于 0")
|
||
if preferred_port == 0:
|
||
server = AdminServer((host, 0), database)
|
||
return server, int(server.server_address[1])
|
||
|
||
last_error: OSError | None = None
|
||
for port in range(preferred_port, min(65535, preferred_port + max_attempts - 1) + 1):
|
||
try:
|
||
return AdminServer((host, port), database), port
|
||
except OSError as exc:
|
||
if not _address_in_use(exc):
|
||
raise
|
||
last_error = exc
|
||
raise OSError(
|
||
f"端口 {preferred_port}-{min(65535, preferred_port + max_attempts - 1)} 均被占用"
|
||
) from last_error
|
||
|
||
|
||
def public_server_url(host: str, port: int) -> str:
|
||
display_host = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host
|
||
if ":" in display_host and not display_host.startswith("["):
|
||
display_host = f"[{display_host}]"
|
||
return f"http://{display_host}:{port}"
|
||
|
||
|
||
def write_runtime_info(
|
||
path: Path,
|
||
host: str,
|
||
port: int,
|
||
*,
|
||
local_sync_token: str = "",
|
||
) -> dict[str, Any]:
|
||
"""发布实际监听端口，供同项目桌面端自动发现。"""
|
||
info = {
|
||
"pid": os.getpid(),
|
||
"host": host,
|
||
"port": int(port),
|
||
"server_url": public_server_url(host, port),
|
||
"local_sync_token": local_sync_token,
|
||
"started_at": now_text(),
|
||
}
|
||
path = path.resolve()
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||
temporary.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
os.replace(temporary, path)
|
||
return info
|
||
|
||
|
||
def clear_runtime_info(path: Path, port: int) -> None:
|
||
"""只清理由当前进程写入的发现文件，避免影响另一个后台实例。"""
|
||
path = path.resolve()
|
||
try:
|
||
info = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError, TypeError):
|
||
return
|
||
if info.get("pid") == os.getpid() and info.get("port") == int(port):
|
||
try:
|
||
path.unlink()
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
class AdminHandler(BaseHTTPRequestHandler):
|
||
server_version = "WeComConfig/1.0"
|
||
|
||
@property
|
||
def db(self) -> Database:
|
||
return self.server.database # type: ignore[attr-defined]
|
||
|
||
@property
|
||
def client_ip(self) -> str:
|
||
forwarded = self.headers.get("X-Forwarded-For", "").split(",", 1)[0].strip()
|
||
return forwarded or self.client_address[0]
|
||
|
||
def log_message(self, format_string: str, *args: Any) -> None:
|
||
sys.stdout.write(
|
||
f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {self.client_ip} {format_string % args}\n"
|
||
)
|
||
|
||
def _security_headers(self) -> None:
|
||
self.send_header("X-Content-Type-Options", "nosniff")
|
||
self.send_header("X-Frame-Options", "DENY")
|
||
self.send_header("Referrer-Policy", "no-referrer")
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header(
|
||
"Content-Security-Policy",
|
||
"default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; base-uri 'none'; frame-ancestors 'none'",
|
||
)
|
||
|
||
def _send(self, status: int, content: str, content_type: str) -> None:
|
||
data = content.encode("utf-8")
|
||
self.send_response(status)
|
||
self._security_headers()
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def json_response(self, status: int, data: dict[str, Any]) -> None:
|
||
self._send(
|
||
status,
|
||
json.dumps(data, ensure_ascii=False),
|
||
"application/json; charset=utf-8",
|
||
)
|
||
|
||
def html_response(self, status: int, content: str) -> None:
|
||
self._send(status, content, "text/html; charset=utf-8")
|
||
|
||
def redirect(self, location: str, *, cookie: str = "") -> None:
|
||
self.send_response(HTTPStatus.SEE_OTHER)
|
||
self._security_headers()
|
||
self.send_header("Location", location)
|
||
if cookie:
|
||
self.send_header("Set-Cookie", cookie)
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def body_bytes(self, limit: int = 512_000) -> bytes:
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
except ValueError:
|
||
raise ValueError("Content-Length 无效")
|
||
if length < 0 or length > limit:
|
||
raise ValueError("请求内容过大")
|
||
return self.rfile.read(length)
|
||
|
||
def json_body(self) -> dict[str, Any]:
|
||
try:
|
||
data = json.loads(self.body_bytes().decode("utf-8") or "{}")
|
||
except (UnicodeDecodeError, ValueError) as exc:
|
||
raise ValueError("请求 JSON 无效") from exc
|
||
if not isinstance(data, dict):
|
||
raise ValueError("请求 JSON 必须是对象")
|
||
return data
|
||
|
||
def form_body(self) -> dict[str, str]:
|
||
try:
|
||
parsed = urllib.parse.parse_qs(
|
||
self.body_bytes().decode("utf-8"), keep_blank_values=True
|
||
)
|
||
except UnicodeDecodeError as exc:
|
||
raise ValueError("表单编码无效") from exc
|
||
return {key: values[-1] for key, values in parsed.items()}
|
||
|
||
def bearer_token(self) -> str:
|
||
authorization = self.headers.get("Authorization", "")
|
||
if authorization.startswith("Bearer "):
|
||
return authorization[7:].strip()
|
||
return ""
|
||
|
||
def cookie_token(self) -> str:
|
||
cookie = SimpleCookie()
|
||
try:
|
||
cookie.load(self.headers.get("Cookie", ""))
|
||
except Exception:
|
||
return ""
|
||
return cookie.get("session_token").value if cookie.get("session_token") else ""
|
||
|
||
def auth(self, api: bool = False) -> tuple[sqlite3.Row | None, str]:
|
||
token = self.bearer_token() if api else self.cookie_token()
|
||
return self.db.session(token), token
|
||
|
||
@staticmethod
|
||
def valid_password(value: str) -> bool:
|
||
return len(value) >= 10 and any(c.isalpha() for c in value) and any(
|
||
c.isdigit() for c in value
|
||
)
|
||
|
||
def require_web_auth(
|
||
self, *, roles: tuple[str, ...] | None = None, allow_password_change: bool = False
|
||
) -> tuple[sqlite3.Row, str] | None:
|
||
user, token = self.auth(False)
|
||
if not user:
|
||
self.redirect("/login")
|
||
return None
|
||
if user["must_change_password"] and not allow_password_change:
|
||
self.redirect("/?error=" + urllib.parse.quote("首次登录请先修改密码"))
|
||
return None
|
||
if roles and user["role"] not in roles:
|
||
self.redirect("/?error=" + urllib.parse.quote("当前角色没有执行此操作的权限"))
|
||
return None
|
||
return user, token
|
||
|
||
def require_api_auth(self) -> sqlite3.Row | None:
|
||
user, _ = self.auth(True)
|
||
if not user:
|
||
self.json_response(HTTPStatus.UNAUTHORIZED, {"error": "登录已失效，请重新登录"})
|
||
return None
|
||
if user["must_change_password"]:
|
||
self.json_response(
|
||
HTTPStatus.FORBIDDEN, {"error": "请先在后台网页修改初始密码"}
|
||
)
|
||
return None
|
||
return user
|
||
|
||
def local_sync_authorized(self) -> bool:
|
||
"""允许同机桌面端使用运行时随机凭证只读模型配置。"""
|
||
supplied = self.headers.get("X-Desktop-Sync-Token", "")
|
||
expected = getattr(self.server, "local_sync_token", "")
|
||
if not supplied or not expected:
|
||
return False
|
||
try:
|
||
is_loopback = ipaddress.ip_address(self.client_address[0]).is_loopback
|
||
except ValueError:
|
||
return False
|
||
return is_loopback and hmac.compare_digest(supplied, expected)
|
||
|
||
@staticmethod
|
||
def csrf_ok(user: sqlite3.Row, form: dict[str, str]) -> bool:
|
||
return hmac.compare_digest(str(user["csrf_token"]), str(form.get("csrf", "")))
|
||
|
||
def do_GET(self) -> None:
|
||
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
|
||
if path == "/health":
|
||
self.json_response(HTTPStatus.OK, {"status": "ok", "time": now_text()})
|
||
elif path == "/login":
|
||
user, _ = self.auth(False)
|
||
if user:
|
||
self.redirect("/")
|
||
else:
|
||
self.render_login()
|
||
elif path == "/":
|
||
auth = self.require_web_auth(allow_password_change=True)
|
||
if auth:
|
||
self.render_dashboard(auth[0])
|
||
elif path == "/api/v1/me":
|
||
user = self.require_api_auth()
|
||
if user:
|
||
self.json_response(HTTPStatus.OK, {"user": self.public_user(user)})
|
||
elif path == "/api/v1/config":
|
||
local_sync = self.local_sync_authorized()
|
||
user = None if local_sync else self.require_api_auth()
|
||
if local_sync or user:
|
||
row = self.db.config()
|
||
self.json_response(
|
||
HTTPStatus.OK,
|
||
{
|
||
"version": row["version"],
|
||
"updated_at": row["updated_at"],
|
||
"updated_by": row["updated_by_name"] or "system",
|
||
"config": json.loads(row["config_json"]),
|
||
},
|
||
)
|
||
else:
|
||
self.json_response(HTTPStatus.NOT_FOUND, {"error": "页面不存在"})
|
||
|
||
def do_POST(self) -> None:
|
||
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
|
||
try:
|
||
if path == "/login":
|
||
self.web_login()
|
||
elif path == "/logout":
|
||
self.web_logout()
|
||
elif path == "/admin/config":
|
||
self.web_save_config()
|
||
elif path == "/admin/users/create":
|
||
self.web_create_user()
|
||
elif path == "/admin/users/update":
|
||
self.web_update_user()
|
||
elif path == "/admin/password":
|
||
self.web_change_password()
|
||
elif path == "/api/v1/auth/login":
|
||
self.api_login()
|
||
elif path == "/api/v1/auth/logout":
|
||
self.api_logout()
|
||
else:
|
||
self.json_response(HTTPStatus.NOT_FOUND, {"error": "接口不存在"})
|
||
except ValueError as exc:
|
||
if path.startswith("/api/"):
|
||
self.json_response(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
|
||
else:
|
||
self.redirect("/?error=" + urllib.parse.quote(str(exc)))
|
||
except sqlite3.IntegrityError:
|
||
if path.startswith("/api/"):
|
||
self.json_response(HTTPStatus.CONFLICT, {"error": "用户名已存在"})
|
||
else:
|
||
self.redirect("/?error=" + urllib.parse.quote("用户名已存在"))
|
||
|
||
@staticmethod
|
||
def public_user(user: sqlite3.Row) -> dict[str, Any]:
|
||
return {
|
||
"id": user["id"],
|
||
"username": user["username"],
|
||
"role": user["role"],
|
||
"role_name": ROLE_LABELS[user["role"]],
|
||
}
|
||
|
||
def render_login(self, error: str = "") -> None:
|
||
message = (
|
||
f"<div class='flash error'>{html.escape(error)}</div>" if error else ""
|
||
)
|
||
body = f"""
|
||
<div class='loginwrap'><section class='login'>
|
||
<div class='brand' style='color:var(--accent)'>ZHEN AI ADMIN</div>
|
||
<h1>配置管理后台</h1><div class='muted'>统一管理桌面端账号、角色和 AI 模型参数。</div>
|
||
{message}
|
||
<form method='post' action='/login'>
|
||
<div><label>用户名</label><input name='username' autocomplete='username' required autofocus></div>
|
||
<div><label>密码</label><input name='password' type='password' autocomplete='current-password' required></div>
|
||
<button type='submit'>登录后台</button>
|
||
</form>
|
||
</section></div>"""
|
||
self.html_response(HTTPStatus.OK, page("登录 · 配置管理后台", body))
|
||
|
||
def web_login(self) -> None:
|
||
form = self.form_body()
|
||
username = form.get("username", "").strip()
|
||
key = f"{self.client_ip}:{username.lower()}"
|
||
if not LOGIN_LIMITER.allowed(key):
|
||
self.render_login("登录失败次数过多，请 5 分钟后再试")
|
||
return
|
||
user = self.db.authenticate(username, form.get("password", ""))
|
||
if not user:
|
||
LOGIN_LIMITER.failure(key)
|
||
self.db.audit(None, "login.failed", f"username={username}", self.client_ip)
|
||
self.render_login("用户名或密码不正确")
|
||
return
|
||
LOGIN_LIMITER.success(key)
|
||
token, _ = self.db.create_token(user["id"], "web", "browser", 12 * 3600)
|
||
self.db.audit(user["id"], "login.web", "", self.client_ip)
|
||
cookie = f"session_token={token}; Path=/; HttpOnly; SameSite=Strict; Max-Age=43200"
|
||
self.redirect("/", cookie=cookie)
|
||
|
||
def web_logout(self) -> None:
|
||
user, token = self.auth(False)
|
||
if user:
|
||
try:
|
||
form = self.form_body()
|
||
except ValueError:
|
||
form = {}
|
||
if self.csrf_ok(user, form):
|
||
self.db.revoke(token)
|
||
self.redirect(
|
||
"/login", cookie="session_token=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0"
|
||
)
|
||
|
||
def api_login(self) -> None:
|
||
data = self.json_body()
|
||
username = str(data.get("username") or "").strip()
|
||
key = f"{self.client_ip}:{username.lower()}"
|
||
if not LOGIN_LIMITER.allowed(key):
|
||
self.json_response(
|
||
HTTPStatus.TOO_MANY_REQUESTS, {"error": "登录失败次数过多，请稍后再试"}
|
||
)
|
||
return
|
||
user = self.db.authenticate(username, str(data.get("password") or ""))
|
||
if not user:
|
||
LOGIN_LIMITER.failure(key)
|
||
self.db.audit(None, "login.failed", f"username={username}", self.client_ip)
|
||
self.json_response(HTTPStatus.UNAUTHORIZED, {"error": "用户名或密码不正确"})
|
||
return
|
||
if user["must_change_password"]:
|
||
self.json_response(
|
||
HTTPStatus.FORBIDDEN, {"error": "首次登录请先在后台网页修改密码"}
|
||
)
|
||
return
|
||
LOGIN_LIMITER.success(key)
|
||
token, _ = self.db.create_token(
|
||
user["id"], "api", str(data.get("device_name") or "desktop"), 30 * 86400
|
||
)
|
||
self.db.audit(user["id"], "login.api", str(data.get("device_name") or ""), self.client_ip)
|
||
self.json_response(
|
||
HTTPStatus.OK,
|
||
{"access_token": token, "expires_in": 30 * 86400, "user": self.public_user(user)},
|
||
)
|
||
|
||
def api_logout(self) -> None:
|
||
user, token = self.auth(True)
|
||
if user:
|
||
self.db.revoke(token)
|
||
self.db.audit(user["id"], "logout.api", "", self.client_ip)
|
||
self.json_response(HTTPStatus.OK, {"ok": True})
|
||
|
||
def web_save_config(self) -> None:
|
||
auth = self.require_web_auth(roles=("admin", "operator"))
|
||
if not auth:
|
||
return
|
||
user, _ = auth
|
||
form = self.form_body()
|
||
if not self.csrf_ok(user, form):
|
||
raise ValueError("页面已过期，请刷新后重试")
|
||
current = json.loads(self.db.config()["config_json"])
|
||
config = validate_config_form(form, current)
|
||
version = self.db.save_config(config, user["id"], self.client_ip)
|
||
self.redirect("/?message=" + urllib.parse.quote(f"配置已发布为 v{version}"))
|
||
|
||
def web_create_user(self) -> None:
|
||
auth = self.require_web_auth(roles=("admin",))
|
||
if not auth:
|
||
return
|
||
user, _ = auth
|
||
form = self.form_body()
|
||
if not self.csrf_ok(user, form):
|
||
raise ValueError("页面已过期，请刷新后重试")
|
||
username = form.get("username", "").strip()
|
||
password = form.get("password", "")
|
||
role = form.get("role", "viewer")
|
||
if not (3 <= len(username) <= 40) or not all(
|
||
char.isalnum() or char in "_.-" for char in username
|
||
):
|
||
raise ValueError("用户名需为 3-40 位字母、数字、点、下划线或短横线")
|
||
if role not in ROLES:
|
||
raise ValueError("角色无效")
|
||
if not self.valid_password(password):
|
||
raise ValueError("初始密码至少 10 位，并同时包含字母和数字")
|
||
self.db.create_user(username, password, role, user["id"], self.client_ip)
|
||
self.redirect("/?message=" + urllib.parse.quote(f"用户 {username} 已创建"))
|
||
|
||
def web_update_user(self) -> None:
|
||
auth = self.require_web_auth(roles=("admin",))
|
||
if not auth:
|
||
return
|
||
user, _ = auth
|
||
form = self.form_body()
|
||
if not self.csrf_ok(user, form):
|
||
raise ValueError("页面已过期，请刷新后重试")
|
||
try:
|
||
user_id = int(form.get("user_id", "0"))
|
||
except ValueError as exc:
|
||
raise ValueError("用户编号无效") from exc
|
||
role = form.get("role", "viewer")
|
||
password = form.get("new_password", "")
|
||
if role not in ROLES:
|
||
raise ValueError("角色无效")
|
||
if password and not self.valid_password(password):
|
||
raise ValueError("重置密码至少 10 位，并同时包含字母和数字")
|
||
self.db.update_user(
|
||
user_id,
|
||
role,
|
||
form.get("active") == "1",
|
||
password,
|
||
user["id"],
|
||
self.client_ip,
|
||
)
|
||
self.redirect("/?message=" + urllib.parse.quote("用户资料已更新"))
|
||
|
||
def web_change_password(self) -> None:
|
||
auth = self.require_web_auth(allow_password_change=True)
|
||
if not auth:
|
||
return
|
||
user, _ = auth
|
||
form = self.form_body()
|
||
if not self.csrf_ok(user, form):
|
||
raise ValueError("页面已过期，请刷新后重试")
|
||
new_password = form.get("new_password", "")
|
||
if new_password != form.get("confirm_password", ""):
|
||
raise ValueError("两次输入的新密码不一致")
|
||
if not self.valid_password(new_password):
|
||
raise ValueError("新密码至少 10 位，并同时包含字母和数字")
|
||
self.db.change_password(
|
||
user["id"], form.get("current_password", ""), new_password, self.client_ip
|
||
)
|
||
self.redirect("/?message=" + urllib.parse.quote("密码已修改"))
|
||
|
||
def render_dashboard(self, user: sqlite3.Row) -> None:
|
||
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
|
||
message = query.get("message", [""])[-1]
|
||
error = query.get("error", [""])[-1]
|
||
flash = ""
|
||
if message:
|
||
flash = f"<div class='flash'>{html.escape(message)}</div>"
|
||
elif error:
|
||
flash = f"<div class='flash error'>{html.escape(error)}</div>"
|
||
csrf = html.escape(user["csrf_token"], quote=True)
|
||
password_card = self.password_card(user, csrf)
|
||
if user["must_change_password"]:
|
||
content = f"""
|
||
<div class='shell'>{self.sidebar(user, csrf)}<main>
|
||
<div class='top'><div><h1>首次登录</h1><div class='muted'>修改初始密码后才能使用配置和同步功能。</div></div></div>
|
||
{flash}<div style='max-width:650px'>{password_card}</div>
|
||
</main></div>"""
|
||
self.html_response(HTTPStatus.OK, page("首次登录 · 配置管理后台", content))
|
||
return
|
||
row = self.db.config()
|
||
config = json.loads(row["config_json"])
|
||
config_card = self.config_card(user, csrf, row, config)
|
||
users_card = self.users_card(user, csrf)
|
||
audit_card = self.audit_card()
|
||
content = f"""
|
||
<div class='shell'>{self.sidebar(user, csrf)}<main>
|
||
<div class='top'><div><h1>模型配置中心</h1><div class='muted'>保存后，已登录桌面端会在启动或定时同步时自动应用。</div></div>
|
||
<div class='version'>CURRENT · v{int(row['version'])}</div></div>
|
||
{flash}<div class='grid'><section>{config_card}</section><section>{users_card}{password_card}{audit_card}</section></div>
|
||
</main></div>"""
|
||
self.html_response(HTTPStatus.OK, page("模型配置中心", content))
|
||
|
||
@staticmethod
|
||
def sidebar(user: sqlite3.Row, csrf: str) -> str:
|
||
return f"""
|
||
<aside><div class='brand'>ZHEN AI ADMIN<small>企微客服统一配置中心</small></div>
|
||
<div class='userbox'><b>{html.escape(user['username'])}</b><br><span class='role'>{ROLE_LABELS[user['role']]}</span></div>
|
||
<nav><a href='/'>模型配置</a><a href='#users'>用户与角色</a><a href='#security'>账号安全</a></nav>
|
||
<form method='post' action='/logout' style='position:absolute;bottom:25px;left:24px;right:24px'>
|
||
<input type='hidden' name='csrf' value='{csrf}'><button class='secondary' style='width:100%'>退出登录</button></form></aside>"""
|
||
|
||
@staticmethod
|
||
def password_card(user: sqlite3.Row, csrf: str) -> str:
|
||
notice = (
|
||
"<div class='notice'>当前仍在使用初始密码，必须修改后才能让桌面端登录。</div>"
|
||
if user["must_change_password"]
|
||
else ""
|
||
)
|
||
return f"""
|
||
<section class='card' id='security'><div class='cardhead'><div><h2>修改密码</h2><div class='muted'>密码至少 10 位，需包含字母和数字。</div></div></div>{notice}
|
||
<form method='post' action='/admin/password'><input type='hidden' name='csrf' value='{csrf}'>
|
||
<div class='formgrid'><div class='full'><label>当前密码</label><input type='password' name='current_password' required></div>
|
||
<div><label>新密码</label><input type='password' name='new_password' required></div>
|
||
<div><label>确认新密码</label><input type='password' name='confirm_password' required></div></div>
|
||
<div class='actions'><button type='submit'>更新密码</button></div></form></section>"""
|
||
|
||
@staticmethod
|
||
def config_card(
|
||
user: sqlite3.Row, csrf: str, row: sqlite3.Row, config: dict[str, Any]
|
||
) -> str:
|
||
normalized = load_initial_config()
|
||
normalized.update(
|
||
{key: config[key] for key in CONFIG_KEYS if key in config}
|
||
)
|
||
config = normalized
|
||
can_edit = user["role"] in ("admin", "operator")
|
||
esc = lambda key: html.escape(
|
||
"" if config.get(key) is None else str(config.get(key, "")), quote=True
|
||
)
|
||
checked = lambda key: " checked" if config.get(key) else ""
|
||
selected = lambda key, value: " selected" if config.get(key) == value else ""
|
||
disabled = " disabled" if not can_edit else ""
|
||
mcp = html.escape(
|
||
json.dumps(config.get("AI_MCP_SERVERS", []), ensure_ascii=False, indent=2)
|
||
)
|
||
dify_inputs = html.escape(
|
||
json.dumps(
|
||
config.get("GROK_DIFY_INPUTS", {}),
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
)
|
||
submit = (
|
||
"<div class='actions'><button type='submit'>保存并发布配置</button></div>"
|
||
if can_edit
|
||
else "<div class='notice'>当前为只读角色，可查看配置但不能修改。</div>"
|
||
)
|
||
return f"""
|
||
<section class='card'><div class='cardhead'><div><h2>能力开关</h2><div class='muted'>控制桌面端自动回复、上下文与工具能力。</div></div><span class='version'>v{row['version']}</span></div>
|
||
<form method='post' action='/admin/config'><input type='hidden' name='csrf' value='{csrf}'>
|
||
<div class='switches'>
|
||
<label class='check'><input type='checkbox' name='AI_ENABLED' value='1'{checked('AI_ENABLED')}{disabled}>启用 AI 回复</label>
|
||
<label class='check'><input type='checkbox' name='AI_CONTEXT_ENABLED' value='1'{checked('AI_CONTEXT_ENABLED')}{disabled}>启用会话上下文</label>
|
||
<label class='check'><input type='checkbox' name='AI_USE_VISION' value='1'{checked('AI_USE_VISION')}{disabled}>启用视觉模式</label>
|
||
<label class='check'><input type='checkbox' name='AI_COUNTER_INSULT_ENABLED' value='1'{checked('AI_COUNTER_INSULT_ENABLED')}{disabled}>启用反辱骂策略</label>
|
||
<label class='check'><input type='checkbox' name='AI_MCP_ENABLED' value='1'{checked('AI_MCP_ENABLED')}{disabled}>启用 MCP 工具</label>
|
||
<label class='check'><input type='checkbox' name='GROK_CUSTOMER_SERVICE_ENABLED' value='1'{checked('GROK_CUSTOMER_SERVICE_ENABLED')}{disabled}>启用 Grok Build 本地客服 Agent</label>
|
||
<label class='check'><input type='checkbox' name='GROK_MODEL_ENABLED' value='1'{checked('GROK_MODEL_ENABLED')}{disabled}>启用 Grok Agent 自有模型（必需）</label>
|
||
</div><div style='height:24px'></div>
|
||
<div class='cardhead'><div><h2>Grok Build 客服 Agent</h2><div class='muted'>Grok Build 只负责 Agent 调度，回复由下方后台自有模型生成。Agent 只能调用项目内置的受控客服 MCP，不嵌入外部聊天网页，也不具备消息发送能力。</div></div></div>
|
||
<div class='formgrid'>
|
||
<div><label>单次回复超时（秒）</label><input type='number' min='30' max='600' name='GROK_CUSTOMER_SERVICE_TIMEOUT' value='{esc('GROK_CUSTOMER_SERVICE_TIMEOUT')}'{disabled}></div>
|
||
<div><label>最多 Agent 轮数</label><input type='number' min='2' max='30' name='GROK_CUSTOMER_SERVICE_MAX_TURNS' value='{esc('GROK_CUSTOMER_SERVICE_MAX_TURNS')}'{disabled}></div>
|
||
<div><label>推理强度</label><select name='GROK_CUSTOMER_SERVICE_EFFORT'{disabled}>
|
||
<option value='low'{selected('GROK_CUSTOMER_SERVICE_EFFORT', 'low')}>low（低延迟）</option>
|
||
<option value='medium'{selected('GROK_CUSTOMER_SERVICE_EFFORT', 'medium')}>medium</option>
|
||
<option value='high'{selected('GROK_CUSTOMER_SERVICE_EFFORT', 'high')}>high</option>
|
||
</select></div>
|
||
</div><div style='height:24px'></div>
|
||
<div class='cardhead'><div><h2>视觉模型与客服身份</h2><div class='muted'>这一组 API 只供可选视觉能力使用，不会作为 Grok Agent 的模型回退。API Key 留空表示保持当前值；页面永不回显密钥。</div></div></div>
|
||
<div class='formgrid'>
|
||
<div><label>视觉 API 地址（可选）</label><input name='AI_API_BASE' value='{esc('AI_API_BASE')}'{disabled}></div>
|
||
<div><label>视觉模型名称</label><input name='AI_MODEL' value='{esc('AI_MODEL')}'{disabled}></div>
|
||
<div><label>API Key</label><input type='password' name='AI_API_KEY' placeholder='已保存；留空不修改'{disabled}></div>
|
||
<div><label>客服名称</label><input name='AI_AGENT_NAME' value='{esc('AI_AGENT_NAME')}' required{disabled}></div>
|
||
<div class='full'><label>机构名称</label><input name='AI_HOSPITAL_NAME' value='{esc('AI_HOSPITAL_NAME')}' required{disabled}></div>
|
||
<div><label>上下文轮数</label><input type='number' min='1' max='50' name='AI_CONTEXT_MAX_ROUNDS' value='{esc('AI_CONTEXT_MAX_ROUNDS')}'{disabled}></div>
|
||
<div><label>最大回复 tokens</label><input type='number' min='50' max='32000' name='AI_MAX_TOKENS' value='{esc('AI_MAX_TOKENS')}'{disabled}></div>
|
||
<div><label>温度</label><input type='number' min='0' max='2' step='.05' name='AI_TEMPERATURE' value='{esc('AI_TEMPERATURE')}'{disabled}></div>
|
||
<div><label>请求超时（秒）</label><input type='number' min='5' max='600' name='AI_TIMEOUT' value='{esc('AI_TIMEOUT')}'{disabled}></div>
|
||
</div><div style='height:24px'></div>
|
||
<div class='cardhead'><div><h2>Grok Agent 自有模型</h2><div class='muted'>这是 Agent 唯一允许使用的模型，不会回退到 Grok/xAI。原生接口直接传递工具调用；Dify Chat Messages 会由桌面端本地适配为 Grok 工具调用协议。远程 Dify 强烈建议使用 HTTPS；API Key 留空表示保持当前值。</div></div></div>
|
||
<div class='formgrid'>
|
||
<div><label>自有模型 API 基址</label><input name='GROK_API_BASE' value='{esc('GROK_API_BASE')}' placeholder='https://model.example.com/v1'{disabled}></div>
|
||
<div><label>自有模型名称</label><input name='GROK_MODEL' value='{esc('GROK_MODEL')}' placeholder='qwen-coder'{disabled}></div>
|
||
<div><label>自有模型 API Key</label><input type='password' name='GROK_API_KEY' placeholder='已保存；留空不修改'{disabled}></div>
|
||
<div><label>接口协议</label><select name='GROK_API_BACKEND'{disabled}>
|
||
<option value='chat_completions'{selected('GROK_API_BACKEND', 'chat_completions')}>OpenAI Chat Completions</option>
|
||
<option value='responses'{selected('GROK_API_BACKEND', 'responses')}>OpenAI Responses</option>
|
||
<option value='messages'{selected('GROK_API_BACKEND', 'messages')}>Anthropic Messages</option>
|
||
<option value='dify'{selected('GROK_API_BACKEND', 'dify')}>Dify Chat Messages（本地工具调用适配）</option>
|
||
</select></div>
|
||
<div><label>认证方式</label><select name='GROK_AUTH_SCHEME'{disabled}>
|
||
<option value='auto'{selected('GROK_AUTH_SCHEME', 'auto')}>自动（Anthropic 官方域名使用 x-api-key）</option>
|
||
<option value='bearer'{selected('GROK_AUTH_SCHEME', 'bearer')}>Authorization Bearer</option>
|
||
<option value='x_api_key'{selected('GROK_AUTH_SCHEME', 'x_api_key')}>x-api-key</option>
|
||
</select></div>
|
||
<div><label>上下文窗口 tokens</label><input type='number' min='4096' max='2000000' name='GROK_CONTEXT_WINDOW' value='{esc('GROK_CONTEXT_WINDOW')}'{disabled}></div>
|
||
<div><label>最大输出 tokens</label><input type='number' min='64' max='262144' name='GROK_MAX_TOKENS' value='{esc('GROK_MAX_TOKENS')}'{disabled}></div>
|
||
<div><label>自有模型温度</label><input type='number' min='0' max='2' step='.05' name='GROK_TEMPERATURE' value='{esc('GROK_TEMPERATURE')}'{disabled}></div>
|
||
<div class='full'><label>Dify inputs JSON（可选）</label><textarea name='GROK_DIFY_INPUTS' placeholder='{{}}'{disabled}>{dify_inputs}</textarea><div class='tiny'>仅供 Dify 应用固定输入变量；请勿在此填写密钥。其他协议会忽略此项。</div></div>
|
||
</div><div style='height:24px'></div>
|
||
<div class='cardhead'><div><h2>MCP 服务器</h2><div class='muted'>填写 JSON 数组，与模型配置一同下发。</div></div></div>
|
||
<textarea name='AI_MCP_SERVERS'{disabled}>{mcp}</textarea>
|
||
<div style='max-width:240px;margin-top:12px'><label>单次最多工具轮数</label><input type='number' min='1' max='20' name='AI_MCP_MAX_ROUNDS' value='{esc('AI_MCP_MAX_ROUNDS')}'{disabled}></div>
|
||
{submit}</form><div class='tiny'>最后更新：{html.escape(row['updated_at'])} · {html.escape(row['updated_by_name'] or 'system')}</div></section>"""
|
||
|
||
def users_card(self, user: sqlite3.Row, csrf: str) -> str:
|
||
if user["role"] != "admin":
|
||
return f"""<section class='card' id='users'><h2>用户与角色</h2><div class='muted' style='margin-top:8px'>仅管理员可以管理登录账号。</div></section>"""
|
||
items = []
|
||
role_options = lambda current: "".join(
|
||
f"<option value='{role}'{' selected' if role == current else ''}>{ROLE_LABELS[role]}</option>"
|
||
for role in ROLES
|
||
)
|
||
for item in self.db.list_users():
|
||
active = " checked" if item["active"] else ""
|
||
items.append(
|
||
f"""<form class='user' method='post' action='/admin/users/update'>
|
||
<input type='hidden' name='csrf' value='{csrf}'><input type='hidden' name='user_id' value='{item['id']}'>
|
||
<div><div class='username'>{html.escape(item['username'])}</div><div class='tiny'>创建于 {html.escape(item['created_at'][:10])}</div></div>
|
||
<div><label>角色</label><select name='role'>{role_options(item['role'])}</select></div>
|
||
<div><label>状态</label><label class='check'><input type='checkbox' name='active' value='1'{active}>启用</label></div>
|
||
<div><label>重置密码（可留空）</label><input type='password' name='new_password' placeholder='至少 10 位'></div>
|
||
<div class='actions'><button class='secondary'>保存</button></div></form>"""
|
||
)
|
||
return f"""
|
||
<section class='card' id='users'><div class='cardhead'><div><h2>用户与角色</h2><div class='muted'>管理员管理全部功能，配置员可发布配置，只读用户仅同步。</div></div></div>
|
||
{''.join(items)}<div style='height:20px'></div><h2>新建用户</h2>
|
||
<form method='post' action='/admin/users/create'><input type='hidden' name='csrf' value='{csrf}'>
|
||
<div class='formgrid'><div><label>用户名</label><input name='username' required></div><div><label>角色</label><select name='role'>{role_options('viewer')}</select></div>
|
||
<div class='full'><label>初始密码</label><input type='password' name='password' placeholder='至少 10 位，包含字母和数字' required></div></div>
|
||
<div class='actions'><button type='submit'>创建用户</button></div></form></section>"""
|
||
|
||
def audit_card(self) -> str:
|
||
events = []
|
||
for item in self.db.recent_audit():
|
||
events.append(
|
||
f"<div class='event'><b>{html.escape(item['action'])}</b> · {html.escape(item['username'] or 'anonymous')}"
|
||
f"<div>{html.escape(item['created_at'])} · {html.escape(item['detail'])}</div></div>"
|
||
)
|
||
return f"""<section class='card'><div class='cardhead'><div><h2>最近操作</h2><div class='muted'>登录、配置和用户变更审计。</div></div></div><div class='audit'>{''.join(events) or '<div class="muted">暂无记录</div>'}</div></section>"""
|
||
|
||
|
||
def validate_config_form(form: dict[str, str], current: dict[str, Any]) -> dict[str, Any]:
|
||
config = load_initial_config()
|
||
config.update({key: current[key] for key in CONFIG_KEYS if key in current})
|
||
for key in BOOL_KEYS:
|
||
config[key] = form.get(key) == "1"
|
||
for key in (
|
||
"AI_API_BASE",
|
||
"AI_MODEL",
|
||
"AI_AGENT_NAME",
|
||
"AI_HOSPITAL_NAME",
|
||
"GROK_API_BASE",
|
||
"GROK_MODEL",
|
||
):
|
||
config[key] = form.get(key, "").strip()
|
||
api_key = form.get("AI_API_KEY", "").strip()
|
||
if api_key:
|
||
config["AI_API_KEY"] = api_key
|
||
grok_api_key = form.get("GROK_API_KEY", "").strip()
|
||
if grok_api_key:
|
||
config["GROK_API_KEY"] = grok_api_key
|
||
if config["AI_USE_VISION"] and not config["AI_API_BASE"]:
|
||
raise ValueError("启用视觉模式时，视觉 API 地址不能为空")
|
||
if not config["AI_AGENT_NAME"] or not config["AI_HOSPITAL_NAME"]:
|
||
raise ValueError("客服名称和机构名称不能为空")
|
||
customer_effort = form.get(
|
||
"GROK_CUSTOMER_SERVICE_EFFORT", "low"
|
||
).strip().lower()
|
||
if customer_effort not in {"low", "medium", "high"}:
|
||
raise ValueError("Grok Build 客服推理强度必须是 low、medium 或 high")
|
||
config["GROK_CUSTOMER_SERVICE_EFFORT"] = customer_effort
|
||
backend = form.get("GROK_API_BACKEND", "chat_completions").strip()
|
||
if backend not in {"chat_completions", "responses", "messages", "dify"}:
|
||
raise ValueError("Grok Build 接口协议无效")
|
||
config["GROK_API_BACKEND"] = backend
|
||
auth_scheme = form.get("GROK_AUTH_SCHEME", "auto").strip()
|
||
if auth_scheme not in {"auto", "bearer", "x_api_key"}:
|
||
raise ValueError("Grok Build 认证方式无效")
|
||
config["GROK_AUTH_SCHEME"] = auth_scheme
|
||
raw_dify_inputs = form.get("GROK_DIFY_INPUTS", "{}").strip() or "{}"
|
||
if len(raw_dify_inputs.encode("utf-8")) > 64 * 1024:
|
||
raise ValueError("Dify inputs JSON 超过 64 KiB 安全限制")
|
||
try:
|
||
dify_inputs = json.loads(raw_dify_inputs)
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"Dify inputs JSON 格式错误：{exc.msg}") from exc
|
||
if not isinstance(dify_inputs, dict):
|
||
raise ValueError("Dify inputs 必须是 JSON 对象")
|
||
config["GROK_DIFY_INPUTS"] = dify_inputs
|
||
if config["GROK_CUSTOMER_SERVICE_ENABLED"] and not config["GROK_MODEL_ENABLED"]:
|
||
raise ValueError(
|
||
"启用 Grok Build 客服 Agent 时必须同时启用并配置自有模型；"
|
||
"系统不会回退到 Grok/xAI 模型"
|
||
)
|
||
if config["GROK_MODEL_ENABLED"]:
|
||
if not config["GROK_API_BASE"]:
|
||
raise ValueError("启用 Grok Agent 自有模型时，API 基址不能为空")
|
||
if backend == "dify" and not config["GROK_MODEL"]:
|
||
config["GROK_MODEL"] = "dify-app"
|
||
elif not config["GROK_MODEL"]:
|
||
raise ValueError("启用 Grok Agent 自有模型时，模型名称不能为空")
|
||
if not config.get("GROK_API_KEY"):
|
||
raise ValueError(
|
||
"启用 Grok Agent 自有模型时必须配置独立 API Key；"
|
||
"系统不会使用 xAI 登录凭据"
|
||
)
|
||
parsed = urllib.parse.urlsplit(config["GROK_API_BASE"])
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||
raise ValueError("Grok Build API 基址必须是有效的 http 或 https URL")
|
||
if parsed.username or parsed.password:
|
||
raise ValueError("Grok Build API 基址不能包含用户名或密码")
|
||
hostname = (parsed.hostname or "").lower()
|
||
if (
|
||
hostname == "x.ai"
|
||
or hostname.endswith(".x.ai")
|
||
or hostname == "grok.com"
|
||
or hostname.endswith(".grok.com")
|
||
):
|
||
raise ValueError(
|
||
"Grok Agent 必须使用后台自有模型，不能配置 xAI/Grok 模型端点"
|
||
)
|
||
if parsed.query or parsed.fragment:
|
||
raise ValueError("Grok Build API 基址暂不支持 query 或 fragment")
|
||
path = parsed.path.rstrip("/").lower()
|
||
if path.endswith("/chat-messages") and backend != "dify":
|
||
raise ValueError(
|
||
"检测到 Dify /chat-messages，请把接口协议选择为 "
|
||
"Dify Chat Messages（本地工具调用适配）"
|
||
)
|
||
if backend == "dify":
|
||
if auth_scheme == "x_api_key":
|
||
raise ValueError("Dify Chat Messages 必须使用 Bearer 认证")
|
||
if any(
|
||
path.endswith(suffix)
|
||
for suffix in ("/chat/completions", "/responses", "/messages")
|
||
):
|
||
raise ValueError(
|
||
"Dify API 地址应填写到 /v1 或完整 /v1/chat-messages，"
|
||
"不能填写 OpenAI/Anthropic 操作路径"
|
||
)
|
||
limits = {
|
||
"AI_CONTEXT_MAX_ROUNDS": (1, 50),
|
||
"AI_MAX_TOKENS": (50, 32000),
|
||
"AI_TIMEOUT": (5, 600),
|
||
"GROK_CUSTOMER_SERVICE_TIMEOUT": (30, 600),
|
||
"GROK_CUSTOMER_SERVICE_MAX_TURNS": (2, 30),
|
||
"AI_MCP_MAX_ROUNDS": (1, 20),
|
||
"GROK_CONTEXT_WINDOW": (4096, 2_000_000),
|
||
"GROK_MAX_TOKENS": (64, 262_144),
|
||
}
|
||
for key, (minimum, maximum) in limits.items():
|
||
try:
|
||
value = int(form.get(key, ""))
|
||
except ValueError as exc:
|
||
raise ValueError(f"{key} 必须是整数") from exc
|
||
if not minimum <= value <= maximum:
|
||
raise ValueError(f"{key} 必须在 {minimum}-{maximum} 之间")
|
||
config[key] = value
|
||
try:
|
||
temperature = float(form.get("AI_TEMPERATURE", ""))
|
||
except ValueError as exc:
|
||
raise ValueError("温度必须是数字") from exc
|
||
if not 0 <= temperature <= 2:
|
||
raise ValueError("温度必须在 0-2 之间")
|
||
config["AI_TEMPERATURE"] = temperature
|
||
try:
|
||
grok_temperature = float(form.get("GROK_TEMPERATURE", ""))
|
||
except ValueError as exc:
|
||
raise ValueError("Grok Agent 自有模型温度必须是数字") from exc
|
||
if not 0 <= grok_temperature <= 2:
|
||
raise ValueError("Grok Agent 自有模型温度必须在 0-2 之间")
|
||
config["GROK_TEMPERATURE"] = grok_temperature
|
||
try:
|
||
servers = json.loads(form.get("AI_MCP_SERVERS", "[]") or "[]")
|
||
except ValueError as exc:
|
||
raise ValueError("MCP 服务器不是有效 JSON") from exc
|
||
if not isinstance(servers, list):
|
||
raise ValueError("MCP 服务器 JSON 根节点必须是数组")
|
||
config["AI_MCP_SERVERS"] = servers
|
||
return config
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="企微客服助手配置管理后台")
|
||
parser.add_argument("--host", default=DEFAULT_HOST, help="监听地址，默认 127.0.0.1")
|
||
parser.add_argument(
|
||
"--port",
|
||
type=int,
|
||
default=DEFAULT_PORT,
|
||
help="首选监听端口，默认 8765；占用时自动递增",
|
||
)
|
||
parser.add_argument(
|
||
"--port-attempts",
|
||
type=int,
|
||
default=DEFAULT_PORT_ATTEMPTS,
|
||
help="端口被占用时最多尝试的端口数量，默认 100",
|
||
)
|
||
parser.add_argument("--db", type=Path, default=DEFAULT_DB, help="SQLite 数据库路径")
|
||
parser.add_argument(
|
||
"--runtime-file",
|
||
type=Path,
|
||
default=DEFAULT_RUNTIME_FILE,
|
||
help="实际端口发现文件路径",
|
||
)
|
||
parser.add_argument(
|
||
"--initial-admin-password",
|
||
default=os.environ.get("WECOM_ADMIN_INITIAL_PASSWORD", DEFAULT_ADMIN_PASSWORD),
|
||
help="首次创建数据库时的 admin 密码",
|
||
)
|
||
parser.add_argument(
|
||
"--reset-admin-password",
|
||
action="store_true",
|
||
help="交互式重置 admin 密码后退出",
|
||
)
|
||
return parser
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> None:
|
||
args = build_parser().parse_args(argv)
|
||
database = Database(args.db.resolve())
|
||
created = database.initialize(args.initial_admin_password)
|
||
if args.reset_admin_password:
|
||
first = getpass.getpass("新的 admin 密码：")
|
||
second = getpass.getpass("再次输入：")
|
||
if first != second:
|
||
raise SystemExit("两次输入不一致")
|
||
if not AdminHandler.valid_password(first):
|
||
raise SystemExit("密码至少 10 位，并同时包含字母和数字")
|
||
database.reset_admin_password(first)
|
||
print("admin 密码已重置，下次登录时必须再次修改。")
|
||
return
|
||
server, actual_port = create_server(
|
||
args.host,
|
||
args.port,
|
||
database,
|
||
max_attempts=args.port_attempts,
|
||
)
|
||
runtime_info = write_runtime_info(
|
||
args.runtime_file,
|
||
args.host,
|
||
actual_port,
|
||
local_sync_token=server.local_sync_token,
|
||
)
|
||
address = runtime_info["server_url"]
|
||
if actual_port != args.port:
|
||
print(f"端口 {args.port} 已被占用，已自动切换到 {actual_port}")
|
||
print(f"配置后台已启动：{address}")
|
||
if created:
|
||
print("首次登录账号：admin")
|
||
print(f"首次登录密码：{args.initial_admin_password}")
|
||
print("登录后必须立即修改初始密码。")
|
||
if args.host not in ("127.0.0.1", "localhost", "::1"):
|
||
print("警告：当前监听非本机地址；生产环境请通过 HTTPS 反向代理访问。")
|
||
try:
|
||
server.serve_forever(poll_interval=0.5)
|
||
except KeyboardInterrupt:
|
||
print("\n正在停止后台…")
|
||
finally:
|
||
server.server_close()
|
||
clear_runtime_info(args.runtime_file, actual_port)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|