Upload folder using huggingface_hub

x_node.py

@@ -1,3 +1,4 @@
+#!/usr/bin/env python3
 from __future__ import annotations
 import os
 import sys
@@ -7,36 +8,41 @@ import subprocess
 import platform
 import getpass
 import datetime
-from typing import Dict, Any
-import psutil
-
-#!/usr/bin/env python3
-"""
-x_node.py
-
-获取全部环境变量、当前进程信息，并检测 curl、wget 是否可调用（并尝试获取版本信息）。
-输出 JSON 到 stdout，兼容 Linux/macOS/Windows。尽量不依赖第三方库，若安装 psutil 会提供更详细信息。
-"""
-
+from typing import Dict, Any, Optional, Sequence
 
 # Try to import psutil if available for richer process info
 try:
+    import psutil  # type: ignore
 except Exception:
     psutil = None  # type: ignore
 
 
 def get_env_vars() -> Dict[str, str]:
     """Return all environment variables as a dict."""
+    # Convert to plain dict to avoid os._Environ metadata in output
     return dict(os.environ)
 
 
-def safe_run_version(cmd: list[str], timeout: float = 3.0) -> Dict[str, Any]:
-    """Try to run `cmd --version` (cmd is full command list) and capture output."""
+def _shorten_text(s: str, max_lines: int = 5, max_chars: int = 2000) -> str:
+    """Trim text to the given number of lines and characters to keep output small."""
+    out = "\n".join(s.splitlines()[:max_lines])
+    if len(out) > max_chars:
+        out = out[: max_chars - 3] + "..."
+    return out
+
+
+def safe_run_version(cmd: Sequence[str], timeout: float = 3.0) -> Dict[str, Any]:
+    """Try to run `cmd` and capture stdout/stderr (combined)."""
     try:
-        p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout, check=False)
+        p = subprocess.run(
+            list(cmd),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            timeout=timeout,
+            check=False,
+        )
         out = p.stdout.decode(errors="replace").strip()
-        # For long output, keep first few lines
-        short_out = "\n".join(out.splitlines()[:5])
+        short_out = _shorten_text(out)
         return {"ok": True, "rc": p.returncode, "output": short_out}
     except FileNotFoundError:
         return {"ok": False, "error": "not found"}
@@ -46,32 +52,49 @@ def safe_run_version(cmd: list[str], timeout: float = 3.0) -> Dict[str, Any]:
         return {"ok": False, "error": str(e)}
 
 
-def check_tools(tools: list[str]) -> Dict[str, Dict[str, Any]]:
-    """Check whether tools are callable (in PATH) and try to get version."""
+# Common flags to try for version/help info
+TOOL_VERSION_FLAGS = ("--version", "-V", "-v", "--help")
+
+
+def check_tools(tools: Sequence[str]) -> Dict[str, Dict[str, Any]]:
+    """Check whether tools are callable (in PATH) and try to get version/help output."""
     results: Dict[str, Dict[str, Any]] = {}
     for t in tools:
-        path = shutil.which(t)
-        info: Dict[str, Any] = {"path": path, "callable": bool(path)}
-        if path:
-            # Prefer conventional --version, fallback to -V, -v
-            for flag in ("--version", "-V", "-v"):
-                res = safe_run_version([t, flag])
-                if res.get("ok") and res.get("output"):
-                    info["version_cmd"] = flag
-                    info["version_output"] = res.get("output")
-                    break
+        found_path: Optional[str] = shutil.which(t)
+        info: Dict[str, Any] = {"path": found_path, "callable": bool(found_path)}
+        # Whether we should try to run "t" or the absolute path
+        cmd_base = found_path or t
+        # Try common flags to get version/help output
+        for flag in TOOL_VERSION_FLAGS:
+            res = safe_run_version([cmd_base, flag])
+            if res.get("ok") and res.get("output"):
+                info["version_cmd"] = flag
+                info["version_output"] = res.get("output")
+                break
+        else:
+            # final fallback: attempt with no args (some tools print version/usage by default)
+            res = safe_run_version([cmd_base])
+            if res.get("ok") and res.get("output"):
+                info["version_cmd"] = None
+                info["version_output"] = res.get("output")
             else:
-                # Try running with no args to see if there's any output
-                res = safe_run_version([t])
-                if res.get("ok"):
-                    info["version_cmd"] = None
-                    info["version_output"] = res.get("output")
-                else:
-                    info["version_error"] = res.get("error")
+                # If not found or error, pass the error info
+                info["version_error"] = res.get("error")
+                # Ensure callable is set correctly: if subprocess reports not found, set callable False
+                if res.get("error") == "not found":
+                    info["callable"] = False
         results[t] = info
     return results
 
 
+def _safe_get_user() -> Optional[str]:
+    try:
+        return getpass.getuser()
+    except Exception:
+        # Fallback to env variables
+        return os.environ.get("USER") or os.environ.get("USERNAME")
+
+
 def get_process_info() -> Dict[str, Any]:
     """Collect basic process information. If psutil is available, include richer info."""
     info: Dict[str, Any] = {}
@@ -83,14 +106,16 @@ def get_process_info() -> Dict[str, Any]:
     info["python_version"] = platform.python_version()
     info["python_executable"] = sys.executable
     info["argv"] = sys.argv
-    info["cwd"] = os.getcwd()
+    try:
+        info["cwd"] = os.getcwd()
+    except Exception:
+        info["cwd"] = None
     info["pid"] = os.getpid()
-    # ppid is available on most platforms
     try:
         info["ppid"] = os.getppid()
     except Exception:
         info["ppid"] = None
-    info["user"] = getpass.getuser()
+    info["user"] = _safe_get_user()
 
     # Add Unix-specific ids if available
     if hasattr(os, "getuid"):
@@ -105,24 +130,58 @@ def get_process_info() -> Dict[str, Any]:
             pass
 
     # Try using psutil for richer details
-    if psutil:
+    if psutil is not None:
         try:
             p = psutil.Process(info["pid"])
             with p.oneshot():
-                info["exe"] = p.exe()
-                info["cmdline"] = p.cmdline()
-                info["create_time"] = datetime.datetime.fromtimestamp(p.create_time()).isoformat()
-                info["cpu_percent"] = p.cpu_percent(interval=0.1)
-                mem = p.memory_info()
-                info["memory_rss"] = mem.rss
-                info["memory_vms"] = mem.vms
-                info["num_threads"] = p.num_threads()
+                try:
+                    info["exe"] = p.exe()
+                except Exception:
+                    info["exe"] = sys.executable
+                try:
+                    info["cmdline"] = p.cmdline()
+                except Exception:
+                    info["cmdline"] = sys.argv
+                try:
+                    ct = p.create_time()
+                    info["create_time"] = datetime.datetime.fromtimestamp(ct).isoformat()
+                except Exception:
+                    info["create_time"] = None
+                try:
+                    info["cpu_percent"] = p.cpu_percent(interval=0.1)
+                except Exception:
+                    info["cpu_percent"] = None
+                try:
+                    mem = p.memory_info()
+                    info["memory_rss"] = getattr(mem, "rss", None)
+                    info["memory_vms"] = getattr(mem, "vms", None)
+                except Exception:
+                    info["memory_rss"] = None
+                    info["memory_vms"] = None
+                try:
+                    info["num_threads"] = p.num_threads()
+                except Exception:
+                    info["num_threads"] = None
                 try:
                     info["open_files"] = [f.path for f in p.open_files()]
                 except Exception:
                     info["open_files"] = None
+                # p.connections() can be expensive and may require permissions; tolerate failures
                 try:
-                    info["connections"] = [str(c) for c in p.connections()]
+                    # Modern psutil uses net_connections
+                    conns = getattr(p, "connections", None)
+                    if callable(conns):
+                        conns_list = p.connections()
+                    else:
+                        net_conns = getattr(p, "net_connections", None)
+                        if callable(net_conns):
+                            conns_list = p.net_connections()
+                        else:
+                            conns_list = None
+                    if conns_list is None:
+                        info["connections"] = None
+                    else:
+                        info["connections"] = [str(c) for c in conns_list]
                 except Exception:
                     info["connections"] = None
         except Exception:
@@ -137,13 +196,22 @@ def get_process_info() -> Dict[str, Any]:
 
 
 def main() -> int:
-    env = get_env_vars()
-    process = get_process_info()
-    tools = check_tools(["curl", "wget"])
-    result = {"env": env, "process": process, "tools": tools}
-    # Print JSON
-    print(json.dumps(result, indent=2, ensure_ascii=False))
-    return 0
+    try:
+        env = get_env_vars()
+        process = get_process_info()
+        tools = check_tools(["curl", "wget"])
+        result = {"env": env, "process": process, "tools": tools}
+        # Ensure stable JSON output
+        out = json.dumps(result, indent=2, ensure_ascii=False, sort_keys=False)
+        print(out, flush=True)
+        return 0
+    except Exception as e:
+        # Print error message in JSON form for easier parsing
+        try:
+            print(json.dumps({"error": str(e)}), flush=True)
+        except Exception:
+            print(f"error: {e}", flush=True)
+        return 2
 
 
 main()