weishaupt/basedball
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7e01b2d456d31da43db9e0d38be67354a555c265
basedball/bot.py
T

1489 lines
62 KiB
Python
Raw Blame History

import irc.bot
import time
import re
import json
import html
import os
import urllib.request
import urllib.error
import socket
from urllib.parse import quote
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import statsapi
from pybaseball import playerid_lookup
import traceback
import sys
import random
import threading
# --- CONFIGURATION ---
SERVER = "irc.rizon.net"
PORT = 6667
CHANNELS = ["#/g/tv", "#blog"]
# Channels where a pasted x.com / twitter.com status link gets announced as
# the first line of the tweet (via the keyless fxtwitter API). Lowercase.
TWEET_CHANNELS = {"#/g/tv", "#blog"}
TWEET_URL_RE = re.compile(r"https?://(?:www\.|mobile\.)?(?:twitter|x)\.com/(\w{1,15})/status(?:es)?/(\d+)", re.I)
TWEET_COOLDOWN = 30 # seconds before the same tweet is announced again in a channel
TWEET_SEEN = {}
# Running log of every announced tweet, pushed to bloggin.space for the wiki page.
TWEET_LOG_FILE = "tweet_log.json"
TWEET_LOG_MAX = 1000
TWEET_PUSH_URL = "https://bloggin.space/api/tweet_log_update.php"
TWEET_LOG_LOCK = threading.Lock()
NICKNAME = "basedball"
NICKSERV_PASS = os.environ.get("NICKSERV_PASS", "")
# Nicknames whose commands the bot should never respond to, no matter what they say --
# guards against bot-to-bot spam loops with other bots in these channels. GTV chat
# logging (see on_pubmsg) happens before this check, so an ignored nick's chat can
# still be logged without the bot ever executing its commands. Matching is case-insensitive.
IGNORED_NICKS = {"murasa"}
# Only these nicks can use sensitive commands like !quit. Matching is case-insensitive.
ADMIN_NICKS = {"tpuahsiew", "weishaupt"}
# --- CLOD (Claude) INTEGRATION ---
# Same proxy the bloggin.space/Clod web widget talks to. Restricted to channel
# ops at runtime (see is_op below), not a static nick list, since op status
# can change without editing/restarting the bot.
CLOD_PROXY_URL = "https://bloggin.space/Clod/claude-proxy.php"
# --- #/g/tv CHAT LOGGER ---
# Logs the latest human (non-command) lines said in GTV_LOG_CHANNEL, persists them to
# disk so the log survives a bot restart, and pushes them to bloggin.space for display.
GTV_LOG_CHANNEL = "#/g/tv"
GTV_LOG_FILE = "gtv_log.json"
GTV_MAX_LINES = 20
GTV_PUSH_URL = "https://bloggin.space/api/gtv_log_update.php"
GTV_ADMIN_KEY = os.environ.get("GTV_ADMIN_KEY", "")
# --- COPS/BODYCAM YOUTUBE TRACKING ---
# Channels checked live, on-demand, by the !cops command. Uses YouTube's public RSS feed
# per channel -- no API key needed.
YOUTUBE_CHANNELS = {
"Sergeant Curtis": "UCPlH2h2Ckq1T-DKqw0latYA",
"PoliceActivity": "UCXMYxKMh3prxnM_4kYZuB3g",
"Code Blue Cam": "UCCKkuXux09y-TCg-BQxCjNA",
"Law&Crime BodyCam": "UCMV3pzWIhJGLYzoHyxBjjNw",
"Body Cam Watch": "UCQFvmd1tcYe4ZIw-Bqf3aLw",
"Donut Operator": "UCwkm_Wcyh0pc7UUmZZfL-6w",
}
# ==================== IRC COLOR HELPERS ====================
def color_text(text, fg, bg=None):
if bg is None:
if fg == 0:
return text
return f"\x03{fg:02d}{text}\x03"
return f"\x03{fg:02d},{bg:02d}{text}\x03"
YOUTUBE_TAG = color_text("[YouTube]", 4, 0) # red text on white background
# ==================== GTV CHAT LOGGER ====================
# Standard mIRC 16-color palette mapped to hex, for rendering logged chat lines as HTML.
# Separate from TEAM_COLORS above, which is the bot's own numeric-index send-side palette --
# this one maps the same numeric codes to actual hex values for browser display.
MIRC_HTML_COLORS = {
"00": "#FFFFFF", "01": "#000000", "02": "#00007F", "03": "#009300",
"04": "#FF0000", "05": "#7F0000", "06": "#9C009C", "07": "#FC7F00",
"08": "#FFFF00", "09": "#00FC00", "10": "#009393", "11": "#00FFFF",
"12": "#0000FC", "13": "#FF00FF", "14": "#7F7F7F", "15": "#D2D2D2",
}
def mirc_to_html(text):
"""Convert a raw IRC message with mIRC control codes into safe HTML with inline
color/bold/italic/underline spans. HTML-escapes the text first so chat content
can't inject markup."""
text = html.escape(text)
result = []
bold = italic = underline = False
fg = bg = None
span_open = False
i, n = 0, len(text)
def open_span():
styles = []
if fg:
styles.append(f"color:{fg}")
if bg:
styles.append(f"background-color:{bg}")
if bold:
styles.append("font-weight:bold")
if italic:
styles.append("font-style:italic")
if underline:
styles.append("text-decoration:underline")
if styles:
result.append(f'<span style="{";".join(styles)}">')
return True
return False
while i < n:
ch = text[i]
if ch == "\x03": # color code, optionally NN[,MM]
if span_open:
result.append("</span>")
span_open = False
i += 1
m = re.match(r"(\d{1,2})?(,(\d{1,2}))?", text[i:])
if m and m.group(0):
num, _, bgnum = m.groups()
fg = MIRC_HTML_COLORS.get(num.zfill(2)) if num else None
bg = MIRC_HTML_COLORS.get(bgnum.zfill(2)) if bgnum else bg
i += len(m.group(0))
else:
fg = bg = None
span_open = open_span()
continue
if ch == "\x02": # bold
if span_open:
result.append("</span>")
bold = not bold
span_open = open_span()
i += 1
continue
if ch == "\x1D": # italic
if span_open:
result.append("</span>")
italic = not italic
span_open = open_span()
i += 1
continue
if ch == "\x1F": # underline
if span_open:
result.append("</span>")
underline = not underline
span_open = open_span()
i += 1
continue
if ch == "\x0F": # reset all
if span_open:
result.append("</span>")
span_open = False
bold = italic = underline = False
fg = bg = None
i += 1
continue
if ch == "\x16": # reverse (swap fg/bg)
if span_open:
result.append("</span>")
fg, bg = bg, fg
span_open = open_span()
i += 1
continue
result.append(ch)
i += 1
if span_open:
result.append("</span>")
return "".join(result)
# Matches bare http(s) URLs. Runs after mirc_to_html() has already HTML-escaped the
# message, so there's no raw < > & left in the text to worry about breaking.
GTV_URL_PATTERN = re.compile(r'(https?://[^\s<]+)')
def linkify(html_text):
"""Wrap bare URLs in the already-mIRC-converted HTML with clickable <a> tags."""
def _sub(m):
url = m.group(1)
return f'<a href="{url}" target="_blank" rel="noopener">{url}</a>'
return GTV_URL_PATTERN.sub(_sub, html_text)
def load_gtv_log():
try:
with open(GTV_LOG_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return []
def save_gtv_log(log):
"""Write via a temp file + rename so a crash mid-write can't corrupt the log."""
tmp_path = GTV_LOG_FILE + ".tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(log, f)
os.replace(tmp_path, GTV_LOG_FILE)
def push_gtv_log(log):
"""Push the latest log to bloggin.space. Runs on the calling thread's behalf --
callers should invoke this via run_async so a slow/hanging request never blocks
the main IRC loop (same reasoning as every other external call in this bot)."""
try:
payload = json.dumps({"admin_key": GTV_ADMIN_KEY, "log": log}).encode("utf-8")
req = urllib.request.Request(
GTV_PUSH_URL,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=10) as resp:
resp.read()
except Exception as e:
print(f"Failed to push gtv log: {e}")
def log_gtv_message(nick, raw_message):
"""Append a human chat line to the persisted GTV log, trim to the latest
GTV_MAX_LINES, save to disk, and push to bloggin.space."""
log = load_gtv_log()
log.append({
"nick": html.escape(nick),
"message_html": linkify(mirc_to_html(raw_message)),
"timestamp": int(time.time()),
})
log = log[-GTV_MAX_LINES:]
save_gtv_log(log)
push_gtv_log(log)
def log_tweet(tweet_id, author, text, nick):
"""Add an announced tweet to the running tweet log (tweet_log.json), skip
tweets that are already logged, then push the whole log to bloggin.space.
Holds a lock so two announcements at once can't clobber each other's write.
Called from the tweet handler's own background thread."""
url = f"https://x.com/{author}/status/{tweet_id}"
try:
with TWEET_LOG_LOCK:
try:
with open(TWEET_LOG_FILE, "r", encoding="utf-8") as f:
log = json.load(f)
if not isinstance(log, list):
log = []
except (FileNotFoundError, ValueError):
log = []
if any(str(e.get("id")) == str(tweet_id) for e in log):
return
log.append({
"id": str(tweet_id),
"url": url,
"author": author,
"text": text,
"nick": nick,
"timestamp": int(time.time()),
})
log = log[-TWEET_LOG_MAX:]
tmp_path = TWEET_LOG_FILE + ".tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(log, f)
os.replace(tmp_path, TWEET_LOG_FILE)
payload = json.dumps({"admin_key": GTV_ADMIN_KEY, "log": log}).encode("utf-8")
req = urllib.request.Request(
TWEET_PUSH_URL,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=10) as resp:
resp.read()
except Exception as e:
print(f"Failed to log tweet: {e}")
def strip_markdown_for_irc(text):
"""Convert Clod's markdown-style formatting into plain IRC-friendly text --
markdown links and bold don't render in an IRC client, so flatten them."""
text = re.sub(r'\[([^\]]+)\]\((https?://[^\s)]+)\)', r'\1 (\2)', text) # [label](url) -> label (url)
text = re.sub(r'\*\*([^*]+)\*\*', r'\1', text) # **bold** -> bold
return text
# mIRC standard palette used below:
# 0 white 1 black 2 navy 3 green 4 red 5 brown 6 purple
# 7 orange 8 yellow 9 lt.green 10 teal 11 cyan 12 royal blue 13 pink 14 grey 15 lt.grey
TEAM_COLORS = {
# team: (foreground, background)
"yankees": (0, 2), # white on navy
"red sox": (4, 2), # red on navy
"blue jays": (0, 12), # white on royal blue
"orioles": (7, 1), # orange on black
"rays": (11, 2), # light cyan on navy
"twins": (4, 2), # red on navy
"tigers": (7, 2), # orange on navy
"royals": (8, 12), # gold on royal blue
"astros": (7, 2), # orange on navy
"guardians": (4, 2), # red on navy
"indians": (4, 2), # red on navy
"mariners": (10, 2), # teal on navy
"angels": (0, 4), # white on red
"rangers": (4, 2), # red on navy
"white sox": (0, 1), # white on black
"athletics": (8, 3), # gold on green
"phillies": (0, 4), # white on red
"braves": (4, 2), # red on navy
"mets": (7, 12), # orange on royal blue
"nationals": (0, 4), # white on red
"marlins": (7, 10), # orange on teal
"cubs": (4, 12), # red on royal blue
"cardinals": (0, 4), # white on red
"pirates": (8, 1), # gold on black
"brewers": (8, 2), # yellow on navy
"reds": (4, 0), # red on white
"dodgers": (12, 0), # blue on white
"padres": (8, 5), # gold on brown
"giants": (7, 1), # orange on black
"diamondbacks": (4, 1), # red on black
"rockies": (6, 1), # purple on black
}
# Team name/nickname substitutions applied everywhere a team name shows up in output.
# Order matters: more specific (full-name) patterns must come before the plain nickname
# pattern they'd otherwise get caught by, so e.g. "Los Angeles Dodgers" doesn't turn into
# the redundant "Los Angeles Los Doyers".
RENAME_RULES = [
(r"Cleveland Guardians", "Cleveland Indians"),
(r"\bGuardians\b", "Indians"),
(r"Los Angeles Dodgers", "Los Doyers"),
(r"\bDodgers\b", "Los Doyers"),
(r"\bBraves\b", "Barves"),
(r"\bRangers\b", "Ragners"),
(r"\bAstros\b", "Lastros"),
(r"\bCubs\b", "Cusb"),
(r"\bRoyals\b", "Roils"),
(r"\bAngels\b", "Angles"),
(r"\bBrewers\b", "Viewers"),
(r"\bTwins\b", "Twinks"),
(r"\bMets\b", "Mest"),
(r"\bGiants\b", "Gints"),
]
def fix_team_name(name):
"""Apply every rename rule in RENAME_RULES, wherever the matched text appears in the
string -- not just when it's the entire string."""
if not name:
return name
for pattern, replacement in RENAME_RULES:
name = re.sub(pattern, replacement, name, flags=re.IGNORECASE)
return name
def get_team_color(team_name):
if not team_name:
return (0, None)
name_lower = team_name.lower()
for key, colors in TEAM_COLORS.items():
if key in name_lower or name_lower in key:
return colors
return (0, None)
def colorize_team(name):
"""Look up the team's color using its real name, then apply the nickname rename for
display -- order matters, since TEAM_COLORS is keyed by the real nickname (e.g. 'brewers'),
not the renamed one ('viewers')."""
fg, bg = get_team_color(name)
renamed = fix_team_name(name)
return color_text(renamed, fg, bg)
# Full official names, used to recognize team names inside raw statsapi text
# (boxscore/highlights/leaders output) that we don't otherwise control the format of.
TEAM_FULL_NAMES = {
"yankees": "New York Yankees",
"red sox": "Boston Red Sox",
"blue jays": "Toronto Blue Jays",
"orioles": "Baltimore Orioles",
"rays": "Tampa Bay Rays",
"twins": "Minnesota Twins",
"tigers": "Detroit Tigers",
"royals": "Kansas City Royals",
"astros": "Houston Astros",
"guardians": "Cleveland Guardians",
"indians": "Cleveland Indians",
"mariners": "Seattle Mariners",
"angels": "Los Angeles Angels",
"rangers": "Texas Rangers",
"white sox": "Chicago White Sox",
"athletics": "Athletics",
"phillies": "Philadelphia Phillies",
"braves": "Atlanta Braves",
"mets": "New York Mets",
"nationals": "Washington Nationals",
"marlins": "Miami Marlins",
"cubs": "Chicago Cubs",
"cardinals": "St. Louis Cardinals",
"pirates": "Pittsburgh Pirates",
"brewers": "Milwaukee Brewers",
"reds": "Cincinnati Reds",
"dodgers": "Los Angeles Dodgers",
"padres": "San Diego Padres",
"giants": "San Francisco Giants",
"diamondbacks": "Arizona Diamondbacks",
"rockies": "Colorado Rockies",
}
# Map every recognizable string (nickname or full name) -> its TEAM_COLORS key,
# longest first so "Boston Red Sox" matches before a bare "Red Sox"/"Sox" fragment would.
_CANDIDATE_TO_KEY = {}
for _key in TEAM_COLORS.keys():
_CANDIDATE_TO_KEY[_key.lower()] = _key
for _key, _full in TEAM_FULL_NAMES.items():
_CANDIDATE_TO_KEY[_full.lower()] = _key
_SORTED_CANDIDATES = sorted(_CANDIDATE_TO_KEY.keys(), key=len, reverse=True)
_TEAM_NAME_PATTERN = re.compile(
"(" + "|".join(re.escape(c) for c in _SORTED_CANDIDATES) + ")",
re.IGNORECASE
)
def colorize_team_names_in_text(text):
"""Find any team name/nickname inside a block of raw text and wrap it in that team's colors.
Used for statsapi's own pre-formatted output (boxscore, highlights, leaders) where we don't
control the exact wording or field layout."""
if not text:
return text
def _sub(m):
matched = m.group(0)
key = _CANDIDATE_TO_KEY.get(matched.lower())
if not key:
return matched
fg, bg = TEAM_COLORS.get(key, (0, None))
renamed = fix_team_name(matched)
return color_text(renamed, fg, bg)
return _TEAM_NAME_PATTERN.sub(_sub, text)
def get_team_id(search_name):
"""Look up an MLB team's numeric ID from a partial name (e.g. 'brewers', 'reds')."""
try:
results = statsapi.lookup_team(search_name)
if results:
return results[0]['id']
except:
pass
return None
def get_player_bio(pid):
"""Look up a player's current position/number/team via statsapi, using their MLBAM id.
Returns None on any failure so callers can gracefully fall back to stats-only output.
Retired/inactive players are detected via the 'active' flag and shown as 'Retired'
rather than guessing at a stale team."""
try:
data = statsapi.get('person', {'personId': pid, 'hydrate': 'currentTeam'})
people = data.get('people', [])
if not people:
return None
person = people[0]
active = bool(person.get('active', False))
position = person.get('primaryPosition', {}).get('abbreviation', '')
number = person.get('primaryNumber')
if active:
team_raw = person.get('currentTeam', {}).get('name')
team_display = colorize_team(team_raw) if team_raw else "Active"
else:
team_display = "Retired"
return {
'position': position,
'number': number,
'team_display': team_display,
'active': active,
}
except:
return None
def format_bio_header(name, bio):
"""Build 'First Last (#NN POS) - Team' when bio info is available, otherwise just the name."""
if not bio:
return name
bits = []
if bio.get('number'):
bits.append(f"#{bio['number']}")
if bio.get('position'):
bits.append(bio['position'])
bio_str = " ".join(bits)
header = f"{name} ({bio_str})" if bio_str else name
return f"{header} - {bio['team_display']}"
class BaseballBot(irc.bot.SingleServerIRCBot):
def __init__(self, channels, nickname, server, port):
super().__init__([(server, port)], nickname, nickname)
self.channels_list = channels
self.last_command_time = 0
self.cooldown = 3
self.current_year = datetime.now().year
# Wrap the connection's privmsg so basedball's own replies in the GTV channel
# get logged too, not just what other people say. IRC servers don't echo a
# client's own outgoing messages back through on_pubmsg by default, so this is
# a send-side hook rather than a receive-side one -- it fires regardless of
# whether a handler calls c.privmsg(...) directly or goes through send_line().
self._real_privmsg = self.connection.privmsg
self.connection.privmsg = self._logging_privmsg
threading.Thread(target=self.console_input, daemon=True).start()
def _logging_privmsg(self, target, text):
self._real_privmsg(target, text)
if target == GTV_LOG_CHANNEL:
self.run_async(log_gtv_message, self.connection.get_nickname() or NICKNAME, text)
def console_input(self):
target = self.channels_list[0]
while True:
try:
message = input()
if message.strip():
self.connection.privmsg(target, message.strip())
except:
break
def fetch_latest_youtube_video(self, channel_id):
"""Return (video_id, title, url) for a channel's most recent upload, or None on failure.
Used on-demand by !cops."""
try:
feed_url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
req = urllib.request.Request(feed_url, headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept": "application/atom+xml,application/xml,text/xml,*/*;q=0.9",
"Accept-Language": "en-US,en;q=0.9",
})
with urllib.request.urlopen(req, timeout=15) as resp:
raw = resp.read()
root = ET.fromstring(raw)
ns = {
"atom": "http://www.w3.org/2005/Atom",
"yt": "http://www.youtube.com/xml/schemas/2015",
}
entry = root.find("atom:entry", ns)
if entry is None:
return None
video_id_el = entry.find("yt:videoId", ns)
title_el = entry.find("atom:title", ns)
video_id = video_id_el.text if video_id_el is not None else None
title = title_el.text if title_el is not None else "New video"
if not video_id:
return None
url = f"https://youtu.be/{video_id}"
return (video_id, title, url)
except Exception as e:
print(f"YouTube feed fetch failed for channel {channel_id}: {e}")
return None
def handle_tweet_link(self, c, target, user, tweet_id, nick=""):
"""Announce the first line of a pasted tweet link. Runs in a background
thread via run_async so a slow request never stalls the IRC loop."""
try:
key = (target.lower(), tweet_id)
now = time.time()
if now - TWEET_SEEN.get(key, 0) < TWEET_COOLDOWN:
return
TWEET_SEEN[key] = now
url = f"https://api.fxtwitter.com/{user}/status/{tweet_id}"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read().decode("utf-8"))
tweet = data.get("tweet") or {}
text = (tweet.get("text") or "").strip()
author = (tweet.get("author") or {}).get("screen_name") or user
first = ""
for line in text.splitlines():
if line.strip():
first = line.strip()
break
first = re.sub(r"[\x00-\x1f]", "", html.unescape(first))
if not first:
return
if len(first) > 300:
first = first[:297] + "..."
self.send_line(c, target, f"\x0304[Tweet]\x0f \x02\x0382@{author}\x0f\x0382: {first}\x0f", delay=0)
log_tweet(tweet_id, author, first, nick)
except Exception:
traceback.print_exc()
def send_line(self, c, target, text, delay=0.5):
"""Send a single IRC line, then pause briefly to avoid flooding the server."""
c.privmsg(target, text)
time.sleep(delay)
def run_async(self, func, *args):
"""Run a command handler in its own background thread instead of the main IRC event
loop. Every command that hits an external API (statsapi, Wikipedia, etc.) goes through
this, so a slow or hanging request can't freeze the whole bot -- including replying to
the server's keepalive PINGs, which would otherwise risk a timeout/disconnect. Sending
replies from a background thread is already the same pattern the YouTube watcher thread
uses successfully."""
threading.Thread(target=func, args=args, daemon=True).start()
def is_rate_limited(self):
now = time.time()
if now - self.last_command_time < self.cooldown:
return True
self.last_command_time = now
return False
def is_op(self, target, nick):
"""Check live channel op status via the IRC library's own channel tracking,
rather than a static admin list -- reflects real +o status at the moment
the command is used."""
try:
channel = self.channels.get(target)
if channel is None:
return False
return channel.is_oper(nick)
except Exception:
return False
def on_welcome(self, c, e):
c.privmsg("NickServ", f"IDENTIFY {NICKSERV_PASS}")
time.sleep(2)
for channel in self.channels_list:
c.join(channel)
time.sleep(1)
def on_pubmsg(self, c, e):
sender_nick = e.source.nick if e.source else ""
msg = e.arguments[0].strip()
# Log human/other-bot (non-command) chat in the GTV channel first, before the
# IGNORED_NICKS gate below. This intentionally runs even for nicks in
# IGNORED_NICKS (e.g. macbot) so their chat still shows up in the log, while
# the gate right after still stops the bot from ever executing their commands.
# Runs in its own thread so a slow push to bloggin.space never stalls the main
# IRC loop.
if e.target == GTV_LOG_CHANNEL and not msg.startswith("!"):
self.run_async(log_gtv_message, sender_nick, msg)
if sender_nick.lower() in IGNORED_NICKS:
return
if e.target.lower() in TWEET_CHANNELS and not msg.startswith("!"):
tm = TWEET_URL_RE.search(msg)
if tm:
self.run_async(self.handle_tweet_link, c, e.target, tm.group(1), tm.group(2), sender_nick)
if not msg.startswith("!"):
return
if self.is_rate_limited():
return
target = e.target
cmd = msg.lower()
if cmd == "!quit":
if sender_nick.lower() in ADMIN_NICKS:
self.disconnect()
return
elif cmd in ["!mlb", "!mlbhelp"]:
self.send_help(c, target)
elif cmd.startswith("!score") or cmd == "!yesterday":
self.run_async(self.handle_scores, c, target, msg)
elif cmd == "!final":
self.run_async(self.handle_final, c, target)
elif cmd == "!live":
self.run_async(self.handle_live, c, target)
elif cmd == "!tomorrow":
self.run_async(self.handle_tomorrow, c, target)
elif cmd == "!probable":
self.run_async(self.handle_probable, c, target)
elif cmd == "!random":
self.run_async(self.handle_random, c, target)
elif cmd.startswith("!stats "):
self.run_async(self.handle_stats, c, target, msg)
elif cmd.startswith("!game "):
self.run_async(self.handle_game, c, target, msg)
elif cmd.startswith("!stand"):
self.run_async(self.handle_standings, c, target, msg)
elif cmd.startswith("!boxscore"):
self.run_async(self.handle_boxscore, c, target, msg)
elif cmd.startswith("!highlights"):
self.run_async(self.handle_highlights, c, target, msg)
elif cmd.startswith("!leaders"):
self.run_async(self.handle_leaders, c, target, msg)
elif cmd.startswith("!roster"):
self.run_async(self.handle_roster, c, target, msg)
elif cmd.startswith("!award"):
self.run_async(self.handle_award, c, target, msg)
elif cmd == "!cops":
self.run_async(self.handle_cops, c, target)
elif cmd.startswith("!wiki "):
self.run_async(self.handle_wiki, c, target, msg)
elif cmd.startswith("!clod"):
if not self.is_op(target, sender_nick):
c.privmsg(target, "Ops only, chief.")
return
self.run_async(self.handle_clod, c, target, msg)
elif cmd.startswith("!mow"):
self.run_async(self.handle_mow, c, target, msg)
def send_help(self, c, target):
c.privmsg(target, "MLB: !score/!live/!yesterday/!final (scores), !tomorrow (schedule), !probable (pitchers), !game [team], !stand [al/nl] [div], !stats [f] [l] (hitting+pitching, season+career), !boxscore [team] (line score), !highlights [team] (video), !leaders [hr/avg/rbi/sb/obp/slg/ops/era/so/wins/saves/whip], !roster [team], !award [mvp/cyyoung/roy/hof] [year] [al/nl], !random | Other: !cops (latest bodycam channel uploads), !wiki [term] (Wikipedia summary), !clod [message] (ops only - talks to Clod/Claude), !mow [zipcode] (lawn mowing forecast)")
def handle_cops(self, c, target):
entries = []
for name, channel_id in YOUTUBE_CHANNELS.items():
result = self.fetch_latest_youtube_video(channel_id)
if result:
_, title, url = result
entries.append(f"{name}: {title} - {url}")
else:
entries.append(f"{name}: (unavailable)")
# 2 channels per line -- with 6 tracked channels that's exactly 3 lines.
for i in range(0, len(entries), 2):
pair = entries[i:i + 2]
line = f"{YOUTUBE_TAG} " + " | ".join(pair)
self.send_line(c, target, line)
def handle_wiki(self, c, target, msg):
try:
parts = msg.split(maxsplit=1)
term = parts[1].strip() if len(parts) > 1 else ""
if not term:
c.privmsg(target, "Usage: !wiki [search term]")
return
# Step 1: resolve whatever the user typed to the closest matching article title.
search_url = f"https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={quote(term)}&format=json&srlimit=1"
req = urllib.request.Request(search_url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=10) as resp:
search_data = json.loads(resp.read())
results = search_data.get("query", {}).get("search", [])
if not results:
c.privmsg(target, f"No Wikipedia article found for '{term}'.")
return
title = results[0]["title"]
# Step 2: pull a clean summary/extract for that exact title.
summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{quote(title.replace(' ', '_'))}"
req2 = urllib.request.Request(summary_url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req2, timeout=10) as resp2:
summary_data = json.loads(resp2.read())
page_title = summary_data.get("title", title)
extract = summary_data.get("extract", "")
page_url = summary_data.get("content_urls", {}).get("desktop", {}).get("page")
if not page_url:
page_url = f"https://en.wikipedia.org/wiki/{quote(title.replace(' ', '_'))}"
if summary_data.get("type") == "disambiguation":
c.privmsg(target, f"'{term}' is ambiguous on Wikipedia -- see: {page_url}")
return
if not extract:
c.privmsg(target, f"{page_title} - {page_url}")
return
if len(extract) > 400:
extract = extract[:400].rsplit(" ", 1)[0] + "..."
self.send_line(c, target, f"{page_title} - {page_url}")
self.send_line(c, target, extract)
except Exception as e:
print(f"Wiki lookup error for '{msg}': {e}")
c.privmsg(target, "Error retrieving Wikipedia article.")
# Lawn-mowing conditions forecast, in the style of mowcast.io.
# Data: Zippopotam.us (zip -> lat/lon) + api.weather.gov (NWS hourly forecast). No API key needed.
def handle_mow(self, c, target, msg):
try:
parts = msg.split(maxsplit=1)
zipcode = parts[1].strip() if len(parts) > 1 else ""
if not re.fullmatch(r"\d{5}", zipcode):
c.privmsg(target, "Usage: !mow [5-digit US zip code]")
return
geo_url = f"https://api.zippopotam.us/us/{zipcode}"
req = urllib.request.Request(geo_url, headers={"User-Agent": "Mozilla/5.0"})
try:
with urllib.request.urlopen(req, timeout=10) as resp:
geo_data = json.loads(resp.read())
except urllib.error.HTTPError as e:
if e.code == 404:
c.privmsg(target, f"Zip code '{zipcode}' not found.")
return
raise
place = geo_data["places"][0]
lat = place["latitude"]
lon = place["longitude"]
place_name = f'{place["place name"]}, {place["state abbreviation"]}'
headers = {"User-Agent": "Basedball IRC Bot, bloggin.space"}
points_url = f"https://api.weather.gov/points/{lat},{lon}"
req = urllib.request.Request(points_url, headers=headers)
with urllib.request.urlopen(req, timeout=10) as resp:
points_data = json.loads(resp.read())
hourly_url = points_data["properties"]["forecastHourly"]
req = urllib.request.Request(hourly_url, headers=headers)
with urllib.request.urlopen(req, timeout=10) as resp:
hourly_data = json.loads(resp.read())
periods = hourly_data["properties"]["periods"][:12]
if not periods:
c.privmsg(target, "No forecast data available for that zip code.")
return
now = periods[0]
temp = now.get("temperature", 70)
wind_now_match = re.search(r"\d+", now.get("windSpeed", "0"))
wind_now = int(wind_now_match.group()) if wind_now_match else 0
humidity = (now.get("relativeHumidity") or {}).get("value") or 50
precip_now = (now.get("probabilityOfPrecipitation") or {}).get("value") or 0
wind_later = wind_now
precip_later = precip_now
for p in periods[1:6]:
m = re.search(r"\d+", p.get("windSpeed", "0"))
if m:
wind_later = max(wind_later, int(m.group()))
pr = (p.get("probabilityOfPrecipitation") or {}).get("value") or 0
precip_later = max(precip_later, pr)
score = 100
score -= precip_now * 0.8
score -= max(0, wind_now - 8) * 2.5
if temp > 90 or temp < 45:
score -= 20
elif temp > 85 or temp < 55:
score -= 8
if humidity > 85:
score -= 10
score = max(0, min(100, round(score)))
if score >= 70:
desc = "Excellent conditions for mowing"
elif score >= 50:
desc = "Good conditions for mowing"
elif score >= 30:
desc = "Fair conditions for mowing"
else:
desc = "Poor conditions for mowing"
notes = []
if precip_later > precip_now + 20:
notes.append("rain is likely later so get to it early")
elif wind_later > wind_now + 5:
notes.append("especially in the morning before winds pick up")
if humidity > 85:
notes.append("grass may be wet from humidity/dew")
if score < 30:
notes.append("consider waiting for better conditions")
note = (", " + "; ".join(notes)) if notes else ""
line = f" [Mowcast] MowScore {score}% {desc} today{note}. ({place_name}) "
c.privmsg(target, color_text(line, 3, 0))
except Exception as e:
print(f"Mow lookup error for '{msg}': {e}")
c.privmsg(target, "Error retrieving mowing forecast for that zip code.")
def handle_clod(self, c, target, msg):
"""Talk to Clod (Claude) via the same proxy the bloggin.space/Clod web
widget uses -- gets the same board read/write and wiki read tools for free."""
try:
question = msg[len("!clod"):].strip()
if not question:
c.privmsg(target, "Usage: !clod [message] (ops only)")
return
payload = json.dumps({"message": question}).encode("utf-8")
req = urllib.request.Request(
CLOD_PROXY_URL,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=130) as resp:
raw = resp.read()
data = json.loads(raw)
if "error" in data:
c.privmsg(target, f"Clod error: {data.get('detail') or data['error']}")
return
reply = strip_markdown_for_irc(data.get("reply", "")).strip()
if not reply:
c.privmsg(target, "Clod didn't have anything to say.")
return
# IRC lines are capped around 512 bytes incl. protocol overhead --
# wrap each paragraph into safe-length chunks, same approach as !roster.
# Cap the whole reply at 4 lines so Clod can't flood the channel.
max_len = 420
max_lines = 4
all_chunks = []
for line in reply.split("\n"):
line = line.strip()
if not line:
continue
words = line.split(" ")
chunk = ""
for w in words:
piece = w if not chunk else " " + w
if len(chunk) + len(piece) > max_len:
all_chunks.append(chunk)
chunk = w
else:
chunk += piece
if chunk:
all_chunks.append(chunk)
truncated = len(all_chunks) > max_lines
to_send = all_chunks[:max_lines]
if truncated and to_send:
to_send[-1] = to_send[-1].rstrip() + " [...]"
for i, chunk in enumerate(to_send):
if i == len(to_send) - 1:
c.privmsg(target, chunk)
else:
self.send_line(c, target, chunk)
except urllib.error.HTTPError as e:
try:
body = e.read().decode("utf-8", errors="replace")
data = json.loads(body)
detail = data.get("detail") or data.get("error") or body
except Exception:
detail = f"HTTP {e.code}: {e.reason}"
print(f"Clod command HTTP error {e.code}: {detail}")
c.privmsg(target, f"Clod error ({e.code}): {str(detail)[:400]}")
except urllib.error.URLError as e:
if isinstance(getattr(e, "reason", None), socket.timeout) or "timed out" in str(e).lower():
print(f"Clod command timed out: {e}")
c.privmsg(target, "Clod's taking too long to answer that one (timed out). Try something simpler or ask again.")
else:
print(f"Clod command network error: {e}")
c.privmsg(target, f"Couldn't reach Clod: {e}")
except Exception as e:
print(f"Clod command error: {e}")
c.privmsg(target, f"Error reaching Clod: {e}")
def handle_scores(self, c, target, msg):
try:
et = ZoneInfo("America/New_York")
if "!yesterday" in msg or "!final" in msg:
date = (datetime.now(et) - timedelta(days=1)).strftime("%Y-%m-%d")
else:
date = datetime.now(et).strftime("%Y-%m-%d")
games = statsapi.schedule(date=date, sportId=1)
self.send_line(c, target, f"--- Scores for {date} ---")
for g in games:
away = colorize_team(g.get('away_name'))
home = colorize_team(g.get('home_name'))
self.send_line(c, target, f"[{g.get('status')}] {away} {g.get('away_score')} @ {home} {g.get('home_score')}")
except:
c.privmsg(target, "Error.")
def handle_final(self, c, target):
self.handle_scores(c, target, "!final")
def handle_live(self, c, target):
try:
et = ZoneInfo("America/New_York")
games = statsapi.schedule(date=datetime.now(et).strftime("%Y-%m-%d"), sportId=1)
def is_live(status):
if not status:
return False
s = status.lower()
# Explicitly not live, checked first so these can't accidentally match a
# live keyword below (e.g. "Final" would otherwise be fine, but better safe).
if any(k in s for k in ("final", "scheduled", "pre-game", "postponed", "cancelled", "canceled")):
return False
return any(k in s for k in ("progress", "delay", "warmup", "suspend", "challenge"))
live = [g for g in games if is_live(g.get('status'))]
if not live:
# Debug: print every game's raw status to the console so we can tell whether
# there genuinely are no live games right now, or the matching logic missed one.
statuses = [(g.get('away_name'), g.get('home_name'), g.get('status')) for g in games]
print(f"No live games matched. Today's raw statuses: {statuses}")
c.privmsg(target, "No live games.")
return
for g in live:
away = colorize_team(g['away_name'])
home = colorize_team(g['home_name'])
self.send_line(c, target, f"[{g['status']}] {away} {g['away_score']} @ {home} {g['home_score']}")
except Exception as e:
print(f"Live games lookup error: {e}")
c.privmsg(target, "Error.")
def handle_tomorrow(self, c, target):
try:
et = ZoneInfo("America/New_York")
date = (datetime.now(et) + timedelta(days=1)).strftime("%Y-%m-%d")
games = statsapi.schedule(date=date, sportId=1)
self.send_line(c, target, f"--- Games for {date} ---")
for g in games:
away = colorize_team(g['away_name'])
home = colorize_team(g['home_name'])
self.send_line(c, target, f"{away} @ {home}")
except:
c.privmsg(target, "Error.")
def handle_probable(self, c, target):
try:
et = ZoneInfo("America/New_York")
games = statsapi.schedule(date=datetime.now(et).strftime("%Y-%m-%d"), sportId=1)
for g in games:
away = colorize_team(g['away_name'])
home = colorize_team(g['home_name'])
self.send_line(c, target, f"{away} ({g.get('away_probable_pitcher', 'TBD')}) @ {home} ({g.get('home_probable_pitcher', 'TBD')})")
except:
c.privmsg(target, "Error.")
def handle_random(self, c, target):
try:
et = ZoneInfo("America/New_York")
today = datetime.now(et).strftime("%Y-%m-%d")
games = statsapi.schedule(date=today, sportId=1)
if not games:
c.privmsg(target, "No games today to pick a random player from.")
return
g = random.choice(games)
team_name = random.choice([g.get('away_name'), g.get('home_name')])
team_id = get_team_id(team_name)
if not team_id:
c.privmsg(target, "Error.")
return
roster_text = statsapi.roster(team_id)
raw_lines = [l for l in roster_text.split("\n") if l.strip()]
names = []
for line in raw_lines:
fields = re.split(r"\s{2,}", line.strip())
if len(fields) >= 3:
names.append(fields[2])
elif len(fields) == 2:
names.append(fields[1])
if not names:
c.privmsg(target, "Error.")
return
full_name = random.choice(names)
name_parts = full_name.split()
if len(name_parts) < 2:
c.privmsg(target, "Error.")
return
first, last = name_parts[0], " ".join(name_parts[1:])
self.handle_stats(c, target, f"!stats {first} {last}")
except:
c.privmsg(target, "Error.")
def get_stat_block(self, pid, group):
"""Return (season_stats, career_stats) for a player's stat group ('hitting' or
'pitching'), or (None, None) if the player has no real appearances in that group
(e.g. a position player's pitching stats, or vice versa)."""
try:
career_data = statsapi.player_stat_data(pid, group=group, type="career")
if not career_data.get('stats') or not career_data['stats'][0].get('stats'):
return None, None
career_stats = career_data['stats'][0]['stats']
games = career_stats.get('gamesPlayed') or career_stats.get('gamesPitched') or 0
try:
games = int(games)
except (TypeError, ValueError):
games = 0
if games <= 0:
return None, None
except Exception as e:
print(f"Career stat lookup failed for pid {pid} group {group}: {e}")
return None, None
season_stats = None
try:
season_data = statsapi.player_stat_data(pid, group=group, type="season", season=self.current_year)
if season_data.get('stats') and season_data['stats'][0].get('stats'):
season_stats = season_data['stats'][0]['stats']
except Exception as e:
print(f"Season stat lookup failed for pid {pid} group {group}: {e}")
return season_stats, career_stats
def handle_stats(self, c, target, msg):
try:
parts = msg.split()
if len(parts) < 3:
c.privmsg(target, "Usage: !stats [first] [last]")
return
first_in, last_in = parts[1], " ".join(parts[2:])
player = playerid_lookup(last_in, first_in, fuzzy=True)
if player.empty:
c.privmsg(target, "Player not found.")
return
pid = int(player.iloc[0]['key_mlbam'])
first_name = str(player.iloc[0]['name_first']).title()
last_name = str(player.iloc[0]['name_last']).title()
bio = get_player_bio(pid)
header = format_bio_header(f"{first_name} {last_name}", bio)
# Retired players (confirmed via bio) skip the season line entirely -- no point
# checking a current season that doesn't exist for them.
is_retired = bool(bio and bio.get('active') is False)
lines = []
hitting_season, hitting_career = self.get_stat_block(pid, "hitting")
if hitting_career:
if not is_retired and hitting_season:
lines.append(f"{header} - {self.current_year} Hitting: AVG: {hitting_season.get('avg')} | HR: {hitting_season.get('homeRuns')} | OPS: {hitting_season.get('ops')}")
lines.append(f"{header} - Career Hitting: AVG: {hitting_career.get('avg')} | HR: {hitting_career.get('homeRuns')} | OPS: {hitting_career.get('ops')}")
pitching_season, pitching_career = self.get_stat_block(pid, "pitching")
if pitching_career:
if not is_retired and pitching_season:
lines.append(f"{header} - {self.current_year} Pitching: ERA: {pitching_season.get('era')} | SO: {pitching_season.get('strikeOuts')} | WHIP: {pitching_season.get('whip')}")
lines.append(f"{header} - Career Pitching: ERA: {pitching_career.get('era')} | SO: {pitching_career.get('strikeOuts')} | WHIP: {pitching_career.get('whip')}")
if not lines:
c.privmsg(target, f"No stats found for {header}.")
return
for line in lines:
self.send_line(c, target, line)
except:
c.privmsg(target, "Error retrieving stats.")
def handle_game(self, c, target, msg):
try:
search = msg.replace("!game", "").strip().lower()
today = datetime.now(ZoneInfo("America/New_York")).strftime("%Y-%m-%d")
games = statsapi.schedule(date=today, sportId=1)
found = False
for g in games:
away_plain = fix_team_name(g['away_name'])
home_plain = fix_team_name(g['home_name'])
if search in away_plain.lower() or search in home_plain.lower():
away = colorize_team(g['away_name'])
home = colorize_team(g['home_name'])
self.send_line(c, target, f"[{g['status']}] {away} {g['away_score']} @ {home} {g['home_score']}")
found = True
continue
if not found:
c.privmsg(target, "No matching game found.")
except:
c.privmsg(target, "Error.")
def handle_standings(self, c, target, msg):
try:
parts = msg.lower().split()
filter_term = " ".join(parts[1:]) if len(parts) > 1 else None
if filter_term:
tokens = filter_term.split()
if tokens[0] in ("al", "a.l."):
tokens[0] = "american league"
elif tokens[0] in ("nl", "n.l."):
tokens[0] = "national league"
filter_term = " ".join(tokens)
s = statsapi.standings_data(season=self.current_year)
for div in s.values():
div_name = div.get('div_name', 'Division')
if filter_term and filter_term not in div_name.lower():
continue
teams = div.get('teams')
if not teams:
self.send_line(c, target, f"{div_name}: No data.")
continue
self.send_line(c, target, f"--- {div_name} ---")
for team in teams:
raw_name = team.get('name') or team.get('team_name', 'Unknown')
colored_team = colorize_team(raw_name)
w = team.get('w', '?')
l = team.get('l', '?')
gb = team.get('gb', '-')
gb_text = "GB: -" if gb in ('-', '', None) else f"GB: {gb}"
self.send_line(c, target, f"{colored_team} ({w}-{l}) {gb_text}")
except Exception as e:
print(f"Standings Error: {e}")
traceback.print_exc()
c.privmsg(target, "Error retrieving standings.")
def handle_boxscore(self, c, target, msg):
try:
search = msg[len("!boxscore"):].strip()
if not search:
c.privmsg(target, "Usage: !boxscore [team]")
return
team_id = get_team_id(search)
if not team_id:
c.privmsg(target, "Team not found.")
return
game_id = statsapi.last_game(team_id)
if not game_id:
c.privmsg(target, "No recent game found for that team.")
return
text = statsapi.linescore(game_id)
lines = [l for l in text.split("\n") if l.strip()]
if not lines:
c.privmsg(target, "No line score available yet.")
return
for l in lines:
self.send_line(c, target, colorize_team_names_in_text(l))
except:
c.privmsg(target, "Error retrieving boxscore.")
def handle_highlights(self, c, target, msg):
try:
search = msg[len("!highlights"):].strip()
if not search:
c.privmsg(target, "Usage: !highlights [team]")
return
team_id = get_team_id(search)
if not team_id:
c.privmsg(target, "Team not found.")
return
game_id = statsapi.last_game(team_id)
if not game_id:
c.privmsg(target, "No recent game found for that team.")
return
text = statsapi.game_highlights(game_id)
if not text or not text.strip():
c.privmsg(target, "No highlights found for that game.")
return
blocks = [b for b in text.split("\n\n") if b.strip()]
if not blocks:
c.privmsg(target, "No highlights found for that game.")
return
for block in blocks[:3]:
block_lines = [l for l in block.split("\n") if l.strip()]
if block_lines:
self.send_line(c, target, colorize_team_names_in_text(block_lines[0]))
if len(block_lines) > 1:
self.send_line(c, target, colorize_team_names_in_text(block_lines[-1]))
except:
c.privmsg(target, "Error retrieving highlights.")
# Friendly stat names -> list of (statCode, statGroup) candidates, tried in order until
# one returns data. statGroup matters: without it, ambiguous stat names like battingAverage
# can resolve to the wrong side (e.g. a pitcher's "average against" instead of real hitters'
# batting average). Only homeRuns/battingAverage/earnedRunAverage are confirmed against the
# live API so far; the rest follow the same camelCase convention but haven't all been
# individually verified.
STAT_ALIASES = {
"hr": [("homeRuns", "hitting")],
"homers": [("homeRuns", "hitting")],
"homeruns": [("homeRuns", "hitting")],
"avg": [("battingAverage", "hitting")],
"average": [("battingAverage", "hitting")],
"battingaverage": [("battingAverage", "hitting")],
"rbi": [("rbi", "hitting"), ("runsBattedIn", "hitting")],
"rbis": [("rbi", "hitting"), ("runsBattedIn", "hitting")],
"sb": [("stolenBases", "hitting"), ("stolenBase", "hitting")],
"stolenbase": [("stolenBases", "hitting"), ("stolenBase", "hitting")],
"stolenbases": [("stolenBases", "hitting"), ("stolenBase", "hitting")],
"obp": [("onBasePercentage", "hitting")],
"slg": [("sluggingPercentage", "hitting")],
"ops": [("onBasePlusSlugging", "hitting"), ("ops", "hitting")],
"era": [("earnedRunAverage", "pitching")],
"so": [("strikeouts", "pitching")],
"k": [("strikeouts", "pitching")],
"strikeouts": [("strikeouts", "pitching")],
"wins": [("wins", "pitching")],
"w": [("wins", "pitching")],
"saves": [("saves", "pitching")],
"whip": [("whip", "pitching")],
}
def handle_leaders(self, c, target, msg):
try:
parts = msg.split()
if len(parts) < 2:
c.privmsg(target, "Usage: !leaders [stat] (e.g. hr, avg, rbi, sb, obp, slg, ops, era, so, wins, saves, whip)")
return
raw_stat = " ".join(parts[1:]).strip()
key = raw_stat.lower().replace(" ", "").replace("_", "")
candidates = self.STAT_ALIASES.get(key, [(raw_stat, None)])
text = None
for stat_code, stat_group in candidates:
try:
kwargs = {"limit": 5}
if stat_group:
kwargs["statGroup"] = stat_group
candidate_text = statsapi.league_leaders(stat_code, **kwargs)
if candidate_text and candidate_text.strip():
text = candidate_text
break
except Exception as e:
print(f"Leaders lookup error for stat '{stat_code}' (group={stat_group}): {e}")
continue
if not text:
c.privmsg(target, f"No leader data found for '{raw_stat}'. Try hr, avg, rbi, sb, obp, slg, ops, era, so, wins, saves, or whip.")
return
lines = [l for l in text.split("\n") if l.strip()]
if not lines:
c.privmsg(target, "No leader data found for that stat.")
return
for l in lines:
self.send_line(c, target, colorize_team_names_in_text(l))
except:
c.privmsg(target, "Error retrieving leaders. Try a stat like hr, avg, or era.")
def handle_roster(self, c, target, msg):
try:
search = msg[len("!roster"):].strip()
if not search:
c.privmsg(target, "Usage: !roster [team]")
return
team_id = get_team_id(search)
if not team_id:
c.privmsg(target, "Team not found.")
return
text = statsapi.roster(team_id)
raw_lines = [l for l in text.split("\n") if l.strip()]
if not raw_lines:
c.privmsg(target, "No roster data found.")
return
players = []
for line in raw_lines:
fields = re.split(r"\s{2,}", line.strip())
if len(fields) >= 3:
# e.g. ['#19', 'LF', 'Some Player']
players.append(f"{fields[1]} {fields[2]}")
elif len(fields) == 2:
players.append(fields[1])
else:
players.append(line.strip())
# Resolve a display name for the header from whatever the caller searched for
team_lookup = statsapi.lookup_team(search)
team_display = colorize_team(team_lookup[0]['name']) if team_lookup else colorize_team(search)
roster_line = f"{team_display} Roster: " + " - ".join(players)
# Safety net: IRC lines are capped around 512 bytes total (incl. protocol overhead).
# If a full active roster pushes past a safe length, split into a couple of messages
# instead of risking server-side truncation.
max_len = 420
if len(roster_line) <= max_len:
c.privmsg(target, roster_line)
else:
header = f"{team_display} Roster: "
chunk = header
first_in_chunk = True
for p in players:
piece = p if first_in_chunk else f" - {p}"
if len(chunk) + len(piece) > max_len:
self.send_line(c, target, chunk)
chunk = p
first_in_chunk = False
else:
chunk += piece
first_in_chunk = False
if chunk:
c.privmsg(target, chunk)
except:
c.privmsg(target, "Error retrieving roster.")
# Mapping of friendly names to real MLB award IDs. "cyyoung" has been confirmed working
# against the live API; mvp/roy/hof follow the same AL*/NL* naming pattern but haven't
# been explicitly tested yet.
AWARD_ALIASES = {
"mvp": ["ALMVP", "NLMVP"],
"cy": ["ALCY", "NLCY"],
"cyyoung": ["ALCY", "NLCY"],
"roy": ["ALROY", "NLROY"],
"rookie": ["ALROY", "NLROY"],
"hof": ["MLBHOF"],
"hall": ["MLBHOF"],
}
def handle_award(self, c, target, msg):
try:
parts = msg.split()
if len(parts) < 2:
c.privmsg(target, "Usage: !award [mvp|cyyoung|roy|hof] [year] [al|nl] (e.g. !award mvp 2024 al)")
return
arg = parts[1].lower()
remaining = parts[2:]
year = None
league_filter = None
for token in remaining:
if token.isdigit() and len(token) == 4:
year = token
elif token.lower() in ("al", "nl"):
league_filter = token.lower()
if not year:
year = str(self.current_year)
if arg in self.AWARD_ALIASES:
award_ids = list(self.AWARD_ALIASES[arg])
if league_filter == "al":
award_ids = [a for a in award_ids if a.startswith("AL")]
elif league_filter == "nl":
award_ids = [a for a in award_ids if a.startswith("NL")]
else:
# Not a known alias -- treat it as a literal award ID (e.g. "!award MLBHOF")
award_ids = [parts[1].upper()]
results = []
for aid in award_ids:
try:
data = statsapi.get('awards', {'awardId': aid, 'season': year, 'recipients': True})
recipients = data.get('awards', []) if isinstance(data, dict) else []
if not recipients:
print(f"Award debug [{aid} {year}] empty/unexpected response shape: {data}")
if recipients:
rec = recipients[0]
# Try several plausible schemas for the player's name -- the exact
# field layout differs by award type and isn't verified against the
# live API from this sandbox.
player_name = (
rec.get('player', {}).get('fullName')
or rec.get('player', {}).get('nameFirstLast')
or rec.get('recipient')
or (rec.get('players', [{}])[0].get('fullName') if rec.get('players') else None)
)
team_name = (
rec.get('team', {}).get('name')
or rec.get('teamName')
or ''
)
if not player_name:
# Couldn't find a name field we recognize -- dump the raw keys/record
# to the console (not IRC) so the actual schema can be inspected and
# the field lookup above corrected.
print(f"Award debug [{aid} {year}] unrecognized recipient structure: {rec}")
player_name = "Unknown"
line = f"{aid} ({year}): {player_name}"
if team_name:
line += f" - {colorize_team(team_name)}"
results.append(line)
except Exception as e:
print(f"Award lookup error for {aid} {year}: {e}")
continue
if not results:
c.privmsg(target, f"No award data found for '{parts[1]}' in {year}.")
return
for line in results:
self.send_line(c, target, line)
except:
c.privmsg(target, "Error retrieving award data.")
if __name__ == "__main__":
bot = BaseballBot(CHANNELS, NICKNAME, SERVER, PORT)
bot.start()
Reference in New Issue View Git Blame Copy Permalink