chatrebot_aireply_plug/scripts/mc_query.py

#!/usr/bin/env python3
"""统一入口：根据参数调用不同MC查询脚本"""

import sys, os, json, subprocess, textwrap

SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))

# 纯文本格式化函数（禁止Markdown）
def fmt_status(data):
    lines = []
    s = data
    if not s.get("online"):
        lines.append("服务器已离线")
        return "\n".join(lines)
    
    players = s.get("players", {})
    lines.append(f"在线玩家: {players.get('online', 0)}/{players.get('max', 0)}")
    plist = s.get("player_list", [])
    if plist:
        lines.append("玩家列表: " + ", ".join(plist))
    
    mem = s.get("memory_mb")
    if mem:
        lines.append(f"内存使用: {mem['mb']}MB ({mem['gb']}GB)")
    
    cpu = s.get("cpu_percent")
    if cpu is not None:
        lines.append(f"CPU占用: {cpu}%")
    
    uptime = s.get("uptime")
    if uptime:
        lines.append(f"运行时长: {uptime}")
    
    tps = s.get("tps")
    if tps:
        lines.append(f"TPS: {tps.get('tps_1m', 'N/A')} (1分钟)")
    
    lines.append(f"版本: {s.get('version', '未知')}")
    
    ws = s.get("world_sizes", {})
    if ws.get("total"):
        lines.append(f"存档总大小: {ws['total']}")
    if ws.get("disk_pct") is not None:
        lines.append(f"磁盘使用: {ws['disk_used_gb']}GB / {ws['disk_total_gb']}GB ({ws['disk_pct']}%)")
    
    load = s.get("system_load")
    if load:
        lines.append(f"系统负载: {load['load_1']} ({load['load_pct']}% @ {load['cpu_count']}核)")
    
    return "\n".join(lines)

def fmt_players(data):
    lines = []
    events = data.get("recent_events", [])
    if events:
        lines.append("最近动态:")
        for e in events:
            icon = "加入" if e["type"] == "join" else "离开"
            lines.append(f"  {e['player']}: {icon}")
    
    pt = data.get("playtime", [])
    if pt:
        lines.append("")
        lines.append("在线时间排行:")
        for i, p in enumerate(pt[:5], 1):
            lines.append(f"  {i}. {p['name']}: {p['formatted']}")
    
    deaths = data.get("deaths", [])
    if deaths:
        lines.append("")
        lines.append("死亡次数排行:")
        for i, p in enumerate(deaths[:5], 1):
            lines.append(f"  {i}. {p['name']}: {p['deaths']}次")
    
    ach = data.get("achievements", [])
    if ach:
        lines.append("")
        lines.append("成就数排行:")
        for i, p in enumerate(ach[:5], 1):
            lines.append(f"  {i}. {p['name']}: {p['achievements']}个")
    
    return "\n".join(lines) if lines else "暂无玩家数据"

def fmt_world(data):
    lines = []
    
    gt = data.get("game_time", {})
    if gt:
        lines.append(f"世界时间: 第{gt.get('day', 1)}天 {gt.get('time', '00:00')} ({gt.get('phase', '')})")
    
    es = data.get("entity_stats", {})
    if es:
        lines.append(f"实体总数: {es.get('total', 0)} (敌对:{es.get('hostile', 0)} 友好:{es.get('passive', 0)} 其他:{es.get('other', 0)})")
        top = es.get("top", [])
        if top:
            lines.append("实体排行: " + ", ".join([f"{name}({cnt})" for name, cnt in top]))
    
    net = data.get("network", {})
    if net:
        lines.append(f"网络流量: RX {net.get('rx', '?')} TX {net.get('tx', '?')}")
    
    mods = data.get("active_mods", [])
    if mods:
        lines.append(f"活跃模组: {', '.join(mods)}")
    
    return "\n".join(lines) if lines else "暂无世界数据"

def run_script(name):
    script = os.path.join(SCRIPTS_DIR, name)
    try:
        result = subprocess.run(
            [sys.executable, script],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode != 0:
            return {"error": f"脚本错误: {result.stderr[:200]}"}
        return json.loads(result.stdout)
    except subprocess.TimeoutExpired:
        return {"error": "查询超时"}
    except Exception as e:
        return {"error": str(e)}

COMMANDS = {
    "status": ("mc_status.py", fmt_status, "服务器状态"),
    "players": ("mc_players.py", fmt_players, "玩家信息"),
    "world": ("mc_world.py", fmt_world, "世界详情"),
    "clear": ("mc_clear_items.py", None, "清理掉落物"),
}

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Minecraft 信息查询工具")
    parser.add_argument("command", choices=list(COMMANDS.keys()) + ["all"], help="查询类型")
    parser.add_argument("--json", action="store_true", help="输出JSON而非纯文本")
    args = parser.parse_args()
    
    if args.command == "all":
        results = {}
        for key, (script, fmt, desc) in COMMANDS.items():
            if key == "clear":
                continue
            data = run_script(script)
            results[key] = {"data": data, "text": fmt(data) if fmt else str(data)}
        if args.json:
            print(json.dumps(results, ensure_ascii=False, indent=2))
        else:
            for key, val in results.items():
                print(f"=== {COMMANDS[key][2]} ===")
                print(val["text"])
                print("")
    else:
        script, fmt, desc = COMMANDS[args.command]
        data = run_script(script)
        if args.json:
            print(json.dumps(data, ensure_ascii=False, indent=2))
        else:
            if "error" in data:
                print(f"查询失败: {data['error']}")
            elif fmt:
                print(fmt(data))
            else:
                print(str(data))