112 lines
3.0 KiB
Python
112 lines
3.0 KiB
Python
import os
|
|
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_TEST = "Discord 420 timer activated. This is a test notification."
|
|
MSG_REMINDER = "5 minute reminder!"
|
|
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": {"text": MSG_REMINDER, "color": COL_ORANGE},
|
|
"halftime": {"text": MSG_HALFTIME, "color": COL_GREEN},
|
|
"notification": {"text": MSG_NOTIFICATION, "color": COL_GREEN},
|
|
"unknown": {"text": "Unknown notification type", "color": COL_UNKNOWN}
|
|
}
|
|
|
|
|
|
def get_message(type: str) -> dict[str, int]:
|
|
"""
|
|
Get the notification message based on the type.
|
|
"""
|
|
if type in messages:
|
|
return messages[type]
|
|
else:
|
|
return messages["unknown"]
|
|
|
|
|
|
def create_embed(message: str) -> dict:
|
|
"""
|
|
Create a Discord embed message.
|
|
"""
|
|
msg = get_message(message)
|
|
embed = {
|
|
"title": message.capitalize(),
|
|
"description": msg["text"],
|
|
"color": msg["color"],
|
|
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"footer": {
|
|
"text": "THC - Toke Hash Coordinated"
|
|
}
|
|
}
|
|
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")
|
|
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()
|