import discord, datetime, os, re, json from selenium import webdriver from bs4 import BeautifulSoup from discord.ext import commands, tasks, bridge from setup.logger import LOGGER from models.cs_models import * BASE_URL = "https://www.hltv.org" PAGE_FILE = os.path.join("data","hltv.html") MATCHES_FILE = os.path.join("data","matches.json") # Cache duration in seconds (6 hours) DATA_CACHE_DURATION = 3600 * 6 # Match age threshold in seconds (2 days) OLD_MATCH_DURATION = 3600 * 24 * 2 # Reaction emojis for match events EMOJI_CREATE_EVENT = "✅" # Create guild event for the match EMOJI_REMOVE_EVENT = "❌" # Remove guild event for the match class CSCog(commands.Cog): def __init__(self, bot:discord.Bot): self.bot = bot self.matches:list[Match] = [] self.message_match_map = {} # Maps message IDs to Match objects self.get_matches.start() def is_up_to_date(self) -> bool: """Check if PAGE_FILE and MATCHES_FILE are up to date.""" try: page_modified = datetime.datetime.fromtimestamp(os.stat(PAGE_FILE).st_ctime) matches_modified = datetime.datetime.fromtimestamp(os.stat(MATCHES_FILE).st_ctime) now = datetime.datetime.now() page_age = now.timestamp() - page_modified.timestamp() matches_age = now.timestamp() - matches_modified.timestamp() return page_age <= DATA_CACHE_DURATION and matches_age <= DATA_CACHE_DURATION except FileNotFoundError: return False def get_page(self) -> None: """Get the page from https://www.hltv.org/matches/ using Firefox webdriver and save to PAGE_FILE.""" driver = webdriver.Firefox() driver.get("https://www.hltv.org/matches/") html = driver.page_source driver.close() with open(PAGE_FILE, 'w') as f: f.write(html) def save(self) -> None: """Save self.matches to MATCHES_FILE.""" matches_data = [match.to_dict() for match in self.matches] with open(MATCHES_FILE, 'w') as f: json.dump(matches_data, f, indent=2) def remove_old_matches(self) -> None: """Remove matches that are older than OLD_MATCH_DURATION.""" now = datetime.datetime.now() filtered_matches = [] for match in self.matches: if match.date is None: # Keep matches with no date filtered_matches.append(match) else: match_age = now.timestamp() - match.date.timestamp() if match_age <= OLD_MATCH_DURATION: filtered_matches.append(match) self.matches = filtered_matches @tasks.loop(minutes=5) async def get_matches(self): if self.is_up_to_date(): # Load from MATCHES_FILE try: with open(MATCHES_FILE) as f: matches_data = json.load(f) self.matches = [Match.from_dict(match_data) for match_data in matches_data] self.remove_old_matches() except Exception as e: LOGGER.error(f"Error loading matches from cache: {e}") self.matches = [] else: # Fetch new data self.get_page() # Read and parse the page with open(PAGE_FILE) as f: html = f.read() page = BeautifulSoup(html) all_matches = [] for section in page.find_all("div", {"class": "matches-chronologically"}): all_matches.extend(section.find_all("div", {"class": "match-wrapper"})) self.matches = [] for match_event in all_matches: event = match_event.find("div", {"class": "match-event"}) event_name = event.attrs["data-event-headline"] # Extract event logo event_logo_container = event.find("div", {"class": "match-event-logo-container"}) event_logo_url = None if event_logo_container: logo_img = event_logo_container.find("img") if logo_img and logo_img.attrs.get("src"): event_logo_url = logo_img.attrs["src"] event_obj = Event(int(event.attrs["data-event-id"]), event_name, event_logo_url) href = str(match_event.find("a", {"class": "match-top"}).attrs["href"]) match_href = "https://hltv.org" + href match_time_elt = match_event.find("div", {"class": "match-time"}) if match_time_elt: match_timestamp = int(match_time_elt.attrs["data-unix"]) match_date = datetime.datetime.fromtimestamp(match_timestamp/1000) else: match_date = None match_format = match_event.find("div", {"class": "match-meta"}).text teams = match_event.find_all("div", {"class": "match-team"}) match_teams = [] for match_team in teams: team = match_team.find("div", {"class": "match-teamname"}) team_obj = Team(team.text, match_team.find("img").attrs["src"]) match_teams.append(team_obj) if len(match_teams) == 0: continue match_obj = Match(match_href, match_format, event_obj, match_teams, match_date) self.matches.append(match_obj) # Remove old matches and save to cache self.remove_old_matches() self.save() if self.matches: print(str(self.matches[0])) def check_event(self, match: Match, event: discord.ScheduledEvent) -> bool: """ Check if a scheduled event matches the given match. Returns True if both team names appear in the event name (case-insensitive). """ event_name_lower = event.name.lower() team1_name_lower = match.teams[0].name.lower() team2_name_lower = match.teams[1].name.lower() # Check if both team names are in the event name return team1_name_lower in event_name_lower and team2_name_lower in event_name_lower async def create_guild_event(self, guild: discord.Guild, match: Match) -> discord.ScheduledEvent | None: """ Create a scheduled guild event for the match. Returns the created event or None if it already exists or creation fails. """ try: # Check if event already exists for event in guild.scheduled_events: if self.check_event(match, event): return event # Create event name event_name = f"{match.teams[0].name} vs {match.teams[1].name}" # Set start and end times - use UTC timezone to match the embed display if match.date: # Convert the timestamp to UTC-aware datetime timestamp = int(match.date.timestamp()) start_time = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc) end_time = start_time + datetime.timedelta(hours=2) # Assume 2 hour match else: start_time = datetime.datetime.now(tz=datetime.timezone.utc) end_time = start_time + datetime.timedelta(hours=2) # Create the event event = await guild.create_scheduled_event( name=event_name, start_time=start_time, end_time=end_time, description=f"Event: {match.event.name}\nFormat: {match.format}\nWatch on HLTV: {match.url}", location="" ) LOGGER.info(f"Created guild event for {event_name} in {guild.name}") return event except Exception as e: LOGGER.error(f"Error creating guild event in {guild.name}: {e}") return None async def remove_guild_event(self, guild: discord.Guild, match: Match) -> bool: """ Remove the scheduled guild event for the match. Returns True if an event was removed, False otherwise. """ try: for event in guild.scheduled_events: if self.check_event(match, event): await event.delete() LOGGER.info(f"Removed guild event {event.name} from {guild.name}") return True return False except Exception as e: LOGGER.error(f"Error removing guild event from {guild.name}: {e}") return False def create_match_embed(self, match: Match) -> discord.Embed: """ Create a Discord embed message containing match information. Args: match: The Match object to display Returns: discord.Embed: A formatted embed with match details """ # Create embed with title as the matchup title = f"{match.teams[0].name} vs {match.teams[1].name}" embed = discord.Embed( title=title, description=f"**Event:** {match.event.name}", url=match.url, color=discord.Color.blue() ) # Add match format embed.add_field( name="Format", value=match.format, inline=True ) # Add match date if available if match.date: embed.add_field( name="Date & Time", value=f"", inline=True ) # Add team information with logos teams_info = f"[{match.teams[0].name}]({match.url})\n[{match.teams[1].name}]({match.url})" embed.add_field( name="Teams", value=teams_info, inline=False ) # Set thumbnail to event logo if available, otherwise use first team logo if match.event.logo_url: embed.set_thumbnail(url=match.event.logo_url) elif match.teams[0].logo_url: embed.set_thumbnail(url=match.teams[0].logo_url) # Set footer with HLTV link embed.set_footer( text="HLTV.org", icon_url="https://www.hltv.org/img/static/TopLogo2x.png" ) return embed @bridge.bridge_command() async def test_match_embed(self, ctx:bridge.BridgeExtContext): """Send the first match as an embed to the test channel.""" TEST_CHANNEL_ID = 1305147782071451718 # Check if there are matches available if not self.matches: await ctx.send("❌ No matches available. Try running the scraper first.") return # Get the first match first_match = self.matches[0] # Create the embed embed = self.create_match_embed(first_match) # Send to test channel try: channel = self.bot.get_channel(TEST_CHANNEL_ID) if channel is None: await ctx.send(f"❌ Could not find channel with ID {TEST_CHANNEL_ID}") return message = await channel.send(embed=embed) # Add reactions await message.add_reaction(EMOJI_CREATE_EVENT) await message.add_reaction(EMOJI_REMOVE_EVENT) # Track the message -> match mapping self.message_match_map[message.id] = first_match await ctx.reply(f"✅ Sent embed for match: {first_match.teams[0].name} vs {first_match.teams[1].name}") except Exception as e: LOGGER.error(f"Error sending embed: {e}") await ctx.send(f"❌ Error sending embed: {e}") @commands.Cog.listener() async def on_raw_reaction_add(self, payload: discord.RawReactionActionEvent): """Handle reactions added to match embed messages.""" # Ignore reactions from the bot itself if payload.user_id == self.bot.user.id: return # Check if this is a tracked message if payload.message_id not in self.message_match_map: return match = self.message_match_map[payload.message_id] emoji = payload.emoji.name # Get the guild guild = self.bot.get_guild(payload.guild_id) if guild is None: return # Get the message for replies message = await self.get_reaction_message(payload) if not message: return try: if emoji == EMOJI_CREATE_EVENT: # Create event event = await self.create_guild_event(guild, match) if event: await message.reply(f"✅ Created guild event: **{event.name}**") elif emoji == EMOJI_REMOVE_EVENT: # Remove event removed = await self.remove_guild_event(guild, match) if removed: await message.reply( f"✅ Removed guild event for {match.teams[0].name} vs {match.teams[1].name}" ) else: await message.reply( f"⚠️ No guild event found for this match" ) except Exception as e: LOGGER.error(f"Error handling reaction {emoji}: {e}") async def get_reaction_message(self, payload: discord.RawReactionActionEvent) -> discord.Message | None: """Get the message object from a raw reaction payload.""" try: channel = self.bot.get_channel(payload.channel_id) if channel: return await channel.fetch_message(payload.message_id) except Exception as e: LOGGER.error(f"Error fetching message: {e}") return None async def check_events(self): """Check scheduled events for matching CS matches.""" for match in self.matches: for guild in self.bot.guilds: events = [event for event in guild.scheduled_events if self.check_event(match, event)] # print(guild.name, match) # @commands.Cog.listener() # async def on_ready(self): # self.check_events.start()