thc-webhook/main.py

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 | None:
    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 = 24
    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 n 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"]
        return []
    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 in ["halftime", "notification"]:
            msg["image"] = {"url": get_tenor_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()