chatrebot_aireply_plug/scripts/mc_world.py

#!/usr/bin/env python3
"""获取Minecraft世界详细信息（JSON输出）"""

import sys, os, subprocess, re
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mc_config import RCON_HOST, RCON_PORT, RCON_PASSWORD, MC_SERVER_DIR
from mc_rcon import RCONClient

def get_entity_stats():
    try:
        with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
            resp = client.send_command("forge entity list")
        body = resp.get("body", "") if resp else ""
        if not body:
            return {}
        m = re.search(r'Total:\s*(\d+)', body)
        total = int(m.group(1)) if m else 0
        entity_map = {}
        for line in body.strip().split("\n")[1:]:
            line = line.strip()
            em = re.match(r'(\d+):\s*(.+)', line)
            if em:
                entity_map[em.group(2).strip()] = int(em.group(1))
        hostile_types = {"zombie", "skeleton", "creeper", "spider", "enderman", "witch", "slime", "phantom", "husk", "stray", "drowned", "pillager", "vindicator", "evoker", "ravager", "vex", "hoglin", "zoglin", "piglin", "piglin_brute", "wither_skeleton", "blaze", "ghast", "magma_cube", "guardian", "elder_guardian", "shulker", "endermite", "silverfish"}
        passive_types = {"sheep", "cow", "pig", "chicken", "horse", "donkey", "mule", "rabbit", "fox", "bee", "cat", "wolf", "parrot", "turtle", "panda", "polar_bear", "llama", "mooshroom", "bat", "squid", "dolphin", "cod", "salmon", "tropical_fish", "pufferfish", "villager", "wandering_trader", "iron_golem", "snow_golem", "frog", "axolotl", "goat", "sniffer", "camel"}
        hostile, passive, other = 0, 0, 0
        for name, count in entity_map.items():
            nl = name.replace("minecraft:", "").lower()
            if nl in hostile_types:
                hostile += count
            elif nl in passive_types:
                passive += count
            else:
                other += count
        top = sorted(entity_map.items(), key=lambda x: -x[1])[:5]
        return {"total": total, "hostile": hostile, "passive": passive, "other": other, "top": top}
    except Exception:
        return {}

def get_game_time():
    try:
        with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
            resp_day = client.send_command("time query day")
            resp_daytime = client.send_command("time query daytime")
        body_day = resp_day.get("body", "") if resp_day else ""
        body_daytime = resp_daytime.get("body", "") if resp_daytime else ""
        m_day = re.search(r'The time is (\d+)', body_day)
        m_daytime = re.search(r'The time is (\d+)', body_daytime)
        day = int(m_day.group(1)) + 1 if m_day else 1
        daytime = int(m_daytime.group(1)) if m_daytime else 0
        years = day // 365
        months = (day % 365) // 30
        days = (day % 365) % 30
        hours = int(daytime / 1000)
        minutes = int((daytime % 1000) * 60 / 1000)
        if 0 <= daytime < 1000:
            phase = "日出"
        elif 1000 <= daytime < 12000:
            phase = "白天"
        elif 12000 <= daytime < 13000:
            phase = "日落"
        elif 13000 <= daytime < 23000:
            phase = "夜晚"
        else:
            phase = "黎明"
        return {"day": day, "years": years, "months": months, "days": days, "time": f"{hours:02d}:{minutes:02d}", "phase": phase}
    except Exception:
        return {}

def get_network_stats():
    try:
        with open("/proc/net/dev", "r") as f:
            lines = f.readlines()
        total_rx, total_tx = 0, 0
        for line in lines[2:]:
            parts = line.strip().split()
            if len(parts) >= 10 and not parts[0].startswith("lo:"):
                total_rx += int(parts[1])
                total_tx += int(parts[9])
        def fmt(b):
            if b >= 1024**3:
                return f"{b/1024**3:.1f}GB"
            elif b >= 1024**2:
                return f"{b/1024**2:.0f}MB"
            elif b >= 1024:
                return f"{b/1024:.0f}KB"
            return f"{b}B"
        return {"rx": fmt(total_rx), "tx": fmt(total_tx)}
    except Exception:
        return {}

def get_active_mods(limit=3):
    log_path = os.path.join(MC_SERVER_DIR, "logs", "debug.log")
    if not os.path.exists(log_path):
        return []
    skip = ["net.minecraft", "net.minecraftforge", "Server thread", "RCON", "Thread", "Generic", "mojang", "java", "sun.", "com.google", "org.apache", "io.netty", "org.mtr", "Mixin", "dev.architectury", "libIPN", "MinecraftMappings", "Yggdrasil", "DataFixer"]
    counts = {}
    try:
        with open(log_path, "r", errors="ignore") as f:
            lines = f.readlines()
        for line in lines[-2000:]:
            m = re.search(r'\[([^\/\]]+)/\]', line)
            if not m:
                continue
            mod = m.group(1)
            if any(mod.startswith(s) for s in skip) or len(mod) < 3:
                continue
            counts[mod] = counts.get(mod, 0) + 1
        return [m[0] for m in sorted(counts.items(), key=lambda x: -x[1])[:limit]]
    except Exception:
        return []

if __name__ == "__main__":
    import json
    result = {
        "entity_stats": get_entity_stats(),
        "game_time": get_game_time(),
        "network": get_network_stats(),
        "active_mods": get_active_mods(),
    }
    print(json.dumps(result, ensure_ascii=False, indent=2))