complete the bot

This commit is contained in:
Rainy21
2026-04-22 00:49:43 -04:00
parent b2f09d6cca
commit c6257c5260
8 changed files with 845 additions and 42 deletions
+166 -43
View File
@@ -1,59 +1,182 @@
import discord import discord
from discord import app_commands from discord.ext import commands, tasks
import asyncio
import os import os
import logging
from datetime import datetime
from dotenv import load_dotenv from dotenv import load_dotenv
load_dotenv() load_dotenv()
Discord_Token = os.getenv("DISCORDTOKEN")
# ── Logging ────────────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[
logging.FileHandler("bot.log"),
logging.StreamHandler(), # also print to console
],
)
log = logging.getLogger("bot")
# ── Bot setup ──────────────────────────────────────────────────────────────────
EXTENSIONS = [
"commands.login",
"commands.profile",
"commands.recent",
]
class LoLBot(commands.Bot):
def __init__(self):
intents = discord.Intents.default() intents = discord.Intents.default()
client = discord.Client(intents=intents) intents.message_content = True
tree = app_commands.CommandTree(client)
# --command --------------------------------------------------------- super().__init__(
command_prefix="!",
intents=intents,
@tree.command(name="ping", description="Test if bot is alive") help_command=None, # disable default help
async def ping(interaction: discord.Interaction):
await interaction.response.send_message("Pong!", ephemeral=True)
@tree.command(name="login", description="Link your Riot account")
async def login(interaction: discord.Interaction, summoner_name: str, region: str):
await interaction.response.send_message(
f"Received: {summoner_name} on {region} — login logic coming soon!",
ephemeral=True,
) )
# ── Lifecycle ──────────────────────────────────────────────────────────────
@tree.command(name="recent", description="See your recent matches") async def setup_hook(self):
async def recent(interaction: discord.Interaction): """Called once before the bot connects — load extensions here."""
await interaction.response.send_message( for ext in EXTENSIONS:
"Fetching your recent matches — coming soon!", ephemeral=True
)
# --StartUp -------------------------------------------------------
@client.event
async def on_ready():
server_id = os.getenv("SERVER_ID")
if server_id:
try: try:
guild = discord.Object(id=int(server_id)) await self.load_extension(ext)
tree.copy_global_to(guild=guild) log.info(f"✅ Loaded extension: {ext}")
await tree.sync(guild=guild) except Exception as e:
print(f"Logged in as {client.user} (Synced to Guild ID: {server_id})") log.error(f"❌ Failed to load {ext}: {e}")
except ValueError:
print(f"Error: SERVER_ID '{server_id}' is not a valid integer. Syncing globally instead.") await self.tree.sync()
await tree.sync() log.info("🔄 Slash commands synced")
print(f"Logged in as {client.user} (Global Sync)")
async def on_ready(self):
log.info(f"✅ Logged in as {self.user} (ID: {self.user.id})")
log.info(f" Guilds : {len(self.guilds)}")
log.info(f" Ping : {round(self.latency * 1000)}ms")
await self.change_presence(
activity=discord.Activity(
type=discord.ActivityType.watching, name="League of Legends 🎮"
)
)
self.status_task.start()
async def on_disconnect(self):
log.warning("⚠️ Bot disconnected from Discord")
async def on_resumed(self):
log.info("🔄 Bot reconnected")
# ── Error handling ─────────────────────────────────────────────────────────
async def on_command_error(self, ctx, error):
log.error(f"Command error in {ctx.command}: {error}")
async def on_application_command_error(
self, interaction: discord.Interaction, error
):
log.error(f"Slash command error: {error}")
msg = "❌ Something went wrong. Please try again later."
if interaction.response.is_done():
await interaction.followup.send(msg, ephemeral=True)
else: else:
await tree.sync() await interaction.response.send_message(msg, ephemeral=True)
print(f"Logged in as {client.user} (Global Sync)")
async def on_guild_join(self, guild: discord.Guild):
log.info(f"➕ Joined guild: {guild.name} (ID: {guild.id})")
async def on_guild_remove(self, guild: discord.Guild):
log.info(f"➖ Left guild: {guild.name} (ID: {guild.id})")
# ── Background task ────────────────────────────────────────────────────────
@tasks.loop(minutes=30)
async def status_task(self):
"""Rotate bot status every 30 minutes."""
statuses = [
discord.Activity(
type=discord.ActivityType.watching, name="League of Legends 🎮"
),
discord.Activity(
type=discord.ActivityType.listening, name="/login | /profile | /recent"
),
discord.Activity(
type=discord.ActivityType.watching, name=f"{len(self.guilds)} servers"
),
]
index = (self.status_task.current_loop) % len(statuses)
await self.change_presence(activity=statuses[index])
@status_task.before_loop
async def before_status_task(self):
await self.wait_until_ready()
if Discord_Token: # ── Dev commands (owner only) ──────────────────────────────────────────────────
client.run(Discord_Token)
else:
print("Error: DISCORDTOKEN not found in .env file.")
bot = LoLBot()
@bot.command(name="sync")
@commands.is_owner()
async def sync(ctx):
"""Force re-sync slash commands (owner only)."""
await bot.tree.sync()
await ctx.send("✅ Slash commands synced.")
log.info("🔄 Manual sync triggered")
@bot.command(name="reload")
@commands.is_owner()
async def reload(ctx, extension: str):
"""Reload a specific extension (owner only). Usage: !reload commands.login"""
try:
await bot.reload_extension(extension)
await ctx.send(f"✅ Reloaded `{extension}`")
log.info(f"🔄 Reloaded extension: {extension}")
except Exception as e:
await ctx.send(f"❌ Failed: {e}")
log.error(f"Reload failed for {extension}: {e}")
@bot.command(name="ping")
async def ping(ctx):
"""Check bot latency."""
await ctx.send(f"🏓 Pong! `{round(bot.latency * 1000)}ms`")
@bot.command(name="status")
@commands.is_owner()
async def status(ctx):
"""Show bot status info (owner only)."""
uptime = datetime.utcnow() - bot.start_time if hasattr(bot, "start_time") else "N/A"
embed = discord.Embed(title="🤖 Bot Status", color=0x1A78C2)
embed.add_field(name="📡 Ping", value=f"{round(bot.latency * 1000)}ms", inline=True)
embed.add_field(name="🏠 Guilds", value=str(len(bot.guilds)), inline=True)
embed.add_field(name="🔌 Shards", value=str(bot.shard_count or 1), inline=True)
await ctx.send(embed=embed)
# ── Entry point ────────────────────────────────────────────────────────────────
async def main():
token = os.getenv("DISCORDTOKEN")
if not token:
log.critical("❌ DISCORDTOKEN not found in .env — exiting")
return
async with bot:
bot.start_time = datetime.utcnow()
await bot.start(token)
asyncio.run(main())
+59
View File
@@ -0,0 +1,59 @@
import discord
from discord import app_commands
from discord.ext import commands
import aiohttp
import os
from database.db import save_user
RIOT_API_KEY = os.getenv("LEAGUEAPI")
class LoginCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
@app_commands.command(name="login", description="Link your Riot account")
@app_commands.describe(summoner="Your Riot ID (e.g. Name#TAG)")
async def login(self, interaction: discord.Interaction, summoner: str):
await interaction.response.defer(ephemeral=True)
if "#" not in summoner:
await interaction.followup.send("❌ Use the format `Name#TAG`", ephemeral=True)
return
try:
game_name, tag_line = summoner.split("#", 1)
except ValueError:
await interaction.followup.send("❌ Invalid format. Use `Name#TAG`", ephemeral=True)
return
discord_id = str(interaction.user.id)
url = (
f"https://americas.api.riotgames.com/riot/account/v1/accounts"
f"/by-riot-id/{game_name}/{tag_line}"
)
headers = {"X-Riot-Token": RIOT_API_KEY}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as resp:
if resp.status == 404:
await interaction.followup.send("❌ Summoner not found. Check your Riot ID.", ephemeral=True)
return
if resp.status != 200:
await interaction.followup.send(f"❌ Riot API error ({resp.status}). Try again later.", ephemeral=True)
return
data = await resp.json()
puuid = data["puuid"]
save_user(discord_id, puuid, game_name, tag_line)
await interaction.followup.send(
f"✅ Linked **{game_name}#{tag_line}** to your Discord account!",
ephemeral=True
)
async def setup(bot):
await bot.add_cog(LoginCog(bot))
+110
View File
@@ -0,0 +1,110 @@
from discord import app_commands
from discord.ext import commands
import aiohttp
import os
from database.db import get_user
RIOT_API_KEY = os.getenv("LEAGUEAPI")
REGION = "na1"
class ProfileCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
@app_commands.command(name="profile", description="Show your LoL rank, level, and win rate")
async def profile(self, interaction: discord.Interaction):
await interaction.response.defer(ephemeral = False)
# get user from the db
discord_id = str(interaction.user.id)
user = get_user(discord_id)
#check if user login
if not user:
await interaction.followup.send(
"❌ You haven't linked your Riot account yet. Use `/login` first."
)
return
_, puuid, game_name, tag_line = user
headers = {"X-Riot-Token": RIOT_API_KEY}
async with aiohttp.ClientSession() as session:
# 1. Get summoner data (level) via PUUID
summoner = await fetch(
session,
f"https://{REGION}.api.riotgames.com/lol/summoner/v4/summoners/by-puuid/{puuid}",
headers
)
if not summoner:
await interaction.followup.send("❌ Could not fetch summoner data.")
return
summoner_id = summoner["id"]
level = summoner["summonerLevel"]
# 2. Get rank & LP (League-v4)
leagues = await fetch(
session,
f"https://{REGION}.api.riotgames.com/lol/league/v4/entries/by-summoner/{summoner_id}",
headers
)
# Parse ranked solo queue data
rank_info = get_solo_rank(leagues)
# Build win rate
wins = rank_info.get("wins", 0)
losses = rank_info.get("losses", 0)
total = wins + losses
win_rate = f"{round(wins / total * 100)}%" if total > 0 else "N/A"
# Build Discord embed
embed = discord.Embed(
title=f"📊 {game_name}#{tag_line}",
color=0x1a78c2
)
embed.add_field(name="🎮 Level", value=str(level), inline=True)
embed.add_field(
name="🏆 Rank",
value=format_rank(rank_info),
inline=True
)
embed.add_field(name="📈 Win Rate", value=win_rate, inline=True)
embed.add_field(
name="📋 Record",
value=f"{wins}W / {losses}L" if total > 0 else "Unranked",
inline=True
)
embed.set_footer(text="Powered by Riot Games API")
await interaction.followup.send(embed=embed)
# ── Helpers ────────────────────────────────────────────────────────────────────
#try to get the information
async def fetch(session:aiohttp.ClientSession, url: str, headers: dict):
async with session.get(url, headers=headers) as resp:
if resp.status != 200:
return None
return await resp.json()
def get_solo_rank(leagues: list) -> dict:
"""Extract solo queue entry from league data."""
if not leagues:
return {}
for entry in leagues:
if entry.get("queueType") == "RANKED_SOLO_5x5":
return entry
return {}
def format_rank(rank_info:dict) -> str:
"""Format rank into a readable string."""
if not rank_info:
return "Unranked"
tier = rank_info.get("tier", "Unranked").capitalize()
division = rank_info.get("rank", "")
lp = rank_info.get("leaguePoints", 0)
return f"{tier} {division} — {lp} LP"
async def setup(bot):
await bot.add_cog(ProfileCog(bot))
+157
View File
@@ -0,0 +1,157 @@
import discord
from discord import app_commands
from discord.ext import commands
import aiohttp
import os
from database.db import get_user
RIOT_API_KEY = os.getenv("LEAGUEAPI")
REGION = "na1"
CLUSTER = "americas"
CLUSTER_MAP = {
"na1": "americas", "br1": "americas", "la1": "americas", "la2": "americas",
"euw1": "europe", "eun1": "europe", "tr1": "europe", "ru": "europe",
"kr": "asia", "jp1": "asia",
}
class RecentCog(commands.Cog):
def __init__(self, bot) -> None:
self.bot = bot
@app_commands.command(name = "recent", description="Show your last N matches")
@app_commands.describe(count="Number of matches to show (1–10, default 5)")
async def recent(self , interaction: discord.Interaction, count: int = 5):
await interaction.response.defer(ephemeral=False)
count = max(1, min(count, 10)) # clamp between 1 and 10
discord_id = str(interaction.user.id)
user = get_user(discord_id)
#check if user
if not user:
await interaction.followup.send(
"❌ You haven't linked your account yet. Use `/login` first."
)
return
#get user
_, puuid, game_name, tag_line = user
cluster = CLUSTER_MAP.get(REGION, "americas")
headers = {"X-Riot-Token": RIOT_API_KEY}
#get user data
async with aiohttp.ClientSession() as session:
# 1. Fetch match ID list
match_ids = await fetch(
session,
f"https://{cluster}.api.riotgames.com/lol/match/v5/matches/by-puuid/{puuid}/ids?count={count}",
headers
)
if not match_ids:
await interaction.followup.send("❌ Could not fetch match history.")
return
# 2. Fetch each match in parallel
match_data = await fetch_all(session, [
f"https://{cluster}.api.riotgames.com/lol/match/v5/matches/{mid}"
for mid in match_ids
], headers)
embeds = []
for match in match_data:
if match:
embed = build_match_embed(match , puuid, game_name,tag_line)
if embed:
embeds.append(embed)
if not embeds:
await interaction.followup.send("❌ No recent matches found.")
return
# Discord allows max 10 embeds per message
await interaction.followup.send(embeds=embeds[:10])
# ── Helpers ────────────────────────────────────────────────────────────────────
async def fetch(session: aiohttp.ClientSession, url: str, headers: dict):
async with session.get(url, headers=headers) as resp:
if resp.status != 200:
return None
return await resp.json()
async def fetch_all(session: aiohttp.ClientSession, urls: list, headers: dict):
"""Fetch multiple URLs concurrently."""
import asyncio
return await asyncio.gather(*[fetch(session, url, headers) for url in urls])
def get_player(match: dict, puuid: str) -> dict:
"""Find this player's participant data within the match."""
participants = match["info"]["participants"]
return next((p for p in participants if p["puuid"] == puuid), None)
def format_kda(kills, deaths, assists) -> str:
ratio = (kills + assists) / max(deaths, 1)
return f"{kills}/{deaths}/{assists} ({ratio:.2f} KDA)"
def format_duration(seconds: int) -> str:
m, s = divmod(seconds, 60)
return f"{m}m {s}s"
def queue_name(queue_id: int) -> str:
queues = {
420: "Ranked Solo",
440: "Ranked Flex",
450: "ARAM",
400: "Normal Draft",
430: "Normal Blind",
}
return queues.get(queue_id, "Other")
def build_match_embed(match: dict, puuid: str, game_name: str, tag_line: str):
player = get_player(match, puuid)
if not player:
return None
won = player["win"]
champion = player["championName"]
kills = player["kills"]
deaths = player["deaths"]
assists = player["assists"]
cs = player["totalMinionsKilled"] + player["neutralMinionsKilled"]
duration = match["info"]["gameDuration"]
queue_id = match["info"]["queueId"]
cs_per_min = round(cs / max(duration / 60, 1), 1)
color = 0x2ecc71 if won else 0xe74c3c # green = win, red = loss
result = "✅ Victory" if won else "❌ Defeat"
embed = discord.Embed(
title=f"{result} — {champion}",
description=f"**{queue_name(queue_id)}** · {format_duration(duration)}",
color=color
)
embed.add_field(name="⚔️ KDA", value=format_kda(kills, deaths, assists), inline=True)
embed.add_field(name="🌾 CS", value=f"{cs} ({cs_per_min}/min)", inline=True)
embed.add_field(name="👤 Player", value=f"{game_name}#{tag_line}", inline=True)
embed.set_thumbnail(
url=f"https://ddragon.leagueoflegends.com/cdn/14.8.1/img/champion/{champion}.png"
)
return embed
#---------------------------------------------------------------------------
#how to start the bot
async def setup(bot):
await bot.add_cog(RecentCog(bot))
+34
View File
@@ -0,0 +1,34 @@
import sqlite3
conn = sqlite3.connect("user.db")
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users(
discord_id TEXT PRIMARY KEY,
puuid TEXT NOT NULL,
game_name TEXT,
tag_line TEXT
)
""")
conn.commit()
def save_user(discord_id: str, puuid: str, game_name: str, tag_line: str):
cursor.execute(
"""
INSERT INTO users(discord_id,puuid,game_name,tag_line)
VALUEs (?,?,?,?)
ON CONFLICT(discord_id) DO UPDATE SET
puuid = excluded.puuid,
game_name=excluded.game_name,
tag_line=excluded.tag_line
""",
(discord_id, puuid, game_name, tag_line),
)
conn.commit()
def get_user(discord_id: str):
cursor.execute("SELECT * FROM users WHERE discord_id = ?", (discord_id,))
return cursor.fetchone()
+2 -1
View File
@@ -1,4 +1,5 @@
discord.py discord.py
python-doten python-dotenv
httpx httpx
aiosqlite aiosqlite
aiohttp
+223
View File
@@ -0,0 +1,223 @@
import discord
from dataclasses import dataclass
# ── Data class ─────────────────────────────────────────────────────────────────
@dataclass
class MatchSummary:
match_id: str
champion: str
won: bool
kills: int
deaths: int
assists: int
cs: int
vision: int
damage: int
gold: int
duration: int # seconds
queue_id: int
items: list[int]
spell1_id: int
spell2_id: int
game_name: str
tag_line: str
patch: str
# ── Parser ─────────────────────────────────────────────────────────────────────
def parse_match(
match: dict, puuid: str, game_name: str, tag_line: str, patch: str
) -> MatchSummary | None:
"""Extract a clean MatchSummary from raw Riot API match data."""
participants = match["info"]["participants"]
player = next((p for p in participants if p["puuid"] == puuid), None)
if not player:
return None
return MatchSummary(
match_id=match["metadata"]["matchId"],
champion=player["championName"],
won=player["win"],
kills=player["kills"],
deaths=player["deaths"],
assists=player["assists"],
cs=player["totalMinionsKilled"] + player["neutralMinionsKilled"],
vision=player["visionScore"],
damage=player["totalDamageDealtToChampions"],
gold=player["goldEarned"],
duration=match["info"]["gameDuration"],
queue_id=match["info"]["queueId"],
items=[player[f"item{i}"] for i in range(7) if player.get(f"item{i}", 0) != 0],
spell1_id=player["summoner1Id"],
spell2_id=player["summoner2Id"],
game_name=game_name,
tag_line=tag_line,
patch=patch,
)
# ── Formatters ─────────────────────────────────────────────────────────────────
def fmt_kda(kills: int, deaths: int, assists: int) -> str:
ratio = (kills + assists) / max(deaths, 1)
return f"{kills}/{deaths}/{assists} ({ratio:.2f} KDA)"
def fmt_duration(seconds: int) -> str:
m, s = divmod(seconds, 60)
return f"{m}m {s}s"
def fmt_cs(cs: int, duration: int) -> str:
per_min = round(cs / max(duration / 60, 1), 1)
return f"{cs} ({per_min}/min)"
def fmt_damage(damage: int) -> str:
return f"{damage:,}"
def fmt_gold(gold: int) -> str:
return f"{gold:,} g"
def fmt_rank_entry(entry: dict | None) -> str:
if not entry:
return "Unranked"
tier = entry["tier"].capitalize()
division = entry["rank"]
lp = entry["leaguePoints"]
wins = entry["wins"]
losses = entry["losses"]
total = wins + losses
wr = f"{round(wins / total * 100)}%" if total > 0 else "N/A"
return f"{tier} {division} — {lp} LP ({wr} WR)"
QUEUE_NAMES = {
420: "Ranked Solo/Duo",
440: "Ranked Flex",
450: "ARAM",
400: "Normal Draft",
430: "Normal Blind",
900: "URF",
1020: "One for All",
1300: "Nexus Blitz",
}
SUMMONER_SPELL_NAMES = {
1: "Cleanse",
3: "Exhaust",
4: "Flash",
6: "Ghost",
7: "Heal",
11: "Smite",
12: "Teleport",
13: "Clarity",
14: "Ignite",
21: "Barrier",
32: "Mark",
}
def fmt_queue(queue_id: int) -> str:
return QUEUE_NAMES.get(queue_id, "Other")
def fmt_spells(spell1_id: int, spell2_id: int) -> str:
s1 = SUMMONER_SPELL_NAMES.get(spell1_id, f"#{spell1_id}")
s2 = SUMMONER_SPELL_NAMES.get(spell2_id, f"#{spell2_id}")
return f"{s1} / {s2}"
# ── Embed builders ─────────────────────────────────────────────────────────────
def build_match_embed(s: MatchSummary) -> discord.Embed:
"""Full match embed with all stats."""
result = "✅ Victory" if s.won else "❌ Defeat"
color = 0x2ECC71 if s.won else 0xE74C3C
embed = discord.Embed(
title=f"{result} — {s.champion}",
description=f"**{fmt_queue(s.queue_id)}** · {fmt_duration(s.duration)}",
color=color,
)
embed.add_field(
name="⚔️ KDA", value=fmt_kda(s.kills, s.deaths, s.assists), inline=True
)
embed.add_field(name="🌾 CS", value=fmt_cs(s.cs, s.duration), inline=True)
embed.add_field(name="👁️ Vision", value=str(s.vision), inline=True)
embed.add_field(name="💥 Damage", value=fmt_damage(s.damage), inline=True)
embed.add_field(name="💰 Gold", value=fmt_gold(s.gold), inline=True)
embed.add_field(
name="🔮 Spells", value=fmt_spells(s.spell1_id, s.spell2_id), inline=True
)
embed.add_field(name="👤 Player", value=f"{s.game_name}#{s.tag_line}", inline=False)
embed.set_thumbnail(
url=f"https://ddragon.leagueoflegends.com/cdn/{s.patch}/img/champion/{s.champion}.png"
)
embed.set_footer(text=f"Match ID: {s.match_id}")
return embed
def build_compact_embed(s: MatchSummary) -> discord.Embed:
"""Minimal one-liner embed, good for showing many matches at once."""
result = "✅" if s.won else "❌"
color = 0x2ECC71 if s.won else 0xE74C3C
kda = fmt_kda(s.kills, s.deaths, s.assists)
cs = fmt_cs(s.cs, s.duration)
embed = discord.Embed(
description=(
f"{result} **{s.champion}** · {fmt_queue(s.queue_id)} · {fmt_duration(s.duration)}\n"
f"⚔️ {kda} · 🌾 {cs} · 💥 {fmt_damage(s.damage)}"
),
color=color,
)
embed.set_thumbnail(
url=f"https://ddragon.leagueoflegends.com/cdn/{s.patch}/img/champion/{s.champion}.png"
)
return embed
def build_match_list_embed(
summaries: list[MatchSummary], game_name: str, tag_line: str
) -> discord.Embed:
"""Single embed summarizing multiple matches at once."""
total = len(summaries)
wins = sum(1 for s in summaries if s.won)
losses = total - wins
wr = f"{round(wins / total * 100)}%" if total > 0 else "N/A"
avg_kda = sum((s.kills + s.assists) / max(s.deaths, 1) for s in summaries) / max(
total, 1
)
embed = discord.Embed(
title=f"📋 Last {total} matches — {game_name}#{tag_line}",
description=f"**{wins}W / {losses}L** · {wr} WR · {avg_kda:.2f} avg KDA",
color=0x1A78C2,
)
for s in summaries:
result = "✅" if s.won else "❌"
embed.add_field(
name=f"{result} {s.champion} — {fmt_queue(s.queue_id)}",
value=(
f"⚔️ {fmt_kda(s.kills, s.deaths, s.assists)} · "
f"🌾 {fmt_cs(s.cs, s.duration)} · "
f"⏱️ {fmt_duration(s.duration)}"
),
inline=False,
)
return embed
+96
View File
@@ -0,0 +1,96 @@
import aiohttp
import asyncio
import os
import json
from dotenv import load_dotenv
from db import save_match, get_match as get_cached_match
RIOT_API_KEY = os.getenv("LEAGUEAPI")
REGION = os.getenv("REGION", "na1")
CLUSTER_MAP = {
"na1": "americas",
"br1": "americas",
"la1": "americas",
"la2": "americas",
"euw1": "europe",
"eun1": "europe",
"tr1": "europe",
"ru": "europe",
"kr": "asia",
"jp1": "asia",
}
CLUSTER = CLUSTER_MAP.get(REGION, "americas")
async def riot_get(url: str, retries: int = 3) -> dict | list | None:
headers = {"X-Riot-Token": RIOT_API_KEY}
async with aiohttp.ClientSession() as session:
for attempt in range(retries):
async with session.get(url, headers=headers) as resp:
if resp.status == 200:
return await resp.json()
elif resp.status == 492:
wait = int(resp.headers.get("Retry-After", 5))
print(f"⚠️ Rate limited. Retrying in {wait}s...")
await asyncio.sleep(wait)
elif resp.status == 404:
return None
else:
print(f"❌ Error {resp.status} (attempt {attempt + 1}): {url}")
await asyncio.sleep(1)
return None
# ── Endpoints ──────────────────────────────────────────────────────────────────
async def get_account(game_name: str, tag_line: str) -> dict | None:
return await riot_get(
f"https://{CLUSTER}.api.riotgames.com"
f"/riot/account/v1/accounts/by-riot-id/{game_name}/{tag_line}"
)
async def get_summoner(puuid: str) -> dict | None:
return await riot_get(
f"https://{REGION}.api.riotgames.com"
f"/lol/summoner/v4/summoners/by-puuid/{puuid}"
)
async def get_rank(summoner_id: str) -> dict | None:
return await riot_get(
f"https://{REGION}.api.riotgames.com"
f"/lol/league/v4/entries/by-summoner/{summoner_id}"
)
async def get_match_ids(puuid: str, cont: int = 5) -> dict | None:
return await riot_get(
f"https://{CLUSTER}.api.riotgames.com"
f"/lol/match/v5/matches/by-puuid/{puuid}/ids?count={count}"
)
async def get_match(match_id: str) -> dict | None:
# check cache first
cached = get_cached_match(match_id)
if cached:
return json.loads(cached)
data = await riot_get(
f"https://{CLUSTER}.api.riotgames.com" f"/lol/match/v5/matches/{match_id}"
)
if data:
save_match(match_id, json.dumps(data))
return data
async def get_latest_patch() -> str:
async with aiohttp.ClientSession() as session:
async with session.get(
"https://ddragon.leagueoflegends.com/api/versions.json"
) as resp:
versions = await resp.json()
return versions[0] if versions else "14.8.1"