Initial Commit

This commit is contained in:
insert 2025-04-24 22:59:30 -04:00
parent fe70a525f8
commit dc18eb5b8c
Signed by: insert
GPG key ID: A70775C389ACF105
9 changed files with 678 additions and 0 deletions

178
.dockerignore Normal file
View file

@ -0,0 +1,178 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
http_cache
.direnv
*.db
.envrc
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc

4
.gitignore vendored
View file

@ -2,6 +2,10 @@
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
*$py.class *$py.class
http_cache
.direnv
*.db
.envrc
# C extensions # C extensions
*.so *.so

8
README.md Normal file
View file

@ -0,0 +1,8 @@
# MMOLBBOT
## Not affliated with mmolb in any way I just couldn't think of a better name
The bot is user-installable so the commands can either be added to a guild or your account
Invite: https://discord.com/oauth2/authorize?client_id=1365149865444905121
# Building
inside the folder
`docker build . -t mmolbbot`
`docker run mmolbbot -e TOKEN=<Token> -e OWNER_GUILD=<guild with management commands>`

62
bot.py Normal file
View file

@ -0,0 +1,62 @@
import nextcord
from nextcord.ext import application_checks, commands
from nextcord import SlashOption
from dotenv import load_dotenv
from random import randint
import aiohttp
import aiosqlite as sqlite3
import traceback
from urllib.parse import urlparse
import os
import logging
import sys
load_dotenv()
bot = commands.Bot()
@bot.event
async def on_ready():
global db
global cur
db = await sqlite3.connect("/data/mmolb.db")
cur = await db.cursor()
res = await cur.execute("SELECT name FROM sqlite_master WHERE name='liveupdate'")
if await res.fetchone() is None:
await cur.execute("CREATE TABLE liveupdate(serverid INTEGER, userid INTEGER, channelid INTEGER, messageid INTEGER, gameid TEXT, offset INTEGER)")
await db.commit()
bot.db = db
bot.cur = cur
bot.load_extension('cogs.liveupdate')
bot.load_extension('cogs.team')
bot.add_all_application_commands()
#await bot.sync_all_application_commands()
@bot.slash_command(
name="managecog",
description="manage cogs",
guild_ids=[int(os.getenv("OWNER_GUILD"))],
)
@application_checks.is_owner()
async def managecog(interaction: nextcord.Interaction,
action: str = SlashOption(choices=["load", "unload", "reload","sync"]),
cog: str = SlashOption(required=False)
):
errors = ""
try:
if action == "load":
bot.load_extension(cog)
elif action == "unload":
bot.unload_extension(cog)
elif action == "reload":
bot.reload_extension(cog)
elif action == "sync":
bot.add_all_application_commands()
await bot.sync_all_application_commands()
await interaction.response.send_message(f"Done!" + (f"\n{errors}" if errors != "" else ""), ephemeral=True)
except Exception as e:
await interaction.response.send_message(e, ephemeral=True)
bot.run(os.getenv("TOKEN"))

98
cogs/liveupdate.py Normal file
View file

@ -0,0 +1,98 @@
import hashlib
import json
from pathlib import Path
from datetime import datetime, timedelta, timezone
import requests
import asyncio
import nextcord
import aiosqlite as sqlite3
from nextcord.ext import commands, application_checks, tasks
from nextcord import TextInputStyle, IntegrationType
class liveupdate(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
@nextcord.slash_command(
name="liveupdates",
description="Get live updates on a game",
integration_types=[
IntegrationType.user_install,
IntegrationType.guild_install,
],
contexts=[
nextcord.InteractionContextType.guild,
nextcord.InteractionContextType.bot_dm,
nextcord.InteractionContextType.private_channel,
],
force_global=True,
)
async def liveupdatecreate(self, interaction: nextcord.Interaction, gameid: str):
data = requests.get(f"https://mmolb.com/api/game/{gameid}").json()
await interaction.response.send_message(f"{data["AwayTeamName"]} {data["AwayTeamEmoji"]} **{data["EventLog"][-1]["away_score"]}** vs {data["HomeTeamName"]} {data["HomeTeamEmoji"]} **{data["EventLog"][-1]["home_score"]}**")
message = await interaction.original_message()
await self.bot.cur.execute(f"""
INSERT INTO liveupdate VALUES
({interaction.guild_id}, {interaction.user.id}, {interaction.channel_id}, {message.id}, "{gameid}", {len(data["EventLog"])})
""")
await self.bot.db.commit()
await self.updatelivegames.start()
@nextcord.slash_command(
name="liveupdatesdelete",
description="Delete a subscribed update",
integration_types=[
IntegrationType.user_install,
IntegrationType.guild_install,
],
contexts=[
nextcord.InteractionContextType.guild,
nextcord.InteractionContextType.bot_dm,
nextcord.InteractionContextType.private_channel,
],
force_global=True,
)
async def liveupdatedelete(self, interaction: nextcord.Interaction, messageid: float):
await self.bot.cur.execute(f"""
DELETE from liveupdate WHERE messageid = {messageid}
""")
await self.bot.db.commit()
await interaction.response.send_message("stopped updates for message") #TODO This will be a button
@tasks.loop(seconds=10.0)
async def updatelivegames(self):
res = await self.bot.cur.execute("SELECT serverid,userid,channelid,messageid,gameid,offset FROM liveupdate")
res = await res.fetchall()
for [serverid,userid,channelid,messageid,gameid,offset] in res:
channel = self.bot.get_channel(channelid)
message = await channel.fetch_message(messageid)
data = requests.get(f"https://mmolb.com/api/game/{gameid}/live?after={offset}").json()
if len(data["entries"]) > 0:
splitstr = message.content.split("\n")
splitstr.pop(0)
while len(splitstr)>(5-len(data["entries"])):
splitstr.pop(0)
print(splitstr)
basedata = requests.get(f"https://mmolb.com/api/game/{gameid}").json()
finalstr = f"{basedata["AwayTeamName"]} {basedata["AwayTeamEmoji"]} **{data["entries"][-1]["away_score"]}** vs {basedata["HomeTeamName"]} {basedata["HomeTeamEmoji"]} **{data["entries"][-1]["home_score"]}**"
for i in splitstr:
finalstr += f"\n{i}"
for i in data["entries"]:
finalstr += f"\n{i['message']}"
await self.bot.cur.execute(f"""
UPDATE liveupdate set offset = {offset+1} WHERE messageid = '{messageid}'
""") #Could do this for every meessage subscribed to the game but since the messages go one by one... maybe I should change that
await self.bot.db.commit()
if i["event"] == "Recordkeeping":
await self.bot.cur.execute(f"""
DELETE from liveupdate WHERE messageid = {messageid}
""")
await self.bot.db.commit()
await message.edit(finalstr)
def setup(bot: commands.Bot):
bot.add_cog(liveupdate(bot))

305
cogs/team.py Normal file
View file

@ -0,0 +1,305 @@
import hashlib
import json
from pathlib import Path
from datetime import datetime, timedelta, timezone
import requests
import re
import asyncio
import nextcord
from nextcord.ext import commands, application_checks
from nextcord import TextInputStyle, IntegrationType
#all of this code is by evilscientist3, full credit to them
HTTP_CACHE_DIR = Path("http_cache")
statdict = {
"BA":(0.4, 0.35, 0.3, 0.25, 0.2, 0.15),
"OBP":(0.475, 0.425, 0.375, 0.325, 0.275, 0.225),
"SLG":(0.55, 0.52, 0.46, 0.4, 0.36, 0.33),
"OPS":(1.0, 0.9, 0.8, 0.7, 0.6, 0.5),
"ERA":(2.75, 3.25, 3.75, 4.75, 5.5, 6.5),
"WHIP":(1, 1.1, 1.25, 1.4, 1.5, 1.6)
}
# whether ascending is GOOD (True) or BAD (False) for figure of merit
NGUdict = {
"BA":True,
"OBP":True,
"SLG":True,
"OPS":True,
"ERA":False,
"WHIP":False
}
# sig figs for each stat
sfdict = {
"BA":3,
"OBP":3,
"SLG":3,
"OPS":3,
"ERA":2,
"WHIP":2
}
def eval_stat(stat, val):
# modifier for if descending is better. in this case logic is reversed; most easily accounted for by multiplying bounds and values by -1
# why bother keeping descending values' bounds positive, you ask? for human readability and ease of modification, I say
mod = 1 if NGUdict[stat] else -1
# set of ANSI colour codes. best to worst, one more than bounds
colours = [32, 72, 36, 30, 33, 31, 31]
# multiply bounds to account for descending
bounds = [mod*i for i in statdict[stat]]
mval = mod*val
# flag for colour being set, as an "else" condition to the entire loop
cset = 0
# main loop - checks quality in descending order
for i, j in enumerate(bounds):
if mval > j:
colour = colours[i]
cset = 1
break
# check if value is below all bounds
if cset == 0:
colour = colours[-1]
# return stat string in correct colour
return f"\033[38;5;{colour}m{val:.{sfdict[stat]}f}\033[0m"
def stable_str_hash(in_val: str) -> str:
return hex(int(hashlib.md5(in_val.encode("utf-8")).hexdigest(), 16))[2:]
def get_json(url: str) -> dict:
HTTP_CACHE_DIR.mkdir(exist_ok=True)
cache = {}
cache_file_path = HTTP_CACHE_DIR / f"{stable_str_hash(url)}.json"
try:
with open(cache_file_path) as cache_file:
cache = json.load(cache_file)
except FileNotFoundError:
pass
# Return from cache if the cache entry is less than 5 minutes old
now = datetime.now(timezone.utc)
if (
url in cache and
"__archived_at" in cache[url] and
cache[url]["__archived_at"] > (now - timedelta(minutes=5)).isoformat()
):
return cache[url]
data = requests.get(url).json()
cache[url] = data
cache[url]["__archived_at"] = now.isoformat()
with open(cache_file_path, "w") as cache_file:
json.dump(cache, cache_file)
return data
def teamstats(MY_TEAM_ID):
finalstr = ""
team_obj = get_json(f"https://mmolb.com/api/team/{MY_TEAM_ID}")
for player in team_obj["Players"]:
player_obj = get_json(f"https://mmolb.com/api/player/{player['PlayerID']}")
try:
# I'm pretty sure IDs are lexicographically ordered, so we want the
# maximum value to get stats for the latest season
stats_obj = player_obj["Stats"][max(player_obj["Stats"].keys())]
except ValueError:
finalstr += (f"No stats for {player["FirstName"]} {player["LastName"]} \n")
continue
singles = stats_obj.get("singles", 0)
doubles = stats_obj.get("doubles", 0)
triples = stats_obj.get("triples", 0)
home_runs = stats_obj.get("home_runs", 0)
hits = singles + doubles + triples + home_runs
bb = stats_obj.get("walked", 0)
hbp = stats_obj.get("hit_by_pitch", 0)
earned_runs = stats_obj.get("earned_runs", 0)
hits_allowed = stats_obj.get("hits_allowed", 0)
walks = stats_obj.get("walks", 0)
try:
ab = stats_obj["at_bats"]
except KeyError:
ba_str = None
else:
ba = hits / ab
ba_str = f"BA: {eval_stat('BA', ba)} ({ab} AB)"
try:
pa = stats_obj["plate_appearances"]
ab = stats_obj["at_bats"]
except KeyError:
ops_str = None
else:
obp = (hits + bb + hbp) / pa
slg = (singles + 2 * doubles + 3 * triples + 4 * home_runs) / ab
ops = obp + slg
pa_str = dot_format(pa)
ops_str = f"OBP: {eval_stat('OBP', obp)}, SLG: {eval_stat('SLG', slg)}, OPS: {eval_stat('OPS', ops)} ({pa_str} PA)"
try:
ip = stats_obj["outs"] / 3
except KeyError:
era_str = None
else:
era = 9 * earned_runs / ip
whip = (walks + hits_allowed) / ip
ip_str = dot_format(ip)
era_str = f"ERA {eval_stat('ERA', era)} WHIP {eval_stat('WHIP', whip)} ({ip_str} IP)"
stats_str = ", ".join(s for s in [ba_str, ops_str, era_str] if s is not None)
if stats_str:
finalstr += (f"{player["Position"]} {player["FirstName"]} {player["LastName"]} ")
finalstr += "" * (30 - (len(player["Position"]) + len(player["FirstName"]) + len(player["LastName"])))
finalstr += f"{stats_str}\n"
return finalstr
def dot_format(in_val: float) -> str:
ip_whole = int(in_val)
ip_remainder = int((in_val - ip_whole) / 3 * 10)
if ip_remainder == 0:
ip_str = f"{ip_whole}"
else:
ip_str = f"{ip_whole}.{ip_remainder}"
return ip_str
#-------
class TeamView(nextcord.ui.View):
def __init__(self):
super().__init__(timeout=None)
@nextcord.ui.button(
label="Players", style=nextcord.ButtonStyle.green, custom_id="team:players"
)
async def getplayersbutton(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
await interaction.response.defer(ephemeral=True)
loop = asyncio.get_event_loop()
ogmsg = interaction.message.embeds
embed = ogmsg[0]
stats = await loop.run_in_executor(None, teamstats,embed.fields[0].value)
splistats = [stats[i:i+1900] for i in range(0, len(stats), 1900)] #TODO make better as in don't do cutoff
for i in splistats:
await interaction.followup.send(f"```ansi\n{i}```",ephemeral=True)
class team(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
@nextcord.slash_command(
name="greaterteam",
description="Get information about a greater leauge team",
integration_types=[
IntegrationType.user_install,
IntegrationType.guild_install,
],
contexts=[
nextcord.InteractionContextType.guild,
nextcord.InteractionContextType.bot_dm,
nextcord.InteractionContextType.private_channel,
],
force_global=True,
)
async def greaterteam(self, interaction: nextcord.Interaction, teamid: str = nextcord.SlashOption(
name = "team",
description = "The greater leauge team",
choices = {
"Seattle Shine" : "6805db0cac48194de3cd40a2",
"Chicago Seers": "6805db0cac48194de3cd40b5",
"Atlanta Tree Frogs": "6805db0cac48194de3cd40ee",
"Baltimore Lady Beetles": "6805db0cac48194de3cd407c",
"St. Louis Archers": "6805db0cac48194de3cd400a",
"Anaheim Angles": "6805db0cac48194de3cd401d",
"Miami Merfolk": "6805db0cac48194de3cd4101",
"Washington Baseball Team": "6805db0cac48194de3cd3ff7",
"Dallas Instruments": "6805db0cac48194de3cd4114",
"Roswell Weather Balloons": "6805db0cac48194de3cd40c8",
"Toronto Northern Lights": "6805db0cac48194de3cd4043",
"Kansas City Stormchasers": "6805db0cac48194de3cd4069",
"Philadelphia Phantasms": "6805db0cac48194de3cd40db",
"Durhamshire Badgers": "6805db0cac48194de3cd4056",
"Boston Street Sweepers": "6805db0cac48194de3cd408f",
"Brooklyn Scooter Dodgers": "6805db0cac48194de3cd4030"
}
)):
await interaction.response.defer()
data = requests.get(f"https://mmolb.com/api/team/{teamid}").json()
color = tuple(int(data["Color"][i:i+2], 16) for i in (0, 2, 4))
embed = nextcord.Embed(title=f"{data["Location"]} {data["Name"]} {data["Emoji"]}", description=f"{data["Motto"]}", colour = nextcord.Color.from_rgb(color[0], color[1], color[2]))
embed.add_field(name="Team ID", value=f"{teamid}", inline=False)
embed.add_field(name="Wins", value=f"{data["Record"]["Regular Season"]["Wins"]}", inline=True)
embed.add_field(name="Losses", value=f"{data["Record"]["Regular Season"]["Losses"]}", inline=True)
embed.add_field(name="Run Differential", value=f"{data["Record"]["Regular Season"]["RunDifferential"]}", inline=True)
embed.add_field(name="Augments", value=f"{data["Augments"]}", inline=True)
embed.add_field(name="Champtionships", value=f"{data["Championships"]}", inline=True)
await interaction.edit_original_message(embed=embed,view=TeamView())
@nextcord.slash_command(
name="lesserteam",
description="Get information about a lesser leauge team",
integration_types=[
IntegrationType.user_install,
IntegrationType.guild_install,
],
contexts=[
nextcord.InteractionContextType.guild,
nextcord.InteractionContextType.bot_dm,
nextcord.InteractionContextType.private_channel,
],
force_global=True,
)
async def lesserteam(self, interaction: nextcord.Interaction, teamid: str = nextcord.SlashOption(
name = "team",
description = "The lesser leauge team"
)):
await interaction.response.defer()
data = requests.get(f"https://mmolb.com/api/team/{teamid}").json()
color = tuple(int(data["Color"][i:i+2], 16) for i in (0, 2, 4))
embed = nextcord.Embed(title=f"{data["Location"]} {data["Name"]} {data["Emoji"]}", description=f"{data["Motto"]}", colour = nextcord.Color.from_rgb(color[0], color[1], color[2]))
embed.add_field(name="Team ID", value=f"{teamid}", inline=False)
embed.add_field(name="Wins", value=f"{data["Record"]["Regular Season"]["Wins"]}", inline=True)
embed.add_field(name="Losses", value=f"{data["Record"]["Regular Season"]["Losses"]}", inline=True)
embed.add_field(name="Run Differential", value=f"{data["Record"]["Regular Season"]["RunDifferential"]}", inline=True)
embed.add_field(name="Augments", value=f"{data["Augments"]}", inline=True)
embed.add_field(name="Champtionships", value=f"{data["Championships"]}", inline=True)
await interaction.edit_original_message(embed=embed,view=TeamView())
@nextcord.slash_command(
name="teamstats",
description="Get a teams stats",
integration_types=[
IntegrationType.user_install,
IntegrationType.guild_install,
],
contexts=[
nextcord.InteractionContextType.guild,
nextcord.InteractionContextType.bot_dm,
nextcord.InteractionContextType.private_channel,
],
force_global=True,
)
async def teamstats(self, interaction: nextcord.Interaction, teamid: str):
await interaction.response.defer()
loop = asyncio.get_event_loop()
stats = await loop.run_in_executor(None, teamstats,teamid)
splistats = [stats[i:i+1900] for i in range(0, len(stats), 1900)] #TODO make better as in don't do cutoff
await interaction.edit_original_message(content=f"```ansi\n{splistats[0]}```")
splistats.pop(0)
for i in splistats:
await interaction.followup.send(f"```ansi\n{i}```")
def setup(bot: commands.Bot):
bot.add_cog(team(bot))

7
dockerfile Normal file
View file

@ -0,0 +1,7 @@
FROM python:3.13
WORKDIR /mmolbbot
COPY . /mmolbbot
RUN pip install -U --timeout 10000 -r requirements.txt
CMD ["python", "bot.py"]

5
requirements.txt Normal file
View file

@ -0,0 +1,5 @@
nextcord
requests
aiosqlite
aiohttp
python-dotenv

11
shell.nix Normal file
View file

@ -0,0 +1,11 @@
{ pkgs ? import <nixpkgs> {}}:
pkgs.mkShell {
packages = [
pkgs.python312
pkgs.python312Packages.pip
pkgs.python312Packages.python-dotenv
pkgs.python312Packages.venvShellHook
];
venvDir = "./.venv";
}