import os import time import logging import random import requests import schedule from dotenv import load_dotenv # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) # Load environment variables load_dotenv() WEBHOOK_URL = os.getenv('DISCORD_WEBHOOK_URL') MSG_FOOTER = "THC - Toke Hash Coordinated | GIF via Tenor" MSG_TEST = "Discord 420 timer activated. This is a test notification." MSG_REMINDER = "This is your 5 minute reminder to 420!" MSG_HALFTIME_REMINDER = "Half-time in 5 minutes!" MSG_HALFTIME = "Half-time!" MSG_NOTIFICATION = "420! Blaze it!" COL_BLUE = 0x3498db COL_ORANGE = 0xe67e22 COL_GREEN = 0x2ecc71 COL_UNKNOWN = 0x95a5a6 messages = { "test": {"text": MSG_TEST, "color": COL_BLUE}, "reminder_halftime": {"text": MSG_HALFTIME_REMINDER, "color": COL_ORANGE}, "halftime": {"text": MSG_HALFTIME, "color": COL_GREEN}, "reminder": {"text": MSG_REMINDER, "color": COL_ORANGE}, "notification": {"text": MSG_NOTIFICATION, "color": COL_GREEN}, "unknown": {"text": "Unknown notification type", "color": COL_UNKNOWN} } def get_tenor_img(search_term: str = "420") -> str: results = [] try: results = search_tenor(search_term=search_term) except Exception as e: logging.error(f"Error fetching Tenor GIF: {e}") if results: # get a random image from the results r = random.choice(results) formats = r["media_formats"] # try to get the tinygif format tinygif = formats.get("tinygif", {}).get("url") if tinygif: return tinygif def search_tenor(search_term: str = "420") -> list: """ Fetch a random GIF URL from Tenor based on the provided slug. """ api_key = os.getenv('TENOR_API_KEY') if not api_key: logging.warning("TENOR_API_KEY not set") return [] lmt = 8 ckey = "thc-time" url = f"https://tenor.googleapis.com/v2/search?q={search_term}&key={api_key}&client_key={ckey}&limit={lmt}" # get the top 8 GIFs for the search term try: response = requests.get(url, timeout=10) response.raise_for_status() data = response.json() if data["results"]: return data["results"] except requests.RequestException as e: logging.error(f"Error fetching Tenor GIF: {e}") return [] def get_message(type: str) -> dict[str, int]: """ Get the notification message based on the type. """ msg = messages["unknown"] if type in messages: msg = messages[type] if type == "notification": img = get_tenor_img() else: img = get_tenor_img(search_term=type) if img: msg["image"] = {"url": img} return msg def create_embed(message: str) -> dict: """ Create a Discord embed message. """ msg = get_message(message) embed = { "title": message.replace("_", " ").capitalize(), "description": msg["text"], "image": msg.get("image"), "color": msg["color"], "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "footer": {"text": MSG_FOOTER} } return embed def send_notification(message: str) -> None: """ Send a notification to the Discord webhook. """ if not WEBHOOK_URL: logging.error("WEBHOOK_URL not set") return embed = create_embed(message) data = {"embeds": [embed]} try: response = requests.post(WEBHOOK_URL, json=data, timeout=10) if response.status_code == 204: logging.info(f"Notification sent: {message}") else: logging.error( f"Failed to send notification: {response.status_code} - " f"{response.text}" ) except requests.RequestException as e: logging.error(f"Error sending notification: {e}") def main() -> None: """ Main function to run the scheduler. """ # Schedule notifications schedule.every().hour.at(":15").do(send_notification, "reminder") schedule.every().hour.at(":20").do(send_notification, "notification") schedule.every().hour.at(":45").do(send_notification, "reminder_halftime") schedule.every().hour.at(":50").do(send_notification, "halftime") logging.info("Scheduler started.") # Test the notification on startup send_notification("test") try: while True: schedule.run_pending() time.sleep(1) # Check every second except KeyboardInterrupt: logging.info("Scheduler stopped by user") except Exception as e: logging.error(f"Unexpected error: {e}") if __name__ == "__main__": main()