diff --git a/bot.py b/bot.py index 555f016..a367973 100644 --- a/bot.py +++ b/bot.py @@ -1,59 +1,182 @@ import discord -from discord import app_commands +from discord.ext import commands, tasks +import asyncio import os +import logging +from datetime import datetime from dotenv import load_dotenv load_dotenv() -Discord_Token = os.getenv("DISCORDTOKEN") -intents = discord.Intents.default() -client = discord.Client(intents=intents) -tree = app_commands.CommandTree(client) +# ── Logging ──────────────────────────────────────────────────────────────────── -# --command --------------------------------------------------------- +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") -@tree.command(name="ping", description="Test if bot is alive") -async def ping(interaction: discord.Interaction): - await interaction.response.send_message("Pong!", ephemeral=True) +# ── Bot setup ────────────────────────────────────────────────────────────────── + +EXTENSIONS = [ + "commands.login", + "commands.profile", + "commands.recent", +] -@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, - ) +class LoLBot(commands.Bot): + def __init__(self): + intents = discord.Intents.default() + intents.message_content = True + + super().__init__( + command_prefix="!", + intents=intents, + help_command=None, # disable default help + ) + + # ── Lifecycle ────────────────────────────────────────────────────────────── + + async def setup_hook(self): + """Called once before the bot connects — load extensions here.""" + for ext in EXTENSIONS: + try: + await self.load_extension(ext) + log.info(f"✅ Loaded extension: {ext}") + except Exception as e: + log.error(f"❌ Failed to load {ext}: {e}") + + await self.tree.sync() + log.info("🔄 Slash commands synced") + + 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: + await interaction.response.send_message(msg, ephemeral=True) + + 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() -@tree.command(name="recent", description="See your recent matches") -async def recent(interaction: discord.Interaction): - await interaction.response.send_message( - "Fetching your recent matches — coming soon!", ephemeral=True - ) +# ── Dev commands (owner only) ────────────────────────────────────────────────── + +bot = LoLBot() -# --StartUp ------------------------------------------------------- -@client.event -async def on_ready(): - server_id = os.getenv("SERVER_ID") - if server_id: - try: - guild = discord.Object(id=int(server_id)) - tree.copy_global_to(guild=guild) - await tree.sync(guild=guild) - print(f"Logged in as {client.user} (Synced to Guild ID: {server_id})") - except ValueError: - print(f"Error: SERVER_ID '{server_id}' is not a valid integer. Syncing globally instead.") - await tree.sync() - print(f"Logged in as {client.user} (Global Sync)") - else: - await tree.sync() - print(f"Logged in as {client.user} (Global Sync)") +@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") -if Discord_Token: - client.run(Discord_Token) -else: - print("Error: DISCORDTOKEN not found in .env file.") +@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()) diff --git a/commands/login.py b/commands/login.py new file mode 100644 index 0000000..f0706e5 --- /dev/null +++ b/commands/login.py @@ -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)) + diff --git a/commands/profile.py b/commands/profile.py new file mode 100644 index 0000000..bd08bb7 --- /dev/null +++ b/commands/profile.py @@ -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)) diff --git a/commands/recent.py b/commands/recent.py new file mode 100644 index 0000000..4cc62a4 --- /dev/null +++ b/commands/recent.py @@ -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)) diff --git a/database/db.py b/database/db.py new file mode 100644 index 0000000..a5553b1 --- /dev/null +++ b/database/db.py @@ -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() diff --git a/requirements.txt b/requirements.txt index 4304048..e84217f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ discord.py -python-doten +python-dotenv httpx aiosqlite +aiohttp diff --git a/services/match_formatter.py b/services/match_formatter.py new file mode 100644 index 0000000..e58f7c4 --- /dev/null +++ b/services/match_formatter.py @@ -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 diff --git a/services/riot_api.py b/services/riot_api.py new file mode 100644 index 0000000..f0fffe2 --- /dev/null +++ b/services/riot_api.py @@ -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"