マイクラ教育版サーバをDiscord監視botで「完全に把握」してみた

マイクラ教育版（Minecraft: Education Edition）、通称EEのサーバ運用楽しんでる？

※前回 マイクラ教育版のサーバを構築しました。

「子供や友達と一緒に遊ぶのに便利～！」って立てたはいいもののさ、

そのままでは、ちょっとまずいのが、バックアップ。

もう一つが、誰か来たのか問題。

「……で、今だれかログインしてんの？」

これよ。いちいちゲーム画面開いて確認するのダルすぎでしょ。

世の中のサーバ運用の基本は「自動化」と「監視」なわけ。

今回は超ニッチだけど、5分おきに誰がインしてるかのログをDiscordにシュッと通知させて、

ついでにマイクラ内のチャットもDiscordに連携する方法を共有します。

5分置きにチャットメッセージをDiscordに送るスクリプト

まずはマイクラEEのサーバ上で動かすスクリプトを書くよ。

要するに「定期的にチャットログを確認してログをDiscordのWebhookに投げる」っていうシンプル設計。

スクリプト本体

①定期バックアップ

下記をCrontabで定期実行。

#!/bin/bash

# バックアップ先ディレクトリ
BACKUP_DIR="/home/ubuntu/backup/"

# 1. バックアップの実行
tar cvfJ "${BACKUP_DIR}mcee-$(date +%Y%m%d_%H%M%S).tar" /opt/minecraft/worlds/

# 2. 現在のバックアップファイル総数をカウント
FILE_COUNT=$(find "$BACKUP_DIR" -maxdepth 1 -name "mcee-*.tar" | wc -l)

# 3. ファイルが7個以上ある場合のみ、1週間より古いファイルを削除
if [ "$FILE_COUNT" -ge 7 ]; then
    find "$BACKUP_DIR" -maxdepth 1 -name "mcee-*.tar" -mtime +7 -delete
    echo "7個以上のファイルが存在するため、1週間より古いファイルを削除しました。（現在の総数: ${FILE_COUNT}個）"
else
    echo "ファイルの総数が7個未満のため、削除処理をスキップしました。（現在の総数: ${FILE_COUNT}個）"
fi

②Discordにメッセージを送るスクリプト

DISCORD_WEBHOOK_URL=XXXXを設定後スクリプトを起動してください。

#!/usr/bin/env python3
"""Watch Minecraft EE chat logs and forward appended lines to Discord.
Also periodically backs up the world directory with tar.

Required:
  DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/..."

Example:
  python3 mcee_chatlog_to_discord.py
"""

from __future__ import annotations

import argparse
import json
import os
import signal
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


DEFAULT_LOG_DIR = "/opt/minecraft/chat_logs"
DEFAULT_INTERVAL_SECONDS = 300
DEFAULT_GLOB = "ChatLog_*.txt"
DISCORD_LIMIT = 2000
CONTENT_LIMIT = 1900

DEFAULT_BACKUP_SOURCE = "/opt/minecraft/worlds/"
DEFAULT_BACKUP_DIR = "/opt/mybackups"
DEFAULT_BACKUP_INTERVAL_SECONDS = 43200  # 12 hours (twice a day)

running = True


def handle_signal(signum: int, frame: Any) -> None:
    del signum, frame
    global running
    running = False


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Forward appended Minecraft EE chat log text to Discord, and back up worlds periodically."
    )
    parser.add_argument(
        "--test",
        action="store_true",
        help="Send a test message with the last few lines of the latest log file to Discord, then exit.",
    )
    parser.add_argument(
        "--log-dir",
        default=os.environ.get("MCEE_CHAT_LOG_DIR", DEFAULT_LOG_DIR),
        help=f"log directory (default: {DEFAULT_LOG_DIR})",
    )
    parser.add_argument(
        "--webhook-url",
        default=os.environ.get("DISCORD_WEBHOOK_URL"),
        help="Discord webhook URL (default: DISCORD_WEBHOOK_URL)",
    )
    parser.add_argument(
        "--interval",
        type=int,
        default=int(os.environ.get("MCEE_POLL_INTERVAL", DEFAULT_INTERVAL_SECONDS)),
        help=f"poll interval seconds (default: {DEFAULT_INTERVAL_SECONDS})",
    )
    parser.add_argument(
        "--state-file",
        default=os.environ.get(
            "MCEE_CHATLOG_STATE",
            str(Path.home() / ".mcee_chatlog_to_discord_state.json"),
        ),
        help="offset state JSON path",
    )
    parser.add_argument(
        "--glob",
        default=os.environ.get("MCEE_CHAT_LOG_GLOB", DEFAULT_GLOB),
        help=f"log filename glob (default: {DEFAULT_GLOB})",
    )
    parser.add_argument(
        "--send-existing",
        action="store_true",
        help="send existing unread contents on first run instead of starting at EOF",
    )

    # --- backup options ---
    parser.add_argument(
        "--no-backup",
        action="store_true",
        default=os.environ.get("MCEE_BACKUP_DISABLE", "") not in ("", "0", "false", "False"),
        help="disable periodic world backups",
    )
    parser.add_argument(
        "--backup-source",
        default=os.environ.get("MCEE_BACKUP_SOURCE", DEFAULT_BACKUP_SOURCE),
        help=f"directory to back up (default: {DEFAULT_BACKUP_SOURCE})",
    )
    parser.add_argument(
        "--backup-dir",
        default=os.environ.get("MCEE_BACKUP_DIR", DEFAULT_BACKUP_DIR),
        help=f"directory to store backup tar files (default: {DEFAULT_BACKUP_DIR})",
    )
    parser.add_argument(
        "--backup-interval",
        type=int,
        default=int(os.environ.get("MCEE_BACKUP_INTERVAL", DEFAULT_BACKUP_INTERVAL_SECONDS)),
        help=f"seconds between backups (default: {DEFAULT_BACKUP_INTERVAL_SECONDS})",
    )
    parser.add_argument(
        "--backup-on-start",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="take a backup immediately on startup instead of waiting for the first interval (default: enabled; use --no-backup-on-start to disable)",
    )
    parser.add_argument(
        "--backup-keep",
        type=int,
        default=int(os.environ.get("MCEE_BACKUP_KEEP", 20)),
        help="number of recent backups to keep (0 = keep all, default: 20)",
    )
    parser.add_argument(
        "--backup-notify",
        action="store_true",
        help="post a Discord message summarizing each backup result",
    )

    return parser.parse_args()


def load_state(path: Path) -> dict[str, dict[str, int]]:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    if not isinstance(data, dict):
        return {}
    return {
        str(key): {"inode": int(value["inode"]), "offset": int(value["offset"])}
        for key, value in data.items()
        if isinstance(value, dict) and "inode" in value and "offset" in value
    }


def save_state(path: Path, state: dict[str, dict[str, int]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(
        json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    tmp.replace(path)


def list_logs(log_dir: Path, pattern: str) -> list[Path]:
    if not log_dir.is_dir():
        raise FileNotFoundError(f"log directory not found: {log_dir}")
    return sorted(path for path in log_dir.glob(pattern) if path.is_file())


def read_new_text(path: Path, offset: int) -> tuple[str, int]:
    size = path.stat().st_size
    if size < offset:
        offset = 0
    if size == offset:
        return "", offset

    with path.open("rb") as file:
        file.seek(offset)
        raw = file.read()
        new_offset = file.tell()

    return raw.decode("utf-8", errors="replace"), new_offset


def chunks(text: str, limit: int) -> list[str]:
    result: list[str] = []
    remaining = text.strip()
    while remaining:
        if len(remaining) <= limit:
            result.append(remaining)
            break
        split_at = remaining.rfind("\n", 0, limit)
        if split_at < limit // 2:
            split_at = limit
        result.append(remaining[:split_at].rstrip())
        remaining = remaining[split_at:].lstrip()
    return result


def post_discord(webhook_url: str, content: str) -> None:
    payload = json.dumps({"content": content}, ensure_ascii=False).encode("utf-8")
    request = Request(
        webhook_url,
        data=payload,
        headers={"Content-Type": "application/json", "User-Agent": "mcee-chatlog/1.0"},
        method="POST",
    )
    with urlopen(request, timeout=30) as response:
        if response.status >= 300:
            raise RuntimeError(f"Discord webhook returned HTTP {response.status}")


def send_log_update(webhook_url: str, path: Path, text: str, is_test: bool = False) -> None:
    prefix_str = f"[TEST] {path.name}" if is_test else path.name
    prefix = f"**{prefix_str}**\n"
    for chunk in chunks(text, CONTENT_LIMIT):
        content = prefix + "```text\n" + chunk + "\n```"
        if len(content) > DISCORD_LIMIT:
            content = prefix + chunk[: DISCORD_LIMIT - len(prefix)]
        post_discord(webhook_url, content)


def poll_once(
    log_dir: Path,
    pattern: str,
    webhook_url: str,
    state: dict[str, dict[str, int]],
    initialize_at_eof: bool,
) -> None:
    for path in list_logs(log_dir, pattern):
        key = str(path.resolve())
        stat = path.stat()
        previous = state.get(key)

        if previous is None or previous["inode"] != stat.st_ino:
            offset = stat.st_size if initialize_at_eof else 0
            state[key] = {"inode": stat.st_ino, "offset": offset}
            previous = state[key]

        text, offset = read_new_text(path, previous["offset"])
        if text:
            send_log_update(webhook_url, path, text)
        state[key] = {"inode": stat.st_ino, "offset": offset}


# --- backup helpers ---------------------------------------------------

def prune_old_backups(backup_dir: Path, keep: int) -> None:
    """Delete oldest backups beyond `keep`. keep <= 0 disables pruning."""
    if keep <= 0:
        return
    backups = sorted(
        backup_dir.glob("mcee-*.tar"),
        key=lambda p: p.stat().st_mtime,
    )
    excess = len(backups) - keep
    for old in backups[:max(excess, 0)]:
        try:
            old.unlink()
        except OSError as exc:
            print(f"backup prune warning: could not delete {old}: {exc}", file=sys.stderr)


def run_backup(source_dir: Path, backup_dir: Path, keep: int) -> tuple[bool, str]:
    """Run tar backup of source_dir into backup_dir. Returns (success, message)."""
    if not source_dir.is_dir():
        return False, f"backup source not found: {source_dir}"

    backup_dir.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    dest = backup_dir / f"mcee-{timestamp}.tar"
    # No compression (plain tar), to keep backup CPU cost low and speed unaffected.
    # (kept as a plain list, no shell=True, to avoid shell-injection risk)
    cmd = ["tar", "cvf", str(dest), str(source_dir)]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=True,
        )
    except FileNotFoundError:
        return False, "tar command not found on this system"
    except subprocess.CalledProcessError as exc:
        # Clean up a partial/broken archive so it doesn't look like a good backup
        if dest.exists():
            try:
                dest.unlink()
            except OSError:
                pass
        stderr_tail = (exc.stderr or "").strip()[-500:]
        return False, f"tar exited with {exc.returncode}: {stderr_tail}"

    size_mb = dest.stat().st_size / (1024 * 1024)
    prune_old_backups(backup_dir, keep)
    del result  # unused; kept for clarity that tar ran to completion
    return True, f"backup complete: {dest.name} ({size_mb:.1f} MB)"


def prompt_for_webhook_url() -> str | None:
    """Interactively ask for the Discord webhook URL. Returns None if unavailable/cancelled."""
    if not sys.stdin.isatty():
        return None
    print("DISCORD_WEBHOOK_URL is not set.", file=sys.stderr)
    try:
        url = input("Discord webhook URL: ").strip()
    except (EOFError, KeyboardInterrupt):
        print()
        return None
    if not url:
        return None
    if not url.startswith("https://discord.com/api/webhooks/") and not url.startswith(
        "https://discordapp.com/api/webhooks/"
    ):
        print(
            "warning: URL doesn't look like a Discord webhook URL "
            "(expected https://discord.com/api/webhooks/...), using it anyway",
            file=sys.stderr,
        )
    return url


def main() -> int:
    args = parse_args()
    if not args.webhook_url:
        args.webhook_url = prompt_for_webhook_url()
    if not args.webhook_url:
        print(
            "DISCORD_WEBHOOK_URL or --webhook-url is required "
            "(no value provided and no interactive terminal to prompt on)",
            file=sys.stderr,
        )
        return 2
    if args.interval <= 0:
        print("--interval must be greater than 0", file=sys.stderr)
        return 2
    if not args.no_backup and args.backup_interval <= 0:
        print("--backup-interval must be greater than 0", file=sys.stderr)
        return 2

    log_dir = Path(args.log_dir)
    backup_source = Path(args.backup_source)
    backup_dir = Path(args.backup_dir)

    if not args.no_backup:
        try:
            backup_dir.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            print(f"error: could not create backup directory {backup_dir}: {exc}", file=sys.stderr)
            return 2

    if args.test:
        try:
            logs = list_logs(log_dir, args.glob)
            if not logs:
                print("No log files found for testing.", file=sys.stderr)
                return 1
            latest_path = logs[-1]
            try:
                with latest_path.open("r", encoding="utf-8", errors="replace") as f:
                    lines = f.readlines()
                last_lines = lines[-5:] if len(lines) >= 5 else lines
                text = "".join(last_lines) if last_lines else "(Empty file)"
            except OSError as exc:
                print(f"Failed to read log file for testing: {exc}", file=sys.stderr)
                return 1

            send_log_update(args.webhook_url, latest_path, text, is_test=True)
            print("Test message sent successfully.")
            return 0
        except Exception as exc:
            print(f"Test failed: {exc}", file=sys.stderr)
            return 1

    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)
    state_file = Path(args.state_file)
    state = load_state(state_file)
    first_poll = True

    # If --backup-on-start, force a backup on the very first loop iteration;
    # otherwise wait a full interval before the first backup.
    last_backup_time = 0.0 if args.backup_on_start else time.monotonic()

    while running:
        try:
            poll_once(
                log_dir=log_dir,
                pattern=args.glob,
                webhook_url=args.webhook_url,
                state=state,
                initialize_at_eof=first_poll and not args.send_existing,
            )
            save_state(state_file, state)
        except (FileNotFoundError, HTTPError, URLError, TimeoutError, RuntimeError) as exc:
            print(f"error: {exc}", file=sys.stderr)

        if not args.no_backup:
            now = time.monotonic()
            if now - last_backup_time >= args.backup_interval:
                ok, message = run_backup(backup_source, backup_dir, args.backup_keep)
                tag = "backup ok" if ok else "backup FAILED"
                print(f"{tag}: {message}")
                if args.backup_notify:
                    try:
                        post_discord(args.webhook_url, f"**Backup**: {message}")
                    except (HTTPError, URLError, TimeoutError, RuntimeError) as exc:
                        print(f"backup notify error: {exc}", file=sys.stderr)
                last_backup_time = now

        first_poll = False
        for _ in range(args.interval):
            if not running:
                break
            time.sleep(1)

    save_state(state_file, state)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())