feat: WebUI 版本检测 — 顶栏按钮 + 设置页备份/回滚/更新管理
顶栏: - 版本号按钮, 有更新时变绿色并显示新版本号 - 每10分钟自动检查 设置页新增「版本与更新」区域: - 当前版本 + 更新状态徽章 - 检查更新 / 立即更新 按钮 - 备份列表 + 手动回滚 API: - GET /api/updates/backups — 备份列表 - POST /api/updates/rollback — 执行回滚 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -43,6 +43,8 @@ def setup_routes(app, prefix=''):
|
||||
app.router.add_get(f'{prefix}/api/updates', panel_auth(get_updates))
|
||||
app.router.add_post(f'{prefix}/api/updates/check', panel_auth(check_now))
|
||||
app.router.add_post(f'{prefix}/api/updates/apply', panel_auth(apply_update))
|
||||
app.router.add_get(f'{prefix}/api/updates/backups', panel_auth(list_backups))
|
||||
app.router.add_post(f'{prefix}/api/updates/rollback', panel_auth(do_rollback))
|
||||
logger.info(f"📡 系统状态WS端点已注册: {prefix}/api/system/ws (已加认证)")
|
||||
|
||||
def _read_version():
|
||||
@@ -160,3 +162,58 @@ async def apply_update(req):
|
||||
return web.json_response({"error": "更新服务未就绪"}, status=503)
|
||||
result = await us.apply_update()
|
||||
return web.json_response(result)
|
||||
|
||||
|
||||
async def list_backups(req):
|
||||
"""列出可用备份"""
|
||||
import glob, os, time as _time
|
||||
backup_dir = os.path.join(os.path.dirname(__file__), "..", "..", "..", "data", "backups")
|
||||
backups = []
|
||||
if os.path.isdir(backup_dir):
|
||||
for f in sorted(glob.glob(os.path.join(backup_dir, "SenSu_backup_*.zip")), key=os.path.getmtime, reverse=True):
|
||||
st = os.stat(f)
|
||||
backups.append({
|
||||
"name": os.path.basename(f),
|
||||
"size_mb": round(st.st_size / 1048576, 1),
|
||||
"time": _time.strftime("%Y-%m-%d %H:%M", _time.localtime(st.st_mtime)),
|
||||
})
|
||||
return web.json_response({"backups": backups})
|
||||
|
||||
|
||||
async def do_rollback(req):
|
||||
"""执行回滚"""
|
||||
import glob, os, zipfile, shutil, tempfile
|
||||
try:
|
||||
data = await req.json()
|
||||
backup_name = data.get("name", "")
|
||||
except Exception:
|
||||
backup_name = ""
|
||||
backup_dir = os.path.join(os.path.dirname(__file__), "..", "..", "..", "data", "backups")
|
||||
if backup_name:
|
||||
backup_path = os.path.join(backup_dir, backup_name)
|
||||
if not os.path.exists(backup_path):
|
||||
return web.json_response({"ok": False, "error": "备份文件不存在"}, status=404)
|
||||
else:
|
||||
# 使用最新备份
|
||||
backups = sorted(glob.glob(os.path.join(backup_dir, "SenSu_backup_*.zip")), key=os.path.getmtime, reverse=True)
|
||||
if not backups:
|
||||
return web.json_response({"ok": False, "error": "没有可用的备份"}, status=404)
|
||||
backup_path = backups[0]
|
||||
try:
|
||||
project_root = os.path.join(os.path.dirname(__file__), "..", "..", "..")
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with zipfile.ZipFile(backup_path, "r") as zf:
|
||||
zf.extractall(tmp)
|
||||
for item in os.listdir(tmp):
|
||||
src = os.path.join(tmp, item)
|
||||
dst = os.path.join(project_root, item)
|
||||
if os.path.isdir(src):
|
||||
shutil.copytree(src, dst, dirs_exist_ok=True)
|
||||
else:
|
||||
shutil.copy2(src, dst)
|
||||
# 写重启标记
|
||||
restart_flag = os.path.join(project_root, ".restart_flag")
|
||||
Path(restart_flag).touch()
|
||||
return web.json_response({"ok": True, "msg": "回滚完成, 框架将重启"})
|
||||
except Exception as e:
|
||||
return web.json_response({"ok": False, "error": str(e)}, status=500)
|
||||
|
||||
Reference in New Issue
Block a user