initial commit
This commit is contained in:
112
main.py
Normal file
112
main.py
Normal file
@@ -0,0 +1,112 @@
|
||||
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_NOTIFICATION = "420!"
|
||||
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},
|
||||
"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 send_notification(message: dict[str, int]) -> None:
|
||||
"""
|
||||
Send a notification to the Discord webhook as an embed.
|
||||
|
||||
Args:
|
||||
message (dict[str, int]): The message to send.
|
||||
"""
|
||||
if not WEBHOOK_URL:
|
||||
logging.error("WEBHOOK_URL not set")
|
||||
return
|
||||
|
||||
embed = {
|
||||
"title": "Notification",
|
||||
"description": message["text"],
|
||||
"color": message["color"],
|
||||
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"footer": {
|
||||
"text": "Discord Webhook App"
|
||||
}
|
||||
}
|
||||
|
||||
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 test_notification() -> None:
|
||||
"""
|
||||
Send a test notification to verify the webhook.
|
||||
"""
|
||||
send_notification(get_message("test"))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""
|
||||
Main function to run the scheduler.
|
||||
"""
|
||||
# Schedule notifications
|
||||
schedule.every().hour.at(":15").do(send_notification, get_message("reminder"))
|
||||
schedule.every().hour.at(":20").do(send_notification, get_message("notification"))
|
||||
|
||||
logging.info(
|
||||
"Scheduler started. Notifications will be sent every hour at :15 and :20")
|
||||
|
||||
# Test the notification on startup
|
||||
test_notification()
|
||||
|
||||
try:
|
||||
while True:
|
||||
schedule.run_pending()
|
||||
time.sleep(60) # Check every minute
|
||||
except KeyboardInterrupt:
|
||||
logging.info("Scheduler stopped by user")
|
||||
except Exception as e:
|
||||
logging.error(f"Unexpected error: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user