Upload folder using huggingface_hub

x_node.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+import os
+import sys
+import json
+import shutil
+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 会提供更详细信息。
+"""
+
+
+# Try to import psutil if available for richer process info
+try:
+except Exception:
+    psutil = None  # type: ignore
+
+
+def get_env_vars() -> Dict[str, str]:
+    """Return all environment variables as a dict."""
+    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."""
+    try:
+        p = subprocess.run(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])
+        return {"ok": True, "rc": p.returncode, "output": short_out}
+    except FileNotFoundError:
+        return {"ok": False, "error": "not found"}
+    except subprocess.TimeoutExpired:
+        return {"ok": False, "error": "timeout"}
+    except Exception as e:
+        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."""
+    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
+            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")
+        results[t] = info
+    return results
+
+
+def get_process_info() -> Dict[str, Any]:
+    """Collect basic process information. If psutil is available, include richer info."""
+    info: Dict[str, Any] = {}
+    info["platform"] = platform.system()
+    info["platform_release"] = platform.release()
+    info["platform_version"] = platform.version()
+    info["machine"] = platform.machine()
+    info["processor"] = platform.processor()
+    info["python_version"] = platform.python_version()
+    info["python_executable"] = sys.executable
+    info["argv"] = sys.argv
+    info["cwd"] = os.getcwd()
+    info["pid"] = os.getpid()
+    # ppid is available on most platforms
+    try:
+        info["ppid"] = os.getppid()
+    except Exception:
+        info["ppid"] = None
+    info["user"] = getpass.getuser()
+
+    # Add Unix-specific ids if available
+    if hasattr(os, "getuid"):
+        try:
+            info["uid"] = os.getuid()
+            info["gid"] = os.getgid()
+            if hasattr(os, "geteuid"):
+                info["euid"] = os.geteuid()
+            if hasattr(os, "getegid"):
+                info["egid"] = os.getegid()
+        except Exception:
+            pass
+
+    # Try using psutil for richer details
+    if psutil:
+        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["open_files"] = [f.path for f in p.open_files()]
+                except Exception:
+                    info["open_files"] = None
+                try:
+                    info["connections"] = [str(c) for c in p.connections()]
+                except Exception:
+                    info["connections"] = None
+        except Exception:
+            # Fall back to minimal info already present
+            pass
+    else:
+        # Fallback values
+        info["exe"] = sys.executable
+        info["cmdline"] = sys.argv
+        info["create_time"] = None
+    return info
+
+
+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
+
+
+main()