refactor(agent,teacher): extract shared utils to common/utils; remove wincli_escape

Author: misakano7545

.gitignore

@@ -15,6 +15,7 @@ __pycache__/
 # Clash Meta
 mihomo.exe
 configs/config.yaml
+agent/configs
 
 # Trae
 .trae/

agent/configs/GeoIP.dat

@@ -3,14 +3,14 @@
 from __future__ import annotations
 
 import os
+import sys
+import time
 import shutil
 import socket
-import subprocess
-import sys
-import tempfile
 import threading
-import time
+import subprocess
 from typing import Any, Dict, List, Optional, Tuple
+from common.utils import is_admin, get_subprocess_flags
 
 _script_dir = os.path.dirname(os.path.abspath(sys.executable)) if getattr(sys, "frozen", False) else os.path.dirname(os.path.abspath(__file__))
 CLASH_CONFIG_DIR = os.path.join(_script_dir, "configs")
@@ -20,20 +20,6 @@
 _clash_process: Optional[subprocess.Popen] = None
 
 
-def _subprocess_flags() -> int:
-    return subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
-
-
-def is_admin() -> bool:
-    if sys.platform != "win32":
-        return False
-    try:
-        import ctypes
-        return bool(ctypes.windll.shell32.IsUserAnAdmin())
-    except Exception:
-        return False
-
-
 def _get_clash_binary_path() -> Tuple[bool, str]:
     script_dir = os.path.dirname(os.path.abspath(__file__))
     project_root = os.path.dirname(script_dir)
@@ -239,7 +225,7 @@ def _start_clash_process(clash_path: str) -> Tuple[bool, str]:
             [clash_path, "-f", CLASH_CONFIG_FILE, "-d", CLASH_CONFIG_DIR],
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
-            creationflags=_subprocess_flags(),
+            creationflags=get_subprocess_flags(),
         )
         time.sleep(1.5)
         if _clash_process.poll() is not None:
@@ -376,4 +362,4 @@ def get_current_restriction_status() -> Dict[str, Any]:
     except Exception:
         pass
 
-    return status
+    return status

agent/configs/cache.db

@@ -2,36 +2,15 @@
 
 from __future__ import annotations
 
-import json
 import re
-import subprocess
 import sys
-from typing import Dict, List, Tuple
-
+import json
+import subprocess
+from typing import List, Tuple
+from common.utils import get_subprocess_flags, parse_wmic_list
 from agent.network_config import get_default_ipv4_interface_name
 
 
-def _subprocess_flags() -> int:
-    return subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
-
-
-def _parse_wmic_list(text: str) -> List[Dict[str, str]]:
-    text = text.replace("\r\r\n", "\n").replace("\r\n", "\n").replace("\r", "\n").strip()
-    blocks = re.split(r"\n\s*\n", text)
-    rows: List[Dict[str, str]] = []
-    for block in blocks:
-        item: Dict[str, str] = {}
-        for line in block.splitlines():
-            line = line.strip()
-            if not line or "=" not in line:
-                continue
-            k, v = line.split("=", 1)
-            item[k.strip().lower()] = v.strip()
-        if item:
-            rows.append(item)
-    return rows
-
-
 def _normalize_mac(mac: str) -> str:
     if not mac:
         return ""
@@ -92,12 +71,12 @@ def _physical_adapter_mac_rows_wmi() -> List[Tuple[str, str]]:
             capture_output=True,
             text=True,
             timeout=30,
-            creationflags=_subprocess_flags(),
+            creationflags=get_subprocess_flags(),
         )
         if p.returncode != 0:
             return []
         out: List[Tuple[str, str]] = []
-        for row in _parse_wmic_list(p.stdout or ""):
+        for row in parse_wmic_list(p.stdout or ""):
             ne = row.get("netenabled", "").strip().lower()
             if ne in ("false", "0"):
                 continue
@@ -139,7 +118,7 @@ def _physical_adapter_mac_rows_powershell() -> List[Tuple[str, str]]:
             capture_output=True,
             text=True,
             timeout=45,
-            creationflags=_subprocess_flags(),
+            creationflags=get_subprocess_flags(),
         )
     except (OSError, subprocess.TimeoutExpired):
         return []
@@ -188,4 +167,4 @@ def get_machine_id() -> str:
         return "nonwin"
 
     mac = _get_preferred_mac_normalized()
-    return mac or "mac-unknown"
+    return mac or "mac-unknown"

agent/firewall_manager.py

@@ -2,21 +2,22 @@
 
 from __future__ import annotations
 
-import argparse
-import json
 import os
-import platform
+import sys
+import json
+import time
 import random
 import socket
-import subprocess
-import sys
+import argparse
+import platform
 import threading
-import time
-from typing import Any, Dict, Optional
-
+import subprocess
 from PIL import Image, ImageDraw
-
+from common.utils import is_admin
+from typing import Any, Dict, Optional
+from agent.rename_host import rename_computer
 from common.paths import default_agent_config_path, resolve_config_path
+from agent.machine_identity import get_machine_id, get_preferred_mac_display
 from common.protocol import (
     MSG_ACK,
     MSG_COMMAND_NETWORK_RESTRICT,
@@ -35,16 +36,12 @@
     read_frame_from_socket,
     write_frame_to_socket,
 )
-
-from agent.machine_identity import get_machine_id, get_preferred_mac_display
 from agent.network_config import (
     apply_ipv4_dhcp,
     apply_ipv4_static,
     get_default_ipv4_detail_snapshot,
     get_default_ipv4_interface_name,
-    is_admin,
 )
-from agent.rename_host import rename_computer
 from agent.firewall_manager import (
     apply_blacklist,
     apply_whitelist,
@@ -209,8 +206,6 @@ def _register(self, conn: socket.socket) -> bool:
                 "ipv4_detail": get_default_ipv4_detail_snapshot(),
                 "os_version": _get_os_version(),
                 "agent_version": AGENT_VERSION,
-                # Keep legacy key for backward compatibility with old teacher builds.
-                "version": AGENT_VERSION,
                 "token": self.token,
             },
         )