e1fbd31c64
- 高危词库从 process.py 抽到 config.toml [security] section - process.py 改为 _get_high_risk_words() 动态加载（配置优先） - 新增 15 个 QQ 操作脚本（已脱敏：QQ号/IP/Token → 占位符） - 新增 8 个 SKILL.md（qq-messenger/management/resolver/napcat-extras 等） - 新增 SKILL.md 入口（完整部署方案文档） - 保留 SDK 框架（plugin_modules.py, file_store_api.py, package.py 等）
211 lines
7.2 KiB
Python
211 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
"""获取Minecraft玩家活动信息（JSON输出）"""
|
||
|
||
import sys, os, re, json
|
||
from datetime import datetime
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from mc_config import MC_SERVER_DIR, PLAYTIME_DB
|
||
|
||
def get_recent_events(limit=5):
|
||
log_path = os.path.join(MC_SERVER_DIR, "logs", "latest.log")
|
||
if not os.path.exists(log_path):
|
||
return []
|
||
events = []
|
||
patterns = [
|
||
(re.compile(r'(\w+)\s+joined\s+the\s+game'), 'join'),
|
||
(re.compile(r'(\w+)\s+left\s+the\s+game'), 'leave'),
|
||
]
|
||
skip = {"Server", "RCON", "Thread", "Generic", "World"}
|
||
try:
|
||
with open(log_path, "r", errors="ignore") as f:
|
||
lines = f.readlines()
|
||
for line in lines[-5000:]:
|
||
for pattern, etype in patterns:
|
||
m = pattern.search(line)
|
||
if m:
|
||
name = m.group(1)
|
||
if name in skip:
|
||
continue
|
||
events.append({"type": etype, "player": name})
|
||
break
|
||
seen = set()
|
||
unique = []
|
||
for e in reversed(events):
|
||
key = (e["player"], e["type"])
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique.append(e)
|
||
if len(unique) >= limit:
|
||
break
|
||
return list(reversed(unique))
|
||
except Exception:
|
||
return []
|
||
|
||
def get_playtime_db():
|
||
if os.path.exists(PLAYTIME_DB):
|
||
try:
|
||
with open(PLAYTIME_DB, "r") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
pass
|
||
return {"players": {}, "last_log_pos": 0}
|
||
|
||
def save_playtime_db(db):
|
||
try:
|
||
with open(PLAYTIME_DB, "w") as f:
|
||
json.dump(db, f, indent=2, ensure_ascii=False)
|
||
except Exception:
|
||
pass
|
||
|
||
def update_playtime():
|
||
db = get_playtime_db()
|
||
log_path = os.path.join(MC_SERVER_DIR, "logs", "latest.log")
|
||
if not os.path.exists(log_path):
|
||
return db
|
||
try:
|
||
with open(log_path, "rb") as f:
|
||
f.seek(0, 2)
|
||
size = f.tell()
|
||
start = min(db.get("last_log_pos", 0), size)
|
||
if start > size:
|
||
start = 0
|
||
with open(log_path, "r", errors="ignore") as f:
|
||
f.seek(start)
|
||
lines = f.readlines()
|
||
now = datetime.now()
|
||
join_p = re.compile(r'joined the game')
|
||
leave_p = re.compile(r'left the game')
|
||
name_p = re.compile(r'\]:\s+(\w+)\s+(?:joined|left)')
|
||
for line in lines:
|
||
m = name_p.search(line)
|
||
if not m:
|
||
continue
|
||
player = m.group(1)
|
||
if join_p.search(line):
|
||
if player not in db["players"]:
|
||
db["players"][player] = {"total_seconds": 0, "last_join": None}
|
||
db["players"][player]["last_join"] = now.isoformat()
|
||
elif leave_p.search(line):
|
||
if player in db["players"] and db["players"][player].get("last_join"):
|
||
join_time = datetime.fromisoformat(db["players"][player]["last_join"])
|
||
sec = (now - join_time).total_seconds()
|
||
if 0 < sec < 86400:
|
||
db["players"][player]["total_seconds"] += sec
|
||
db["players"][player]["last_join"] = None
|
||
db["last_log_pos"] = size
|
||
save_playtime_db(db)
|
||
except Exception:
|
||
pass
|
||
return db
|
||
|
||
def format_time(seconds):
|
||
h = seconds // 3600
|
||
m = (seconds % 3600) // 60
|
||
if h >= 24:
|
||
d = h // 24
|
||
h = h % 24
|
||
return f"{d}天{h}小时{m}分"
|
||
elif h > 0:
|
||
return f"{h}小时{m}分"
|
||
return f"{m}分钟"
|
||
|
||
def get_playtime_leaderboard(top_n=10):
|
||
db = update_playtime()
|
||
players = db.get("players", {})
|
||
lb = []
|
||
for name, data in players.items():
|
||
total = data.get("total_seconds", 0)
|
||
if data.get("last_join"):
|
||
try:
|
||
jt = datetime.fromisoformat(data["last_join"])
|
||
s = (datetime.now() - jt).total_seconds()
|
||
if 0 < s < 86400:
|
||
total += s
|
||
except Exception:
|
||
pass
|
||
if total > 0:
|
||
lb.append({"name": name, "seconds": int(total), "formatted": format_time(int(total))})
|
||
lb.sort(key=lambda x: -x["seconds"])
|
||
return lb[:top_n]
|
||
|
||
def _load_uuid_to_name():
|
||
"""从 usercache.json 加载 UUID->名字映射，回退 ops.json/whitelist.json"""
|
||
mapping = {}
|
||
# 主源：usercache.json（包含所有登录过的玩家）
|
||
for src_file in ["usercache.json", "ops.json", "whitelist.json"]:
|
||
path = os.path.join(MC_SERVER_DIR, src_file)
|
||
try:
|
||
with open(path, "r") as f:
|
||
for entry in json.load(f):
|
||
if 'uuid' in entry and 'name' in entry:
|
||
mapping[entry['uuid']] = entry['name']
|
||
except Exception:
|
||
pass
|
||
return mapping
|
||
|
||
def get_death_leaderboard(top_n=10):
|
||
uuid_to_name = _load_uuid_to_name()
|
||
stats_dir = os.path.join(MC_SERVER_DIR, "world", "stats")
|
||
if not os.path.exists(stats_dir):
|
||
return []
|
||
deaths = {}
|
||
for fname in os.listdir(stats_dir):
|
||
if not fname.endswith('.json'):
|
||
continue
|
||
uuid = fname.replace('.json', '')
|
||
fpath = os.path.join(stats_dir, fname)
|
||
try:
|
||
with open(fpath, "r") as f:
|
||
data = json.load(f)
|
||
dc = data.get('stats', {}).get('minecraft:custom', {}).get('minecraft:deaths', 0)
|
||
if dc > 0:
|
||
name = uuid_to_name.get(uuid, uuid[:8])
|
||
deaths[name] = dc
|
||
except Exception:
|
||
continue
|
||
return [{"name": n, "deaths": d} for n, d in sorted(deaths.items(), key=lambda x: -x[1])[:top_n]]
|
||
|
||
def get_achievement_leaderboard(top_n=10):
|
||
uuid_to_name = _load_uuid_to_name()
|
||
adv_dir = os.path.join(MC_SERVER_DIR, "world", "advancements")
|
||
if not os.path.exists(adv_dir):
|
||
return []
|
||
ach = {}
|
||
for fname in os.listdir(adv_dir):
|
||
if not fname.endswith('.json'):
|
||
continue
|
||
uuid = fname.replace('.json', '')
|
||
fpath = os.path.join(adv_dir, fname)
|
||
try:
|
||
with open(fpath, "r") as f:
|
||
data = json.load(f)
|
||
name = uuid_to_name.get(uuid)
|
||
if not name:
|
||
name = uuid[:8] # 名字查不到就用 UUID 前缀
|
||
done = 0
|
||
for k, v in data.items():
|
||
if not isinstance(v, dict):
|
||
continue
|
||
if 'recipes/' in k or ':recipes/' in k:
|
||
continue
|
||
if v.get('done'):
|
||
done += 1
|
||
if done > 0:
|
||
ach[name] = done
|
||
except Exception:
|
||
continue
|
||
return [{"name": n, "achievements": a} for n, a in sorted(ach.items(), key=lambda x: -x[1])[:top_n]]
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--top", type=int, default=10)
|
||
args = parser.parse_args()
|
||
result = {
|
||
"recent_events": get_recent_events(5),
|
||
"playtime": get_playtime_leaderboard(args.top),
|
||
"deaths": get_death_leaderboard(args.top),
|
||
"achievements": get_achievement_leaderboard(args.top),
|
||
}
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|