skill-agent-log-s3/lib/collect.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 未来飞马
# 纯标准库，Python 3.6+ 兼容，禁止 pip 依赖

"""采集本机运行指标，输出 JSON（纯标准库，3.6+ 兼容，禁止 boto3/obs/pip 安装）"""

import json
import os
import subprocess
import datetime
import socket
import sys
import time
import argparse


def sh(cmd, timeout=15):
    """执行 shell 命令，失败返回空字符串"""
    try:
        result = subprocess.run(
            cmd,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
            timeout=timeout
        )
        return result.stdout.strip()
    except Exception:
        return ""


def collect(agent_id=None, userid=None):
    now = datetime.datetime.now()
    data = {
        "schema_version": "2.0",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%S+08:00"),
        "agent": agent_id or os.environ.get("AGENT_ID") or socket.gethostname(),
        "agent_id": agent_id or os.environ.get("FMODE_AGENT_ID") or os.environ.get("AGENT_ID") or socket.gethostname(),
        "userid": userid or os.environ.get("FMODE_USERID") or "",
        "hostname": socket.gethostname(),
        "resources": {},
        "concurrency": {},
        "services": {},
        "events": {}
    }

    # CPU / 内存（/proc/meminfo + /proc/loadavg）
    try:
        meminfo = {}
        with open("/proc/meminfo") as f:
            for line in f:
                k, v = line.split(":")
                meminfo[k.strip()] = int(v.strip().split()[0])
        total = meminfo["MemTotal"]
        avail = meminfo.get("MemAvailable", meminfo.get("MemFree", 0))
        data["resources"]["memory"] = {
            "total_mb": round(total / 1024),
            "used_mb": round((total - avail) / 1024),
            "percent": round((total - avail) / total * 100, 1)
        }
        load = open("/proc/loadavg").read().split()
        data["resources"]["cpu_load"] = {
            "m1": float(load[0]),
            "m5": float(load[1]),
            "m15": float(load[2])
        }
        cpu = sh("top -bn1 2>/dev/null | grep 'Cpu(s)' | awk '{print $2+$4}'")
        if cpu:
            try:
                data["resources"]["cpu_percent"] = round(float(cpu), 1)
            except ValueError:
                data["resources"]["cpu_percent"] = None
        else:
            data["resources"]["cpu_percent"] = None
    except Exception as e:
        data["resources"]["error"] = str(e)[:100]

    # 磁盘（根分区）
    disk = sh("df -B1 / 2>/dev/null | tail -1 | awk '{print $2, $3, $5}'")
    if disk:
        parts = disk.split()
        if len(parts) == 3:
            try:
                data["resources"]["disk"] = {
                    "total_gb": round(int(parts[0]) / 1e9, 1),
                    "used_gb": round(int(parts[1]) / 1e9, 1),
                    "percent": parts[2]
                }
            except ValueError:
                pass

    # 网络（主网卡，累计上下行）
    net = sh(
        "cat /proc/net/dev 2>/dev/null | "
        "awk 'NR>2 && $1!~/lo:/ {gsub(\":\",\"\",$1); print $1, $2, $10}' | "
        "sort -k2 -rn | head -1"
    )
    if net:
        parts = net.split()
        if len(parts) == 3:
            try:
                data["resources"]["network"] = {
                    "iface": parts[0],
                    "rx_total_mb": round(int(parts[1]) / 1e6, 1),
                    "tx_total_mb": round(int(parts[2]) / 1e6, 1)
                }
            except ValueError:
                pass

    # GPU（有则记，无则跳过）
    gpu = sh(
        "nvidia-smi --query-gpu=utilization.gpu,memory.used,memory.total "
        "--format=csv,noheader,nounits 2>/dev/null"
    )
    if gpu:
        parts = [p.strip() for p in gpu.split(",")]
        if len(parts) == 3:
            try:
                data["resources"]["gpu"] = {
                    "percent": int(parts[0]),
                    "mem_used_mb": int(parts[1]),
                    "mem_total_mb": int(parts[2])
                }
            except ValueError:
                pass

    # 并发：profiles 总数
    profiles = sh("ls /opt/data/profiles/ 2>/dev/null | wc -l")
    data["concurrency"]["profiles_total"] = int(profiles) if profiles.isdigit() else None

    # 并发：活跃 session / 运行中 subagent（hermes state.db）
    try:
        import sqlite3
        conn = sqlite3.connect("file:/opt/data/state.db?mode=ro", uri=True, timeout=5)
        cur = conn.cursor()
        cur.execute(
            "SELECT COUNT(DISTINCT session_id) FROM messages WHERE timestamp > ?",
            (time.time() - 1800,)
        )
        data["concurrency"]["active_sessions_30min"] = cur.fetchone()[0]
        cur.execute("SELECT COUNT(*) FROM async_delegations WHERE status='running'")
        data["concurrency"]["running_subagents"] = cur.fetchone()[0]
        conn.close()
    except Exception:
        data["concurrency"]["active_sessions_30min"] = None
        data["concurrency"]["running_subagents"] = None

    # 服务存活（gateway / dashboard / studio）
    for svc, pat in [
        ("gateway", "hermes gateway"),
        ("dashboard", "hermes-dashboard"),
        ("studio", "fmode-studio")
    ]:
        out = sh("pgrep -f '{}' 2>/dev/null | wc -l".format(pat))
        data["services"][svc] = "up" if out.isdigit() and int(out) > 0 else "down"

    # 事件：最近 1 小时错误计数
    errc = sh(
        "find /opt/data/logs -name '*.log' -mmin -60 -exec "
        "grep -ci 'error\\|fatal' {} + 2>/dev/null | "
        "awk -F: '{s+=$2} END {print s}'"
    )
    data["events"]["errors_last_1h"] = int(errc) if errc.isdigit() else None

    rest = sh("grep -c 'Reconnected' /opt/data/logs/gateway.log 2>/dev/null")
    data["events"]["wecom_reconnects_total"] = int(rest) if rest.isdigit() else None

    return data


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="采集本机指标输出 JSON")
    parser.add_argument("--agent-id", default=None, help="Agent 标识")
    parser.add_argument("--userid", default=None, help="用户 ID")
    args = parser.parse_args()

    result = collect(agent_id=args.agent_id, userid=args.userid)
    print(json.dumps(result, ensure_ascii=False))