Mathéo Nebulo9 DESSAUVAGES
d14ca4ab3e
Some checks failed
CD / Deploy to VPS (push) Has been cancelled
392 lines
15 KiB
Python
392 lines
15 KiB
Python
import discord, datetime, os, re, json
|
|
from selenium import webdriver
|
|
from selenium.webdriver.firefox.options import Options
|
|
from selenium.webdriver.firefox.service import Service
|
|
from bs4 import BeautifulSoup
|
|
from discord.ext import commands, tasks, bridge
|
|
from setup.logger import LOGGER
|
|
from setup.config import Config
|
|
from models.cs_models import *
|
|
|
|
BASE_URL = "https://www.hltv.org"
|
|
|
|
PAGE_FILE = os.path.join(Config.CONFIG_PATH,"hltv.html")
|
|
|
|
MATCHES_FILE = os.path.join(Config.CONFIG_PATH,"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 a headless Firefox webdriver and save to PAGE_FILE."""
|
|
options = Options()
|
|
options.add_argument("--headless=new")
|
|
options.add_argument("--no-sandbox")
|
|
options.add_argument("--disable-dev-shm-usage")
|
|
options.add_argument("--window-size=1920,1080")
|
|
|
|
geckodriver_path = os.getenv("GECKODRIVER_PATH")
|
|
service = Service(executable_path=geckodriver_path) if geckodriver_path else None
|
|
|
|
driver = None
|
|
try:
|
|
driver = webdriver.Firefox(service=service, options=options)
|
|
driver.get("https://www.hltv.org/matches/")
|
|
html = driver.page_source
|
|
|
|
with open(PAGE_FILE, 'w', encoding='utf-8') as f:
|
|
f.write(html)
|
|
finally:
|
|
if driver is not None:
|
|
driver.quit()
|
|
|
|
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"<t:{int(match.date.timestamp())}: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()
|
|
|