feat: Add misandrist insults and big CS2 events matches fetch from HLTV.org
This commit is contained in:
375
cogs/cs.py
Normal file
375
cogs/cs.py
Normal file
@@ -0,0 +1,375 @@
|
||||
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"<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()
|
||||
|
||||
31
cogs/insult.py
Normal file
31
cogs/insult.py
Normal file
@@ -0,0 +1,31 @@
|
||||
import discord, json, os, random, re
|
||||
import collections
|
||||
from discord.ext import commands
|
||||
from setup.logger import LOGGER
|
||||
|
||||
class InsultsCog(commands.Cog):
|
||||
def __init__(self, bot:discord.Bot):
|
||||
self.bot = bot
|
||||
|
||||
with open(os.path.join("data", "insults.txt")) as f:
|
||||
self.insults = [l.strip() for l in f.readlines()]
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_message(self, message:discord.Message):
|
||||
if len(message.content.split()) > 3:
|
||||
author = message.author
|
||||
if any(author.id == u.id for u in self.men):
|
||||
r = int(random.random()*100)
|
||||
print(r)
|
||||
if r < 5:
|
||||
insult = random.choice(self.insults)
|
||||
await message.reply(insult)
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self):
|
||||
self.men:list[discord.Member] = []
|
||||
for guild in self.bot.guilds:
|
||||
for role in guild.roles:
|
||||
if re.match(r'^([hH]im|[hH]e).*$',role.name):
|
||||
self.men.extend(role.members)
|
||||
self.men = list(set(self.men))
|
||||
Reference in New Issue
Block a user