refator_comm1

app.py

@@ -10,12 +10,16 @@ import subprocess
 from typing import List, Dict, Any, Optional, Tuple
 from flask import Flask, request, jsonify, render_template
 
+# NEW: WebSockets
+from flask_sock import Sock
+
 APP_TITLE = "PVE HA Panel"
 DEFAULT_NODE = socket.gethostname()
 HA_UNITS_START = ["watchdog-mux", "pve-ha-crm", "pve-ha-lrm"]
 HA_UNITS_STOP = list(reversed(HA_UNITS_START))
 
 app = Flask(__name__, template_folder="templates", static_folder="static")
+sock = Sock(app)  # NEW
 
 # ---------------- exec helpers ----------------
 def run(cmd: List[str], timeout: int = 25) -> subprocess.CompletedProcess:
@@ -41,12 +45,16 @@ def post_json(cmd: List[str], timeout: Optional[int] = None) -> Any:
     if cmd and cmd[0] == "pvesh" and len(cmd) > 2 and cmd[1] != "create":
         cmd = ["pvesh", "create"] + cmd[1:]
     r = run(cmd, timeout=timeout or 25)
-    if r.returncode != 0 or not r.stdout.strip():
-        return None
+    if r.returncode != 0:
+        # pvesh create zwykle zwraca JSON tylko przy 200
+        try:
+            return json.loads(r.stdout) if r.stdout.strip() else {"error": r.stderr.strip(), "rc": r.returncode}
+        except Exception:
+            return {"error": r.stderr.strip(), "rc": r.returncode}
     try:
-        return json.loads(r.stdout)
+        return json.loads(r.stdout) if r.stdout.strip() else {}
     except Exception:
-        return None
+        return {}
 
 def is_active(unit: str) -> bool:
     return run(["systemctl", "is-active", "--quiet", unit]).returncode == 0
@@ -106,8 +114,7 @@ def enrich_nodes(nodes_list: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     for n in nodes_list or []:
         name = n.get("node")
         if not name:
-            out.append(n)
-            continue
+            out.append(n); continue
         detail = get_json(["pvesh", "get", f"/nodes/{name}/status"]) or {}
         if "loadavg" in detail: n["loadavg"] = detail["loadavg"]
         if "cpu" in detail: n["cpu"] = detail["cpu"]
@@ -374,9 +381,7 @@ def api_replication_all():
     jobs: List[Dict[str, Any]] = []
     nodes = [n.get("node") for n in (api_cluster_data().get("nodes") or []) if n.get("node")]
     for name in nodes:
-        # PVE 7/8: GET /nodes/{node}/replication lub /nodes/{node}/replication/jobs
         data = get_json(["pvesh", "get", f"/nodes/{name}/replication"]) or get_json(["pvesh","get",f"/nodes/{name}/replication/jobs"]) or []
-        # fallback: pvesr status na zdalnym nodzie bywa utrudnione, więc parsujemy standardową strukturę API
         for it in (data or []):
             jobs.append({
                 "node": name,
@@ -391,8 +396,7 @@ def api_replication_all():
             })
     return jsonify({ "jobs": jobs })
 
-# --- istniejące endpointy detali i list ---
-
+# --- Detale ---
 @app.get("/api/vm")
 def api_vm_detail():
     sid = request.args.get("sid", "")
@@ -410,6 +414,7 @@ def api_list_vmct():
     nonha = [v for k, v in meta.items() if k not in ha_sids]
     return jsonify({"nonha": nonha, "ha_index": list(ha_sids), "count_nonha": len(nonha), "count_all_vmct": len(meta)})
 
+# --- Enable/Disable maintenance ---
 @app.post("/api/enable")
 def api_enable():
     if os.geteuid() != 0:
@@ -432,7 +437,7 @@ def api_disable():
     ha_node_maint(False, node, log)
     return jsonify(ok=True, log=log)
 
-# --- VM/CT admin actions API (pozostaje jak było; bez usuwania) ---
+# --- VM/CT admin actions API (zwrot UPID dla live) ---
 
 def vm_locked(typ: str, node: str, vmid: int) -> bool:
     base = f"/nodes/{node}/{typ}/{vmid}"
@@ -466,32 +471,40 @@ def api_vm_action():
     if not node:
         return jsonify(ok=False, error="unknown node"), 400
 
-    if action == "migrate_offline":
-        action = "migrate"
-
     try:
         if action == "unlock":
+            # brak UPID (działa natychmiast)
             if typ != "qemu":
                 return jsonify(ok=False, error="unlock only for qemu"), 400
             if not vm_locked(typ, node, vmid):
                 return jsonify(ok=True, msg="not locked")
             r = run(["qm", "unlock", str(vmid)])
-            return jsonify(ok=(r.returncode == 0), stdout=r.stdout, stderr=r.stderr)
+            return jsonify(ok=(r.returncode == 0), stdout=r.stdout, stderr=r.stderr, upid=None, source_node=node)
 
         elif action in ("start", "stop", "shutdown"):
-            r = run_qm_pct(typ, vmid, action)
-            return jsonify(ok=(r.returncode == 0), stdout=r.stdout, stderr=r.stderr)
-
-        elif action == "migrate":
-            if not target or target == node:
-                return jsonify(ok=False, error="target required and must differ from source"), 400
-            base = f"/nodes/{node}/{typ}/{vmid}/migrate"
-            cmd = ["pvesh", "create", base, "-target", target, "-online", "0"]
-            res = post_json(cmd, timeout=120)
+            # PVE API -> zwraca UPID
+            base = f"/nodes/{node}/{typ}/{vmid}/status/{action}"
+            res = post_json(["pvesh", "create", base], timeout=120) or {}
             upid = None
             if isinstance(res, dict):
                 upid = res.get("data") or res.get("upid")
-            return jsonify(ok=True, result=res or {}, upid=upid, source_node=node)
+            if not upid:
+                # fallback do qm/pct (bez UPID), ale zwrócimy ok
+                r = run_qm_pct(typ, vmid, action)
+                return jsonify(ok=(r.returncode == 0), stdout=r.stdout, stderr=r.stderr, upid=None, source_node=node)
+            return jsonify(ok=True, result=res, upid=upid, source_node=node)
+
+        elif action == "migrate" or action == "migrate_offline":
+            if action == "migrate_offline":  # zgodność wstecz
+                action = "migrate"
+            if not target or target == node:
+                return jsonify(ok=False, error="target required and must differ from source"), 400
+            base = f"/nodes/{node}/{typ}/{vmid}/migrate"
+            res = post_json(["pvesh", "create", base, "-target", target, "-online", "0"], timeout=120) or {}
+            upid = None
+            if isinstance(res, dict):
+                upid = res.get("data") or res.get("upid")
+            return jsonify(ok=True, result=res, upid=upid, source_node=node)
 
         else:
             return jsonify(ok=False, error="unknown action"), 400
@@ -528,6 +541,146 @@ def api_task_log():
             next_start = start_i
     return jsonify(ok=True, lines=lines or [], next_start=next_start)
 
+# ---------------- WebSocket: live tail zadań ----------------
+@sock.route("/ws/task")
+def ws_task(ws):
+    # query: upid, node
+    q = ws.environ.get("QUERY_STRING", "")
+    params = {}
+    for part in q.split("&"):
+        if not part: continue
+        k, _, v = part.partition("=")
+        params[k] = v
+    upid = params.get("upid", "").strip()
+    node = params.get("node", "").strip()
+    if not upid or not node:
+        ws.send(json.dumps({"type":"error","error":"upid and node are required"}))
+        return
+    start = 0
+    try:
+        while True:
+            st = get_json(["pvesh", "get", f"/nodes/{node}/tasks/{upid}/status"]) or {}
+            ws.send(json.dumps({"type":"status","status":st}))
+            lines = get_json(["pvesh", "get", f"/nodes/{node}/tasks/{upid}/log", "-start", str(start)]) or []
+            if isinstance(lines, list) and lines:
+                for ln in lines:
+                    txt = (ln.get("t") if isinstance(ln, dict) else None)
+                    if txt:
+                        ws.send(json.dumps({"type":"log","line":txt}))
+                try:
+                    start = max((int(x.get("n", start)) for x in lines if isinstance(x, dict)), default=start) + 1
+                except Exception:
+                    pass
+            # koniec
+            if isinstance(st, dict) and (str(st.get("status","")).lower() == "stopped" or st.get("exitstatus")):
+                ok = (str(st.get("exitstatus","")).upper() == "OK")
+                ws.send(json.dumps({"type":"done","ok":ok,"exitstatus":st.get("exitstatus")}))
+                break
+            time.sleep(1.2)
+    except Exception:
+        try:
+            ws.close()
+        except Exception:
+            pass
+
+# ---------------- WebSocket: live tail zadań ----------------
+@sock.route("/ws/task")
+def ws_task(ws):
+    # (bez zmian – jak w mojej poprzedniej wersji)
+    q = ws.environ.get("QUERY_STRING", "")
+    params = {}
+    for part in q.split("&"):
+        if not part: continue
+        k, _, v = part.partition("=")
+        params[k] = v
+    upid = params.get("upid", "").strip()
+    node = params.get("node", "").strip()
+    if not upid or not node:
+        ws.send(json.dumps({"type":"error","error":"upid and node are required"}))
+        return
+    start = 0
+    try:
+        while True:
+            st = get_json(["pvesh", "get", f"/nodes/{node}/tasks/{upid}/status"]) or {}
+            ws.send(json.dumps({"type":"status","status":st}))
+            lines = get_json(["pvesh", "get", f"/nodes/{node}/tasks/{upid}/log", "-start", str(start)]) or []
+            if isinstance(lines, list) and lines:
+                for ln in lines:
+                    txt = (ln.get("t") if isinstance(ln, dict) else None)
+                    if txt:
+                        ws.send(json.dumps({"type":"log","line":txt}))
+                try:
+                    start = max((int(x.get("n", start)) for x in lines if isinstance(x, dict)), default=start) + 1
+                except Exception:
+                    pass
+            if isinstance(st, dict) and (str(st.get("status","")).lower() == "stopped" or st.get("exitstatus")):
+                ok = (str(st.get("exitstatus","")).upper() == "OK")
+                ws.send(json.dumps({"type":"done","ok":ok,"exitstatus":st.get("exitstatus")}))
+                break
+            time.sleep(1.2)
+    except Exception:
+        try: ws.close()
+        except Exception: pass
+
+# ---------------- WebSocket: broadcast observe per sid ----------------
+@sock.route("/ws/observe")
+def ws_observe(ws):
+    """
+    Streamuje zmiany stanu VM/CT i aktywne taski dla danego SID.
+    Query: sid=vm:101 (albo ct:123)
+    """
+    q = ws.environ.get("QUERY_STRING", "")
+    params = {}
+    for part in q.split("&"):
+        if not part: continue
+        k, _, v = part.partition("="); params[k] = v
+    sid = (params.get("sid") or "").strip()
+    if not sid:
+        ws.send(json.dumps({"type":"error","error":"sid required"})); return
+
+    meta = cluster_vmct_meta()
+    tup = sid_to_tuple(sid, meta)
+    if not tup:
+        ws.send(json.dumps({"type":"error","error":"unknown sid"})); return
+    typ, vmid, node = tup
+    if not node:
+        ws.send(json.dumps({"type":"error","error":"could not resolve node"})); return
+
+    last_hash = None
+    seen_upids = set()
+    try:
+        while True:
+            base = f"/nodes/{node}/{typ}/{vmid}"
+            cur = get_json(["pvesh", "get", f"{base}/status/current"]) or {}
+            # wyślij tylko gdy zmiana
+            cur_hash = json.dumps(cur, sort_keys=True)
+            if cur_hash != last_hash:
+                last_hash = cur_hash
+                ws.send(json.dumps({"type":"vm","current":cur,"meta":{"sid":norm_sid(sid),"node":node,"typ":typ,"vmid":vmid}}))
+
+            tasks = get_json(["pvesh","get",f"/nodes/{node}/tasks","-limit","50"]) or []
+            for t in tasks:
+                upid = t.get("upid") or t.get("pstart")  # upid musi być stringiem
+                tid = (t.get("id") or "")  # np. "qemu/101" lub "lxc/123"
+                if not isinstance(upid, str): continue
+                if (str(vmid) in tid) or (f"{'qemu' if typ=='qemu' else 'lxc'}/{vmid}" in tid):
+                    # status szczegółowy
+                    st = get_json(["pvesh","get",f"/nodes/{node}/tasks/{upid}/status"]) or {}
+                    ev = {"type":"task","upid":upid,"status":st.get("status"),"exitstatus":st.get("exitstatus"),"node":node}
+                    ws.send(json.dumps(ev))
+                    # nowy "running" task → ogłoś raz
+                    if upid not in seen_upids and (str(st.get("status","")).lower() != "stopped"):
+                        seen_upids.add(upid)
+                        ws.send(json.dumps({"type":"task-start","upid":upid,"node":node}))
+                    # zakończenie
+                    if str(st.get("status","")).lower() == "stopped" or st.get("exitstatus"):
+                        ws.send(json.dumps({"type":"done","upid":upid,"ok":str(st.get('exitstatus','')).upper()=='OK'}))
+
+            time.sleep(2.0)
+    except Exception:
+        try: ws.close()
+        except Exception: pass
+
 if __name__ == "__main__":
     import argparse
     p = argparse.ArgumentParser()