MastodonへのAPI経由post

#!/usr/bin/env python3
"""Minecraft Java server's System chat -> Mastodon notifications.

Supports player join/leave, advancements (including goals/challenges), and
vanilla death messages loaded from the matching Minecraft en_us.json.

No external Python packages required. By default this script only prints
matching events. --send is required for actual posts.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import sys
import time
from dataclasses import dataclass
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen

# 検知するログの正規表現
LOG_PREFIX = re.compile(
    r"^\[\d{2}:\d{2}:\d{2}\] \[Server thread/INFO\]: System chat: (.+)$"
)
PLAYER = r"[A-Za-z0-9_]{3,16}"
JOIN_LEAVE = re.compile(rf"^(?P<player>{PLAYER}) (?P<action>joined|left) the game$")
ADVANCEMENT = re.compile(
    rf"^(?P<player>{PLAYER}) "
    r"(?P<action>has made the advancement|has completed the challenge|has reached the goal) "
    r"\[(?P<title>.+)\]$"
)
# Mojang's language placeholders: %1$s, %2$s, ..., %s, and literal %%.
PLACEHOLDER = re.compile(r"%%|%(?:(\d+)\$)?s")


@dataclass(frozen=True)
class Event:
    kind: str
    player: str
    text: str

# 死亡メッセージの定義を正規表現に変換
def compile_death_template(template: str) -> re.Pattern[str] | None:
    """Turn a player-first Mojang translation template into a full match.

    Captures %1$s as the player name; other placeholders may contain free text
    (mob names, customized items). Unknown or unsupported templates are ignored.
    """
    if not template.startswith("%1$s "):
        return None
    chunks: list[str] = []
    pos = 0
    unnumbered = 1
    player_seen = False
    for match in PLACEHOLDER.finditer(template):
        chunks.append(re.escape(template[pos:match.start()]))
        token = match.group(0)
        if token == "%%":
            chunks.append("%")
        else:
            index = int(match.group(1)) if match.group(1) else unnumbered
            if match.group(1) is None:
                unnumbered += 1
            if index == 1:
                if player_seen:
                    chunks.append(r"(?P=player)")
                else:
                    chunks.append(rf"(?P<player>{PLAYER})")
                    player_seen = True
            else:
                chunks.append(r".+?")
        pos = match.end()
    chunks.append(re.escape(template[pos:]))
    if not player_seen:
        return None
    return re.compile("^" + "".join(chunks) + "$")

# en_us.json から死亡パターンを収集
def load_death_patterns(lang_path: Path) -> list[re.Pattern[str]]:
    try:
        with lang_path.open("r", encoding="utf-8") as f:
            translations = json.load(f)
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"en_us.json を読み込めない: {lang_path}: {exc}") from exc
    if not isinstance(translations, dict):
        raise ValueError("en_us.json の形式が正しくない")

    patterns: list[re.Pattern[str]] = []
    for key, value in translations.items():
        if not (key.startswith("death.attack.") or key.startswith("death.fell.")):
            continue
        if not isinstance(value, str):
            continue
        pattern = compile_death_template(value)
        if pattern is not None:
            patterns.append(pattern)
    if not patterns:
        raise ValueError("死亡メッセージの定義が見つからなかった。en_us.json の内容を確認してほしい")
    return patterns

# ログ1行を解析してイベント判定
def parse_event(line: str, death_patterns: list[re.Pattern[str]]) -> tuple[Event | None, bool]:
    """Returns (event, is_system_chat); ignores all other log lines."""
    match = LOG_PREFIX.fullmatch(line.rstrip("\r\n"))
    if not match:
        return None, False
    message = match.group(1)

    match = JOIN_LEAVE.fullmatch(message)
    if match:
        player = match.group("player")
        kind = "login" if match.group("action") == "joined" else "logout"
        verb = "ログイン" if kind == "login" else "ログアウト"
        return Event(kind, player, f"【Minecraft】 {player} が{verb}しました。"), True

    match = ADVANCEMENT.fullmatch(message)
    if match:
        player = match.group("player")
        title = match.group("title")
        return Event("advancement", player,
                     f"【Minecraft】 進捗達成\n{player} が「{title}」を達成しました。"), True

    for pattern in death_patterns:
        match = pattern.fullmatch(message)
        if match:
            player = match.group("player")
            return Event("death", player, f"【Minecraft】 死亡\n{message}"), True

    return None, True

# ログファイルを継続して読み取る
def log_lines(path: Path, replay: bool):
    """Yield (line, stable_source_id) as bytes offsets.

    Normal mode tails *new* lines only and follows latest.log file rotation.
    Replay mode processes the existing file once for a no-posting test.
    """
    if replay:
        with path.open("rb") as f:
            inode = os.fstat(f.fileno()).st_ino
            while True:
                offset = f.tell()
                line = f.readline()
                if not line:
                    break
                yield line.decode("utf-8", errors="replace").rstrip("\r\n"), f"replay:{inode}:{offset}"
        return

    stream = None
    opened_inode = None
    attached_once = False
    try:
        while True:
            try:
                info = path.stat()
            except FileNotFoundError:
                time.sleep(1)
                continue

            file_id = (info.st_dev, info.st_ino)
            if stream is None or file_id != opened_inode:
                if stream is not None:
                    stream.close()
                stream = path.open("rb")
                opened_inode = file_id
                # First attach starts at EOF. After rotation, start at the
                # beginning of the replacement file.
                if not attached_once:
                    stream.seek(0, os.SEEK_END)
                    attached_once = True

            if info.st_size < stream.tell():
                # File was truncated in-place.
                stream.seek(0)
            position = stream.tell()
            line = stream.readline()
            if not line:
                time.sleep(0.5)
                continue
            if not line.endswith(b"\n"):
                stream.seek(position)
                time.sleep(0.5)
                continue
            yield line.decode("utf-8", errors="replace").rstrip("\r\n"), (
                f"{opened_inode[0]}:{opened_inode[1]}:{position}"
            )
    finally:
        if stream is not None:
            stream.close()

# Mastodon APIに投稿
def post_status(base_url: str, token: str, text: str, source_id: str):
    data = urlencode({"status": text, "visibility": "unlisted"}).encode("utf-8")
    # Stable for a given log record, so an intentional retry can use the same key.
    request_id = hashlib.sha256(source_id.encode("utf-8")).hexdigest()
    req = Request(
        base_url.rstrip("/") + "/api/v1/statuses",
        data=data,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
            "User-Agent": "minecraft-mastodon-bot/1.0",
            "Idempotency-Key": request_id,
        },
        method="POST",
    )
    with urlopen(req, timeout=15) as response:
        return json.load(response)["id"]

# 実行本体
def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--log", type=Path, default=Path.home() / "logs" / "latest.log")
    parser.add_argument("--lang", type=Path, default=Path(__file__).with_name("en_us.json"))
    parser.add_argument("--replay", action="store_true", help="既存ログを先頭から検証（投稿しない）")
    parser.add_argument("--send", action="store_true", help="Mastodonに実際に投稿（通常監視のみ）")
    parser.add_argument("--show-unmatched", action="store_true", help="未分類System chatを表示")
    args = parser.parse_args()
    if args.send and args.replay:
        parser.error("安全のため --replay と --send は同時に使えない")
    if not args.log.is_file():
        parser.error(f"ログファイルが見つからない: {args.log}")
    try:
        death_patterns = load_death_patterns(args.lang)
    except ValueError as exc:
        parser.error(str(exc))

    token = os.environ.get("MASTODON_ACCESS_TOKEN", "")
    if args.send and not token:
        parser.error("--send のときは MASTODON_ACCESS_TOKEN を設定してほしい")
    base_url = os.environ.get("MASTODON_BASE_URL", "https://mstdn.cigneg.com")
    if args.send and not base_url.startswith("https://"):
        parser.error("MASTODON_BASE_URL は https:// で始まる必要がある")

    print(f"死亡メッセージ {len(death_patterns)} パターン読み込み済み", file=sys.stderr, flush=True)
    print("MODE: POST" if args.send else "MODE: DRY RUN", file=sys.stderr, flush=True)

    for line, source_id in log_lines(args.log, args.replay):
        event, system_chat = parse_event(line, death_patterns)
        if event is None:
            if system_chat and args.show_unmatched:
                print(f"[UNMATCHED] {line}", flush=True)
            continue
        if not args.send:
            print(f"[{event.kind}] {event.text.replace(chr(10), ' / ')}", flush=True)
            continue
        try:
            status_id = post_status(base_url, token, event.text, source_id)
            print(f"[POSTED] {event.kind} {event.player} id={status_id}", flush=True)
        except HTTPError as exc:
            detail = exc.read(300).decode("utf-8", errors="replace")
            print(f"[ERROR] Mastodon HTTP {exc.code}: {detail}", file=sys.stderr, flush=True)
        except (URLError, TimeoutError, OSError, ValueError, KeyError) as exc:
            print(f"[ERROR] Mastodon投稿失敗: {exc}", file=sys.stderr, flush=True)
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\n停止しました", file=sys.stderr)

MASTODON_ACCESS_TOKEN={mc@mstdn.cigneg.comのアクセストークン}
MASTODON_BASE_URL=https://mstdn.cigneg.com

[Unit]
Description=Minecraft Mastodon Event Bot
Wants=network-online.target
After=network-online.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/mc-bot
EnvironmentFile=/etc/minecraft-mastodon-bot.env
ExecStart=/usr/bin/python3 /home/ubuntu/mc-bot/minecraft_mastodon_bot.py --send
Restart=on-failure
RestartSec=5
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target