Initial commit: DHCP Web Manager
Flask-based web UI for managing ISC DHCP server (dhcpd). Features: - subnet / static host binding CRUD via web - listen interface configuration (INTERFACESv4) - active lease query + pool utilization stats - service status, reload/restart, syntax check - user auth + operation logs - automatic backup before each config change - safe change pipeline: write file → dhcpd -t → systemctl restart + is-active verify - error messages include journalctl tail for diagnosis
This commit is contained in:
DHCP Service Bot
@@ -0,0 +1,602 @@
|
||||
"""
|
||||
DHCP Web 管理界面 - Flask 主程序
|
||||
- 用户认证（用户名/密码，session）
|
||||
- subnet/host 增删改，配置自动重载
|
||||
- 租约查询、地址池使用情况
|
||||
- 接口绑定配置
|
||||
- 操作日志
|
||||
|
||||
设计：用户的元数据（subnet/host/操作日志/接口）全在 SQLite，实际配置写在
|
||||
/etc/dhcp/dhcpd.conf。所有改动通过"Web -> 写文件 -> dhcpd -t -> systemctl reload" 流程。
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from functools import wraps
|
||||
|
||||
from flask import (Flask, render_template, request, redirect, url_for,
|
||||
session, jsonify, flash, abort)
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from sqlalchemy import event
|
||||
from werkzeug.security import generate_password_hash, check_password_hash
|
||||
|
||||
# 强制使用 pbkdf2:sha256，避免 scrypt 在某些环境下跨进程兼容性 bug
|
||||
_PASSWORD_METHOD = "pbkdf2:sha256"
|
||||
|
||||
import dhcp_ops
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Flask & DB 配置
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
INSTANCE_DIR = os.path.join(BASE_DIR, "instance")
|
||||
os.makedirs(INSTANCE_DIR, exist_ok=True)
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config["SECRET_KEY"] = os.environ.get("DHCPMGR_SECRET", "change-me-on-first-deploy")
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{os.path.join(INSTANCE_DIR, 'dhcp.db')}"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
app.config["JSON_AS_ASCII"] = False
|
||||
|
||||
db = SQLAlchemy(app)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 模型
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class User(db.Model):
|
||||
__tablename__ = "users"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
username = db.Column(db.String(80), unique=True, nullable=False)
|
||||
password_hash = db.Column(db.String(200), nullable=False)
|
||||
role = db.Column(db.String(20), default="admin") # 简单起见只有 admin / user
|
||||
created_at = db.Column(db.DateTime, default=lambda: datetime.utcnow())
|
||||
last_login = db.Column(db.DateTime)
|
||||
|
||||
def set_password(self, pw):
|
||||
self.password_hash = generate_password_hash(pw, method="pbkdf2:sha256")
|
||||
|
||||
def check_password(self, pw):
|
||||
return check_password_hash(self.password_hash, pw)
|
||||
|
||||
|
||||
class OperationLog(db.Model):
|
||||
__tablename__ = "op_logs"
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
timestamp = db.Column(db.DateTime, default=lambda: datetime.utcnow())
|
||||
user = db.Column(db.String(80))
|
||||
action = db.Column(db.String(50)) # create_subnet / update_subnet / delete_subnet / etc.
|
||||
target = db.Column(db.String(200)) # 子网网络、主机名等
|
||||
detail = db.Column(db.Text) # JSON / 文本详情
|
||||
result = db.Column(db.String(20)) # ok / error
|
||||
message = db.Column(db.Text)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 辅助函数
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def current_user():
|
||||
if "user_id" not in session:
|
||||
return None
|
||||
return db.session.get(User, session["user_id"])
|
||||
|
||||
|
||||
def login_required(f):
|
||||
@wraps(f)
|
||||
def wrapper(*a, **kw):
|
||||
if "user_id" not in session:
|
||||
return redirect(url_for("login", next=request.path))
|
||||
return f(*a, **kw)
|
||||
return wrapper
|
||||
|
||||
|
||||
def admin_required(f):
|
||||
@wraps(f)
|
||||
def wrapper(*a, **kw):
|
||||
u = current_user()
|
||||
if not u:
|
||||
return redirect(url_for("login", next=request.path))
|
||||
if u.role != "admin":
|
||||
abort(403)
|
||||
return f(*a, **kw)
|
||||
return wrapper
|
||||
|
||||
|
||||
def log_op(action, target, detail, result, message):
|
||||
u = current_user()
|
||||
entry = OperationLog(
|
||||
user=(u.username if u else "anonymous"),
|
||||
action=action, target=target,
|
||||
detail=detail if isinstance(detail, str) else json.dumps(detail, ensure_ascii=False),
|
||||
result=result, message=message,
|
||||
)
|
||||
db.session.add(entry)
|
||||
db.session.commit()
|
||||
return entry
|
||||
|
||||
|
||||
def get_subnets() -> list:
|
||||
block = dhcp_ops.read_managed_block()
|
||||
subs, _ = dhcp_ops.parse_managed_block(block)
|
||||
return subs
|
||||
|
||||
|
||||
def get_hosts() -> list:
|
||||
block = dhcp_ops.read_managed_block()
|
||||
_, hosts = dhcp_ops.parse_managed_block(block)
|
||||
return hosts
|
||||
|
||||
|
||||
def apply_and_reload(action: str, target: str, extra: dict = "",
|
||||
subnets_override=None, hosts_override=None):
|
||||
"""
|
||||
通用的"改 conf -> 语法检查 -> reload" 流程。
|
||||
subnets_override / hosts_override：调用方已修改内存中的对象，绕过从文件重读。
|
||||
返回 (success, message)
|
||||
"""
|
||||
subnets = subnets_override if subnets_override is not None else get_subnets()
|
||||
hosts = hosts_override if hosts_override is not None else get_hosts()
|
||||
ok, msg = dhcp_ops.write_managed_block(subnets, hosts)
|
||||
if not ok:
|
||||
log_op(action, target, extra, "error", msg)
|
||||
return False, msg
|
||||
syntax_ok, syntax_msg = dhcp_ops.check_syntax()
|
||||
if not syntax_ok:
|
||||
log_op(action, target, extra, "error", f"语法错误: {syntax_msg}")
|
||||
return False, f"语法检查失败:\n{syntax_msg}\n（配置未生效，已备份到 backups/）"
|
||||
r_ok, r_msg = dhcp_ops.reload_service()
|
||||
if not r_ok:
|
||||
log_op(action, target, extra, "error", f"reload 失败: {r_msg}")
|
||||
return False, f"reload 失败: {r_msg}"
|
||||
log_op(action, target, extra, "ok", f"{msg}; 语法 OK; {r_msg}")
|
||||
return True, f"{msg}，{r_msg}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 上下文：菜单 / 当前用户
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.context_processor
|
||||
def inject_globals():
|
||||
u = current_user()
|
||||
return dict(
|
||||
current_user=u,
|
||||
is_admin=(u.role == "admin" if u else False),
|
||||
now=datetime.utcnow(),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 路由
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username", "").strip()
|
||||
password = request.form.get("password", "")
|
||||
user = User.query.filter_by(username=username).first()
|
||||
if user and user.check_password(password):
|
||||
session["user_id"] = user.id
|
||||
user.last_login = datetime.utcnow()
|
||||
db.session.commit()
|
||||
return redirect(request.args.get("next") or url_for("dashboard"))
|
||||
flash("用户名或密码错误", "error")
|
||||
return render_template("login.html")
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
def logout():
|
||||
session.pop("user_id", None)
|
||||
return redirect(url_for("login"))
|
||||
|
||||
|
||||
@app.route("/")
|
||||
@login_required
|
||||
def dashboard():
|
||||
subnets = get_subnets()
|
||||
hosts = get_hosts()
|
||||
leases = dhcp_ops.active_leases()
|
||||
svc = dhcp_ops.get_service_status()
|
||||
|
||||
# 计算地址池使用率
|
||||
pool_stats = []
|
||||
lease_ips = {l.ip for l in leases}
|
||||
for s in subnets:
|
||||
try:
|
||||
from ipaddress import ip_network, ip_address
|
||||
net = ip_network(f"{s.network}/{s.netmask}", strict=False)
|
||||
start = ip_address(s.range_start)
|
||||
end = ip_address(s.range_end)
|
||||
pool_total = int(end) - int(start) + 1
|
||||
used = sum(1 for ip in lease_ips
|
||||
if start <= ip_address(ip) <= end and ip_address(ip) in net)
|
||||
# 减去 host 静态绑定数
|
||||
host_static_in_pool = sum(
|
||||
1 for h in hosts
|
||||
if h.ip and s.range_start <= h.ip <= s.range_end
|
||||
)
|
||||
pool_stats.append({
|
||||
"subnet": f"{s.network}/{s.netmask}",
|
||||
"range": f"{s.range_start} - {s.range_end}",
|
||||
"total": pool_total,
|
||||
"used": used,
|
||||
"static": host_static_in_pool,
|
||||
"available": pool_total - used - host_static_in_pool,
|
||||
"util_pct": round(used / pool_total * 100, 1) if pool_total > 0 else 0,
|
||||
})
|
||||
except Exception as e:
|
||||
pool_stats.append({"subnet": s.network, "error": str(e)})
|
||||
|
||||
return render_template("dashboard.html",
|
||||
subnets=subnets, hosts=hosts,
|
||||
leases=leases[:10],
|
||||
lease_total=len(leases),
|
||||
pool_stats=pool_stats,
|
||||
svc=svc)
|
||||
|
||||
|
||||
# ---------- subnet ----------
|
||||
|
||||
@app.route("/subnets")
|
||||
@login_required
|
||||
def subnet_list():
|
||||
return render_template("subnets.html",
|
||||
subnets=get_subnets(),
|
||||
hosts=get_hosts())
|
||||
|
||||
|
||||
@app.route("/subnets/new", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def subnet_new():
|
||||
if request.method == "POST":
|
||||
try:
|
||||
s = dhcp_ops.Subnet(
|
||||
network=request.form["network"].strip(),
|
||||
netmask=request.form["netmask"].strip(),
|
||||
routers=request.form["routers"].strip(),
|
||||
dns=request.form["dns"].strip(),
|
||||
range_start=request.form["range_start"].strip(),
|
||||
range_end=request.form["range_end"].strip(),
|
||||
default_lease=int(request.form.get("default_lease") or 600),
|
||||
max_lease=int(request.form.get("max_lease") or 7200),
|
||||
domain_name=request.form.get("domain_name", "").strip(),
|
||||
authoritative=bool(request.form.get("authoritative")),
|
||||
)
|
||||
except (KeyError, ValueError) as e:
|
||||
flash(f"表单错误: {e}", "error")
|
||||
return render_template("subnet_form.html", subnet={}, form=request.form)
|
||||
errs = s.validate()
|
||||
if errs:
|
||||
for e in errs:
|
||||
flash(e, "error")
|
||||
return render_template("subnet_form.html", subnet={}, form=request.form)
|
||||
# 写入：先获取现有，追加，再走 apply_and_reload
|
||||
subnets = get_subnets()
|
||||
if any(x.network == s.network and x.netmask == s.netmask for x in subnets):
|
||||
flash(f"已存在相同 subnet {s.network}/{s.netmask}", "error")
|
||||
return render_template("subnet_form.html", subnet={}, form=request.form)
|
||||
subnets.append(s)
|
||||
ok, msg = apply_and_reload("create_subnet", s.network, asdict_safe(s),
|
||||
subnets_override=subnets)
|
||||
if ok:
|
||||
flash(f"subnet {s.network}/{s.netmask} 已创建，{msg}", "ok")
|
||||
return redirect(url_for("subnet_list"))
|
||||
flash(msg, "error")
|
||||
return render_template("subnet_form.html", subnet={}, form=request.form)
|
||||
return render_template("subnet_form.html", subnet={}, form={})
|
||||
|
||||
|
||||
@app.route("/subnets/<network>/<netmask>/edit", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def subnet_edit(network, netmask):
|
||||
subnets = get_subnets()
|
||||
target = next((s for s in subnets if s.network == network and s.netmask == netmask), None)
|
||||
if not target:
|
||||
abort(404)
|
||||
if request.method == "POST":
|
||||
try:
|
||||
new_s = dhcp_ops.Subnet(
|
||||
network=request.form["network"].strip(),
|
||||
netmask=request.form["netmask"].strip(),
|
||||
routers=request.form["routers"].strip(),
|
||||
dns=request.form["dns"].strip(),
|
||||
range_start=request.form["range_start"].strip(),
|
||||
range_end=request.form["range_end"].strip(),
|
||||
default_lease=int(request.form.get("default_lease") or 600),
|
||||
max_lease=int(request.form.get("max_lease") or 7200),
|
||||
domain_name=request.form.get("domain_name", "").strip(),
|
||||
authoritative=bool(request.form.get("authoritative")),
|
||||
)
|
||||
except (KeyError, ValueError) as e:
|
||||
flash(f"表单错误: {e}", "error")
|
||||
return render_template("subnet_form.html", subnet=target, form=request.form)
|
||||
errs = new_s.validate()
|
||||
if errs:
|
||||
for e in errs:
|
||||
flash(e, "error")
|
||||
return render_template("subnet_form.html", subnet=new_s, form=request.form)
|
||||
subnets = [new_s if (s.network == network and s.netmask == netmask) else s
|
||||
for s in subnets]
|
||||
ok, msg = apply_and_reload(
|
||||
"update_subnet", new_s.network,
|
||||
{"old": f"{network}/{netmask}", "new": asdict_safe(new_s)},
|
||||
subnets_override=subnets,
|
||||
)
|
||||
if ok:
|
||||
flash(f"subnet 已更新，{msg}", "ok")
|
||||
return redirect(url_for("subnet_list"))
|
||||
flash(msg, "error")
|
||||
return render_template("subnet_form.html", subnet=new_s, form=request.form)
|
||||
return render_template("subnet_form.html", subnet=target, form=asdict_safe(target))
|
||||
|
||||
|
||||
@app.route("/subnets/<network>/<netmask>/delete", methods=["POST"])
|
||||
@admin_required
|
||||
def subnet_delete(network, netmask):
|
||||
subnets = [s for s in get_subnets()
|
||||
if not (s.network == network and s.netmask == netmask)]
|
||||
ok, msg = apply_and_reload(
|
||||
"delete_subnet", f"{network}/{netmask}",
|
||||
{"remaining_subnets": len(subnets)},
|
||||
subnets_override=subnets,
|
||||
)
|
||||
flash(f"已删除 subnet {network}/{netmask}，{msg}", "ok" if ok else "error")
|
||||
return redirect(url_for("subnet_list"))
|
||||
|
||||
|
||||
# ---------- host ----------
|
||||
|
||||
@app.route("/hosts")
|
||||
@login_required
|
||||
def host_list():
|
||||
return render_template("hosts.html", hosts=get_hosts(), subnets=get_subnets())
|
||||
|
||||
|
||||
@app.route("/hosts/new", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def host_new():
|
||||
if request.method == "POST":
|
||||
try:
|
||||
h = dhcp_ops.Host(
|
||||
name=request.form["name"].strip(),
|
||||
mac=request.form["mac"].strip().lower(),
|
||||
ip=request.form["ip"].strip(),
|
||||
subnet_network=request.form.get("subnet_network", "").strip(),
|
||||
)
|
||||
except (KeyError, ValueError) as e:
|
||||
flash(f"表单错误: {e}", "error")
|
||||
return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets())
|
||||
errs = h.validate()
|
||||
if errs:
|
||||
for e in errs:
|
||||
flash(e, "error")
|
||||
return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets())
|
||||
hosts = get_hosts()
|
||||
if any(x.name == h.name for x in hosts):
|
||||
flash(f"已存在同名 host: {h.name}", "error")
|
||||
return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets())
|
||||
if any(x.ip == h.ip for x in hosts):
|
||||
flash(f"IP {h.ip} 已被其他 host 占用", "error")
|
||||
return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets())
|
||||
hosts.append(h)
|
||||
ok, msg = apply_and_reload("create_host", h.name, asdict_safe(h),
|
||||
hosts_override=hosts)
|
||||
if ok:
|
||||
flash(f"host {h.name} 已创建，{msg}", "ok")
|
||||
return redirect(url_for("host_list"))
|
||||
flash(msg, "error")
|
||||
return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets())
|
||||
return render_template("host_form.html", host={}, form={}, subnets=get_subnets())
|
||||
|
||||
|
||||
@app.route("/hosts/<name>/edit", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def host_edit(name):
|
||||
hosts = get_hosts()
|
||||
target = next((h for h in hosts if h.name == name), None)
|
||||
if not target:
|
||||
abort(404)
|
||||
if request.method == "POST":
|
||||
try:
|
||||
new_h = dhcp_ops.Host(
|
||||
name=request.form["name"].strip(),
|
||||
mac=request.form["mac"].strip().lower(),
|
||||
ip=request.form["ip"].strip(),
|
||||
subnet_network=request.form.get("subnet_network", "").strip(),
|
||||
)
|
||||
except (KeyError, ValueError) as e:
|
||||
flash(f"表单错误: {e}", "error")
|
||||
return render_template("host_form.html", host=target, form=request.form, subnets=get_subnets())
|
||||
errs = new_h.validate()
|
||||
if errs:
|
||||
for e in errs:
|
||||
flash(e, "error")
|
||||
return render_template("host_form.html", host=new_h, form=request.form, subnets=get_subnets())
|
||||
# 名字变更则检查冲突
|
||||
if new_h.name != name and any(h.name == new_h.name for h in hosts):
|
||||
flash(f"已存在同名 host: {new_h.name}", "error")
|
||||
return render_template("host_form.html", host=new_h, form=request.form, subnets=get_subnets())
|
||||
hosts = [new_h if h.name == name else h for h in hosts]
|
||||
ok, msg = apply_and_reload("update_host", new_h.name,
|
||||
{"old": name, "new": asdict_safe(new_h)},
|
||||
hosts_override=hosts)
|
||||
if ok:
|
||||
flash(f"host 已更新，{msg}", "ok")
|
||||
return redirect(url_for("host_list"))
|
||||
flash(msg, "error")
|
||||
return render_template("host_form.html", host=new_h, form=request.form, subnets=get_subnets())
|
||||
return render_template("host_form.html", host=target, form=asdict_safe(target), subnets=get_subnets())
|
||||
|
||||
|
||||
@app.route("/hosts/<name>/delete", methods=["POST"])
|
||||
@admin_required
|
||||
def host_delete(name):
|
||||
hosts = [h for h in get_hosts() if h.name != name]
|
||||
ok, msg = apply_and_reload("delete_host", name, {"remaining_hosts": len(hosts)},
|
||||
hosts_override=hosts)
|
||||
flash(f"已删除 host {name}，{msg}", "ok" if ok else "error")
|
||||
return redirect(url_for("host_list"))
|
||||
|
||||
|
||||
# ---------- leases ----------
|
||||
|
||||
@app.route("/leases")
|
||||
@login_required
|
||||
def lease_list():
|
||||
leases = dhcp_ops.parse_leases()
|
||||
q = request.args.get("q", "").strip().lower()
|
||||
if q:
|
||||
leases = [l for l in leases if q in l.ip.lower()
|
||||
or q in l.mac.lower()
|
||||
or q in l.client_hostname.lower()]
|
||||
leases.sort(key=lambda l: l.remaining_seconds() or -1, reverse=True)
|
||||
return render_template("leases.html", leases=leases, q=q)
|
||||
|
||||
|
||||
# ---------- config (interfaces + 原始 conf) ----------
|
||||
|
||||
@app.route("/config", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def config_view():
|
||||
if request.method == "POST":
|
||||
action = request.form.get("action")
|
||||
if action == "save_interfaces":
|
||||
ifaces = request.form.get("interfaces", "").strip().split()
|
||||
ok, msg = dhcp_ops.save_interfaces(ifaces)
|
||||
log_op("save_interfaces", ",".join(ifaces), "", "ok" if ok else "error", msg)
|
||||
flash(msg, "ok" if ok else "error")
|
||||
if ok:
|
||||
# 重新 reload 服务（需要重启以应用新 interface）
|
||||
r_ok, r_msg = dhcp_ops.restart_service()
|
||||
flash(r_msg, "ok" if r_ok else "error")
|
||||
log_op("restart_service", "", "", "ok" if r_ok else "error", r_msg)
|
||||
elif action == "reload":
|
||||
r_ok, r_msg = dhcp_ops.reload_service()
|
||||
flash(r_msg, "ok" if r_ok else "error")
|
||||
log_op("reload_service", "", "", "ok" if r_ok else "error", r_msg)
|
||||
elif action == "restart":
|
||||
r_ok, r_msg = dhcp_ops.restart_service()
|
||||
flash(r_msg, "ok" if r_ok else "error")
|
||||
log_op("restart_service", "", "", "ok" if r_ok else "error", r_msg)
|
||||
return redirect(url_for("config_view"))
|
||||
|
||||
ifaces = dhcp_ops.get_interfaces()
|
||||
conf_content = ""
|
||||
if os.path.exists(dhcp_ops.DHCPD_CONF):
|
||||
with open(dhcp_ops.DHCPD_CONF) as f:
|
||||
conf_content = f.read()
|
||||
svc = dhcp_ops.get_service_status()
|
||||
syntax_ok, syntax_msg = dhcp_ops.check_syntax()
|
||||
return render_template("config.html",
|
||||
ifaces=ifaces, conf=conf_content,
|
||||
svc=svc, syntax_ok=syntax_ok, syntax_msg=syntax_msg)
|
||||
|
||||
|
||||
# ---------- 操作日志 ----------
|
||||
|
||||
@app.route("/logs")
|
||||
@admin_required
|
||||
def log_list():
|
||||
page = request.args.get("page", 1, type=int)
|
||||
per = 50
|
||||
q = OperationLog.query.order_by(OperationLog.id.desc())
|
||||
total = q.count()
|
||||
logs = q.offset((page - 1) * per).limit(per).all()
|
||||
pages = (total + per - 1) // per
|
||||
return render_template("logs.html", logs=logs, page=page, pages=pages, total=total)
|
||||
|
||||
|
||||
# ---------- 用户管理（修改密码） ----------
|
||||
|
||||
@app.route("/profile", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def profile():
|
||||
u = current_user()
|
||||
if request.method == "POST":
|
||||
old = request.form.get("old_password", "")
|
||||
new = request.form.get("new_password", "")
|
||||
if not u.check_password(old):
|
||||
flash("当前密码错误", "error")
|
||||
elif len(new) < 6:
|
||||
flash("新密码至少 6 位", "error")
|
||||
else:
|
||||
u.set_password(new)
|
||||
db.session.commit()
|
||||
flash("密码已更新", "ok")
|
||||
return redirect(url_for("profile"))
|
||||
return render_template("profile.html", user=u)
|
||||
|
||||
|
||||
# ---------- API（轻量查询） ----------
|
||||
|
||||
@app.route("/api/syntax_check")
|
||||
@admin_required
|
||||
def api_syntax():
|
||||
ok, msg = dhcp_ops.check_syntax()
|
||||
return jsonify(ok=ok, msg=msg)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 辅助：把 dataclass 转 dict 而非崩溃
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def asdict_safe(obj):
|
||||
try:
|
||||
from dataclasses import asdict
|
||||
return asdict(obj)
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _format_duration(seconds):
|
||||
"""把秒数格式化为 d/h/m/s 友好显示"""
|
||||
if seconds is None:
|
||||
return "—"
|
||||
seconds = int(seconds)
|
||||
if seconds < 0:
|
||||
return "已过期"
|
||||
if seconds < 60:
|
||||
return f"{seconds}s"
|
||||
if seconds < 3600:
|
||||
return f"{seconds // 60}m {seconds % 60}s"
|
||||
if seconds < 86400:
|
||||
h, rem = divmod(seconds, 3600)
|
||||
return f"{h}h {rem // 60}m"
|
||||
d, rem = divmod(seconds, 86400)
|
||||
h = rem // 3600
|
||||
return f"{d}d {h}h"
|
||||
|
||||
|
||||
app.jinja_env.filters["format_duration"] = _format_duration
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 初始化
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def seed_admin():
|
||||
"""首次启动时创建默认管理员 admin/admin"""
|
||||
if User.query.count() == 0:
|
||||
admin = User(username="admin", role="admin")
|
||||
admin.set_password("admin")
|
||||
db.session.add(admin)
|
||||
db.session.commit()
|
||||
print("[init] created default user: admin / admin（请尽快修改）")
|
||||
|
||||
|
||||
def init_app():
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
seed_admin()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
init_app()
|
||||
port = int(os.environ.get("PORT", "5100"))
|
||||
app.run(host="0.0.0.0", port=port, debug=False)
|
||||
Reference in New Issue
Block a user