lambdagent/agentexample/agent67/tools/system_info.py

"""
agent67.tools.system_info — 系统信息查询

Lambda: λ query_json. system_info(query)
"""
import json
import platform
import subprocess


def _run(cmd: str) -> str:
    try:
        r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
        return r.stdout.strip()
    except Exception:
        return "N/A"


def query_system(input_json: str) -> str:
    """
    查询系统信息。

    支持: cpu / memory / disk / network / process / battery / all
    """
    try:
        data = json.loads(input_json)
        query = data.get("query", "all")
    except (json.JSONDecodeError, AttributeError):
        query = str(input_json).strip().lower()

    is_mac = platform.system() == "Darwin"
    sections = []

    if query in ("cpu", "all"):
        if is_mac:
            cpu_info = _run("sysctl -n machdep.cpu.brand_string")
            cores = _run("sysctl -n hw.ncpu")
            cpu_usage = _run("top -l 1 -n 0 | grep 'CPU usage'")
            sections.append(f"🖥️ CPU:\n  型号: {cpu_info}\n  核心数: {cores}\n  {cpu_usage}")
        else:
            sections.append(f"🖥️ CPU:\n  {_run('lscpu | head -15')}")

    if query in ("memory", "mem", "all"):
        if is_mac:
            total = _run("sysctl -n hw.memsize")
            try:
                total_gb = int(total) / (1024 ** 3)
                sections.append(f"🧠 内存: {total_gb:.1f} GB")
            except ValueError:
                sections.append(f"🧠 内存: {total}")
        else:
            sections.append(f"🧠 内存:\n  {_run('free -h')}")

    if query in ("disk", "all"):
        sections.append(f"💾 磁盘:\n  {_run('df -h / | tail -1')}")

    if query in ("network", "net", "all"):
        ip_info = _run("ifconfig | grep 'inet ' | grep -v 127.0.0.1")
        sections.append(f"🌐 网络:\n  {ip_info}")

    if query in ("process", "proc", "top"):
        top_procs = _run("ps aux --sort=-%cpu 2>/dev/null | head -6 || ps aux -r | head -6")
        sections.append(f"📊 Top 进程:\n{top_procs}")

    if query in ("battery", "bat", "all") and is_mac:
        battery = _run("pmset -g batt")
        sections.append(f"🔋 电池:\n  {battery}")

    if not sections:
        return f"❌ 未知查询: {query}. 支持: cpu/memory/disk/network/process/battery/all"

    return "\n\n".join(sections)