import time import json import asyncio import logging from aiohttp import web from ..utils.system_info import SystemInfoCollector from ..utils.auth import panel_auth collector = SystemInfoCollector() logger = logging.getLogger(__name__) # Track active system-status WS clients _sys_ws_clients: set = set() def _ws_auth_wrapper(handler): """WebSocket 鉴权 — Cookie 优先(同源自动发送), query token 备用""" async def wrapper(request): # 1. Cookie (浏览器同源自动发送, 支持 HttpOnly) token = request.cookies.get("panel_token", "") # 2. Query string token 备用 (跨域/非浏览器客户端) if not token: token = request.query.get("token", "") # 3. API Key 验证 session_store = request.app.get("panel_session_store", {}) is_valid = token and token in session_store if not is_valid and token: from services.web_panel.routes.apikeys import validate_api_key is_valid = validate_api_key(token) is not None if not is_valid: ws = web.WebSocketResponse() await ws.prepare(request) await ws.send_str(json.dumps({"error": "Unauthorized"})) await ws.close(code=4001, message="Unauthorized") return ws return await handler(request) return wrapper def setup_routes(app, prefix=''): app.router.add_get(f'{prefix}/api/framework', panel_auth(get_framework)) app.router.add_get(f'{prefix}/api/system', panel_auth(get_system)) app.router.add_get(f'{prefix}/api/system/ws', _ws_auth_wrapper(sys_ws_handler)) 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_post(f'{prefix}/api/updates/backup', panel_auth(create_backup)) 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(): """Read version from config file (shared helper)""" ver = "v0.6.0" try: import yaml, os config_path = os.path.join(os.path.dirname(__file__), "..", "..", "..", "config", "framework", "base_config.yaml") with open(config_path) as f: cfg = yaml.safe_load(f) ver = cfg.get("framework", {}).get("version", ver) except: pass return ver def _get_framework_data(sm): """Collect framework status (reused by HTTP and WS handlers)""" ver = _read_version() ps = sm.get_service("plugin") if sm else None uptime = int(time.time() - getattr(sm, 'start_time', time.time())) return { "version": ver, "uptime": uptime, "plugins": len(ps.plugins) if ps else 0 } async def get_framework(req): sm = req.app.get('service_manager') if not sm: return web.json_response({"error": "Missing"}, status=500) return web.json_response(_get_framework_data(sm)) def _get_collector(req): """获取共享的 SystemInfoCollector（优先从 service_manager），fallback 到模块级实例""" sm = req.app.get('service_manager') if sm: try: shared = sm.get_service("sys_collector") if shared: return shared except Exception: pass return collector async def get_system(req): c = _get_collector(req) return web.json_response(c.get_all()) async def sys_ws_handler(req): """WebSocket push endpoint: system+framwork stats every 2s (no logging per message)""" ws = web.WebSocketResponse(heartbeat=30.0) await ws.prepare(req) _sys_ws_clients.add(ws) async def push(): """Send one snapshot to this client (silent on error)""" try: sm = req.app.get('service_manager') c = _get_collector(req) payload = json.dumps({ "type": "sys", "system": c.get_all(), "framework": _get_framework_data(sm) }, ensure_ascii=False) if not ws.closed: await ws.send_str(payload) except Exception: pass # Background push loop — runs until client disconnects async def push_loop(): while not ws.closed: await push() await asyncio.sleep(2) task = asyncio.ensure_future(push_loop()) try: async for msg in ws: if msg.type == web.WSMsgType.ERROR: break finally: task.cancel() try: await task except asyncio.CancelledError: pass _sys_ws_clients.discard(ws) return ws async def get_updates(req): """获取更新状态""" sm = req.app.get("service_manager") us = sm.get_service("update") if sm else None if not us: return web.json_response({"error": "更新服务未就绪"}, status=503) return web.json_response(us.status) async def check_now(req): """手动触发更新检查""" sm = req.app.get("service_manager") us = sm.get_service("update") if sm else None if not us: return web.json_response({"error": "更新服务未就绪"}, status=503) await us.check_now() return web.json_response(us.status) async def apply_update(req): """执行更新""" sm = req.app.get("service_manager") us = sm.get_service("update") if sm else None if not us: 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 create_backup(req): """手动创建备份""" import time as _time, os as _os from services.update_service import _backup_current, _PROJECT_ROOT backup_dir = _os.path.join(_PROJECT_ROOT, "data", "backups") _os.makedirs(backup_dir, exist_ok=True) try: sm = req.app.get("service_manager") ver = "unknown" if sm: try: init = sm.get_service("init") ver = init.get_config("base").get("framework", {}).get("version", "0.0.0") except: pass name = f"SenSu_backup_{ver}_{_time.strftime('%Y%m%d_%H%M%S')}.zip" dest = _os.path.join(backup_dir, name) loop = asyncio.get_event_loop() await loop.run_in_executor(None, _backup_current, dest) # 清理旧备份 backups = sorted(glob.glob(_os.path.join(backup_dir, "SenSu_backup_*.zip")), key=_os.path.getmtime, reverse=True) for old_b in backups[5:]: _os.unlink(old_b) return web.json_response({"ok": True, "name": name}) except Exception as e: return web.json_response({"ok": False, "error": str(e)}, status=500) 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)