144 lines
4.4 KiB
Python
144 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""IRC bot that watches monop-irc games and writes game state to JSON."""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import time
|
|
import irc.client
|
|
import irc.connection
|
|
from datetime import datetime, timezone
|
|
from parser import MonopParser
|
|
|
|
CONFIG_FILE = os.environ.get("MONOPBOT_CONFIG",
|
|
os.path.join(os.path.dirname(__file__), "config.json"))
|
|
|
|
|
|
def load_config():
|
|
with open(CONFIG_FILE) as f:
|
|
return json.load(f)
|
|
|
|
|
|
class MonopBot:
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.parser = MonopParser()
|
|
self.state_file = config.get("state_file", "../game-state.json")
|
|
if not os.path.isabs(self.state_file):
|
|
self.state_file = os.path.join(os.path.dirname(__file__), self.state_file)
|
|
self.game_nick = config.get("game_nick", "monop").lower()
|
|
self._last_write = 0
|
|
|
|
# Track player setup
|
|
self._awaiting_player_count = False
|
|
self._awaiting_player_name = False
|
|
self._expected_players = 0
|
|
self._setup_done = False
|
|
|
|
def write_state(self):
|
|
"""Write current game state to JSON file."""
|
|
now = time.time()
|
|
# Rate limit writes to every 0.5s
|
|
if now - self._last_write < 0.5:
|
|
return
|
|
self._last_write = now
|
|
|
|
state = self.parser.get_state()
|
|
state["lastUpdated"] = datetime.now(timezone.utc).isoformat()
|
|
|
|
tmp = self.state_file + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
os.replace(tmp, self.state_file)
|
|
|
|
def on_connect(self, connection, event):
|
|
channel = self.config["channel"]
|
|
print(f"Connected. Joining {channel}")
|
|
connection.join(channel)
|
|
|
|
def on_join(self, connection, event):
|
|
print(f"Joined {event.target}")
|
|
|
|
def on_pubmsg(self, connection, event):
|
|
"""Handle public messages in channel."""
|
|
nick = event.source.nick.lower() if event.source else ""
|
|
msg = event.arguments[0] if event.arguments else ""
|
|
|
|
# We care about messages from the game bot
|
|
if nick == self.game_nick or self._is_game_output(nick, msg):
|
|
self._process_game_line(msg)
|
|
|
|
def on_privmsg(self, connection, event):
|
|
"""Handle private messages (game might DM)."""
|
|
nick = event.source.nick.lower() if event.source else ""
|
|
msg = event.arguments[0] if event.arguments else ""
|
|
if nick == self.game_nick:
|
|
self._process_game_line(msg)
|
|
|
|
def _is_game_output(self, nick, msg):
|
|
"""Heuristic: detect if a message looks like monop output."""
|
|
game_patterns = [
|
|
r"roll is \d+, \d+",
|
|
r"That puts you on",
|
|
r"That would cost \$",
|
|
r"rent is \d+",
|
|
r"Name\s+Own\s+Price",
|
|
r"'s \(\d+\) holdings",
|
|
]
|
|
for pat in game_patterns:
|
|
if re.search(pat, msg):
|
|
return True
|
|
return False
|
|
|
|
def _process_game_line(self, line):
|
|
"""Process a line of game output."""
|
|
# Handle multi-line messages (some IRC clients split them)
|
|
for subline in line.split("\n"):
|
|
changed = self.parser.parse_line(subline)
|
|
if changed:
|
|
self.write_state()
|
|
|
|
def run(self):
|
|
config = self.config
|
|
reactor = irc.client.Reactor()
|
|
|
|
try:
|
|
connect_params = {}
|
|
if config.get("tls", False):
|
|
ssl_factory = irc.connection.Factory(wrapper=ssl.wrap_socket)
|
|
connect_params["connect_factory"] = ssl_factory
|
|
|
|
server = reactor.server()
|
|
server.connect(
|
|
config["server"],
|
|
config["port"],
|
|
config["nick"],
|
|
**connect_params,
|
|
)
|
|
except irc.client.ServerConnectionError as e:
|
|
print(f"Connection error: {e}")
|
|
sys.exit(1)
|
|
|
|
server.add_global_handler("welcome", self.on_connect)
|
|
server.add_global_handler("join", self.on_join)
|
|
server.add_global_handler("pubmsg", self.on_pubmsg)
|
|
server.add_global_handler("privmsg", self.on_privmsg)
|
|
|
|
print(f"Connecting to {config['server']}:{config['port']}...")
|
|
|
|
# Write initial empty state
|
|
self.write_state()
|
|
|
|
reactor.process_forever()
|
|
|
|
|
|
def main():
|
|
config = load_config()
|
|
bot = MonopBot(config)
|
|
bot.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|