thc-webhook/main.py

import os
import pytz
from datetime import datetime
import time
import logging
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"

MSG_REMINDER = "This is your 5 minute reminder to 420!"
MSG_HALFTIME_REMINDER = "Half-time in 5 minutes!"
MSG_HALFTIME = "Half-time!"
MSG_NOTIFICATION = "Blaze it!"

COL_BLUE = 0x3498db
COL_ORANGE = 0xe67e22
COL_GREEN = 0x2ecc71
COL_UNKNOWN = 0x95a5a6

messages = {
    "reminder_halftime": {"text": MSG_HALFTIME_REMINDER, "color": COL_ORANGE},
    "halftime": {"text": MSG_HALFTIME, "color": COL_GREEN},
    "reminder": {"text": MSG_REMINDER, "color": COL_ORANGE},
    "420": {"text": MSG_NOTIFICATION, "color": COL_GREEN},
    "unknown": {"text": "Unknown notification type", "color": COL_UNKNOWN}
}


def load_timezones() -> list[dict]:
    """Load timezones from csv file."""
    # Read the CSV file and return a list of timezones
    with open("tzdb/TimeZoneDB.csv/time_zone.csv", "r", encoding="utf-8") as f:
        lines = f.readlines()
    # Fields: zone_name,country_code,abbreviation,time_start,gmt_offset,dst
    timezones = []
    for line in lines:
        fields = line.strip().split(",")
        if len(fields) >= 5:
            timezones.append({
                "zone_name": fields[0],
                "country_code": fields[1],
                "abbreviation": fields[2],
                "time_start": fields[3],
                "gmt_offset": int(fields[4]),
                "dst": fields[5] == '1'
            })
    return timezones


def load_countries() -> list[dict]:
    """Load countries from csv file."""
    # Read the CSV file and return a list of countries
    with open("tzdb/TimeZoneDB.csv/country.csv", "r", encoding="utf-8") as f:
        lines = f.readlines()
    # Fields: country_code,country_name
    countries = []
    for line in lines:
        fields = line.strip().split(",")
        if len(fields) >= 2:
            countries.append({
                "country_code": fields[0],
                "country_name": fields[1]
            })
    return countries


timezones = load_timezones()
countries = load_countries()


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", "420"]:
            msg["image"] = {
                "url": "https://www.freepnglogos.com/uploads/weed-leaf-png/cannabis-weed-leaf-png-clipart-images-24.png"}
        if type == "420":
            tz_list = where_is_it_420(timezones, countries)
            if tz_list:
                tz_str = "\n".join(tz_list)
                msg["text"] += f"\nIt's 4:20 in:\n{tz_str}"
            else:
                msg["text"] += " It's not 4:20 anywhere right now."
    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 get_tz_info(tz_name: str, timezones: list[dict]) -> dict | None:
    """Get timezone info by name."""
    return next((tz for tz in timezones if tz["zone_name"] == tz_name), None)


def get_country_info(country_code: str, countries: list[dict]) -> dict | None:
    """Get country info by country code."""
    return next((c for c in countries if c["country_code"] == country_code), None)


def where_is_it_420(timezones: list[dict], countries: list[dict]) -> list[str]:
    """Get timezones where the current hour is 4 or 16, indicating it's 4:20 there.

    Returns:
        list[str]: A list of timezones where it's currently 4:20 PM or AM.
    """
    tz_list = []
    for tz in pytz.all_timezones:
        now = datetime.now(pytz.timezone(tz))
        if now.hour == 4 or now.hour == 16:
            # Find the timezone in the loaded timezones
            tz_info = get_tz_info(tz, timezones)
            if tz_info:
                country = get_country_info(tz_info["country_code"], countries)
                if country and country["country_name"] not in tz_list:
                    country_name = country["country_name"].strip().strip('"')
                    tz_list.append(country_name)
    return tz_list


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, "420")
    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("420")

    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()