mail

collect.py

@@ -1,3 +1,4 @@
+from typing import List, Dict, Tuple, Set, Any, Optional, Union
 #!/usr/bin/env python3
 """
 collect.py — Palladium Z1 远程数据采集推送脚本
@@ -46,7 +47,7 @@ DEFAULT_RETRY_DELAY = 5
 class Logger:
     """同时输出到 stderr 和可选的日志文件。"""
 
-    def __init__(self, log_file: str | None = None):
+    def __init__(self, log_file: Optional[str] = None):
         self.log_file = log_file
 
     def _emit(self, level: str, msg: str, ts: str) -> None:
@@ -73,7 +74,7 @@ class Logger:
 # ═══════════════════════════════════════════════════════════════════════════
 #  Command Execution
 # ═══════════════════════════════════════════════════════════════════════════
-def build_shell_cmd(cmd: str, env_script: str | None) -> str:
+def build_shell_cmd(cmd: str, env_script: Optional[str]) -> str:
     """
     如果指定了 env_script，把 source env 之后再执行 cmd 拼成一个 shell 串。
     不指定则直接返回 cmd。
@@ -90,7 +91,7 @@ def shlex_quote(s: str) -> str:
     return "'" + s.replace("'", "'\\''") + "'"
 
 
-def run_test_server(cmd: str, timeout: int, env_script: str | None) -> tuple[str, int, str]:
+def run_test_server(cmd: str, timeout: int, env_script: Optional[str]) -> Tuple[str, int, str]:
     """
     执行 test_server -short, 返回 (stdout, returncode, stderr)。
     timeout 触发时 subprocess.run 不会抛 (>=3.12 subprocessing 时可能抛), 这里显式处理。
@@ -101,7 +102,7 @@ def run_test_server(cmd: str, timeout: int, env_script: str | None) -> tuple[str
             full_cmd,
             shell=True,
             executable="/bin/bash",
-            capture_output=True,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
             timeout=timeout,
         )
     except subprocess.TimeoutExpired as e:
@@ -120,7 +121,7 @@ def run_test_server(cmd: str, timeout: int, env_script: str | None) -> tuple[str
 # ═══════════════════════════════════════════════════════════════════════════
 def push_to_monitor(url: str, raw_output: str, source: str,
                     retries: int, retry_delay: int,
-                    log: Logger) -> tuple[bool, dict | None, str]:
+                    log: Logger) -> Tuple[bool, Optional[Dict], str]:
     """带重试的 HTTP 推送, 返回 (ok, json_body_or_None, error_msg)。"""
     payload = {"raw_output": raw_output, "source": source}
     last_err = ""