zwitschi/thc-webhook
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: zwitschi/thc-webhook:3f5630da2cac28b2896934452d2230c68a927a13
Branches Tags
zwitschi/thc-webhook:main
..
compare: zwitschi/thc-webhook:a4c92470b6317d8ba1f15ec4213de6504500d41c
Branches Tags
zwitschi/thc-webhook:main

2 Commits
3f5630da2c .. a4c92470b6

Author SHA1 Message Date
zwitschi a4c92470b6 feat: add maintenance module for message deletion and related utilities
Build and Deploy Docker Container / test (push) Successful in 37s
Details
Build and Deploy Docker Container / build-and-deploy (push) Failing after 1m3s
Details 
feat: implement tests for maintenance functions and dashboard interactions
feat: add timezone and country tests for thctime module
 2026-05-10 12:36:02 +02:00
zwitschi 01d94376d4 feat: implement timezone and country data handling in thctime module 2026-05-10 12:20:40 +02:00
9 changed files with 867 additions and 670 deletions
Expand all files Collapse all files
main.py
+14 -535
View File
@@ -1,16 +1,25 @@
import os import os
import pytz
from datetime import datetime, timedelta from datetime import datetime, timedelta
import time import time
import logging import logging
import threading import threading
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import requests import requests
import schedule import schedule
from dotenv import load_dotenv from dotenv import load_dotenv
from templates import load_templates from templates import load_templates
from dashboard import create_app from dashboard import create_app
from maintenance import delete_old_messages
from thctime import (
get_country_info,
get_tz_info,
get_tzdb_cache,
init_tzdb_cache,
load_countries,
load_timezones,
split_tz_name,
where_is_it_420,
)
SCHEDULED_NOTIFICATIONS = [ SCHEDULED_NOTIFICATIONS = [
(15, "reminder"), (15, "reminder"),
@@ -19,10 +28,6 @@ SCHEDULED_NOTIFICATIONS = [
(50, "halftime"), (50, "halftime"),
] ]
TZDB_CACHE: dict | None = None
STATE_LOCK = threading.Lock() STATE_LOCK = threading.Lock()
STATE: dict = { STATE: dict = {
"running": True, "running": True,
@@ -45,9 +50,6 @@ logging.basicConfig(
load_dotenv() load_dotenv()
WEBHOOK_URL = os.getenv('DISCORD_WEBHOOK_URL') WEBHOOK_URL = os.getenv('DISCORD_WEBHOOK_URL')
DISCORD_BOT_TOKEN = os.getenv('DISCORD_BOT_TOKEN')
DISCORD_CHANNEL_ID = os.getenv('DISCORD_CHANNEL_ID')
GUILD_ID = os.getenv('DISCORD_GUILD_ID')
def get_state() -> dict: def get_state() -> dict:
@@ -59,88 +61,6 @@ def get_next_event() -> dict:
return get_next_scheduled_event() return get_next_scheduled_event()
def init_tzdb_cache() -> dict:
"""Initialize a cached lookup structure for tzdb data.
This keeps the hourly scheduler fast by:
- Building O(1) maps (zone_name -> country_code, country_code -> country_name)
- Precomputing a list of tz names that exist in both tzdb CSVs and `pytz`
Note: this is intentionally NOT run at import time so tests can monkeypatch
`load_timezones`/`load_countries` without needing to reset global state.
"""
global TZDB_CACHE
if TZDB_CACHE is not None:
return TZDB_CACHE
timezones = load_timezones()
countries = load_countries()
tz_to_country_code: dict[str, str] = {}
tz_meta: dict[str, dict] = {}
for tz in timezones:
zone_name = tz.get("zone_name")
country_code = tz.get("country_code")
if not isinstance(zone_name, str) or not zone_name:
continue
if not isinstance(country_code, str) or not country_code:
continue
tz_to_country_code[zone_name] = country_code
region, city = split_tz_name(zone_name)
tz_meta[zone_name] = {
"zone_name": zone_name,
"country_code": country_code,
"region": region,
"city": city,
}
country_code_to_name: dict[str, str] = {}
for c in countries:
code = c.get("country_code")
name = c.get("country_name")
if code and name:
country_code_to_name[code] = str(name).strip().strip('"')
# Attach resolved country names onto tz_meta (storage-only for now).
for zone_name, meta in tz_meta.items():
code = meta.get("country_code")
if isinstance(code, str):
meta["country_name"] = country_code_to_name.get(code)
# Vetted tz list: only names that are present in tzdb and loadable by zoneinfo.
# Installing the `tzdata` package keeps this mapping up-to-date.
tz_names: list[str] = []
for zone_name in tz_to_country_code.keys():
try:
ZoneInfo(zone_name)
except ZoneInfoNotFoundError:
continue
tz_names.append(zone_name)
TZDB_CACHE = {
"tz_to_country_code": tz_to_country_code,
"country_code_to_name": country_code_to_name,
"tz_names": tz_names,
"tz_meta": tz_meta,
}
return TZDB_CACHE
def split_tz_name(zone_name: str) -> tuple[str, str]:
"""Split an IANA timezone name into (region, city).
Examples:
- "America/New_York" -> ("America", "New_York")
- "America/Argentina/Buenos_Aires" -> ("America", "Argentina/Buenos_Aires")
- "UTC" -> ("UTC", "")
"""
if "/" not in zone_name:
return zone_name, ""
region, rest = zone_name.split("/", 1)
return region, rest
def _update_state(**updates) -> None: def _update_state(**updates) -> None:
with STATE_LOCK: with STATE_LOCK:
STATE.update(updates) STATE.update(updates)
@@ -172,51 +92,13 @@ def get_next_scheduled_event(now: datetime | None = None) -> dict:
return {"at": next_dt, "type": next_type} return {"at": next_dt, "type": next_type}
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
def create_embed(type: str, tz_list: list[str] | None = None) -> dict: def create_embed(type: str, tz_list: list[str] | None = None) -> dict:
""" """
Create a Discord embed message. Create a Discord embed message.
""" """
templates_path = os.getenv("TEMPLATES_PATH", "templates.json") templates_path = os.getenv("TEMPLATES_PATH", "templates.json")
messages = load_templates(templates_path) messages = load_templates(templates_path)
cache = TZDB_CACHE cache = get_tzdb_cache()
timezones = load_timezones() if cache is None else [] timezones = load_timezones() if cache is None else []
countries = load_countries() if cache is None else [] countries = load_countries() if cache is None else []
if type in messages: if type in messages:
@@ -253,341 +135,6 @@ def create_embed(type: str, tz_list: list[str] | None = None) -> dict:
return embed return embed
def get_discord_headers() -> dict[str, str]:
return {
"Authorization": f"Bot {DISCORD_BOT_TOKEN}",
"Content-Type": "application/json",
}
def parse_message_timestamp(message: dict) -> datetime:
return datetime.fromisoformat(message["timestamp"].replace("Z", ""))
def build_delete_entry(message: dict) -> dict:
return {
"id": message.get("id"),
"timestamp": parse_message_timestamp(message),
}
def parse_float(value: str | int | float | None) -> float | None:
if value is None:
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def should_delete_message(
message: dict,
webhook_id: str,
author_id: str,
cutoff: int,
) -> bool:
message_timestamp = int(parse_message_timestamp(message).timestamp())
return (
message_timestamp <= cutoff
and message.get("webhook_id") == webhook_id
and message.get("author", {}).get("id") == author_id
)
def get_rate_limit_retry_after(response: requests.Response) -> float | None:
header_retry_after = parse_float(response.headers.get("Retry-After"))
if header_retry_after is not None:
return header_retry_after
reset_after = parse_float(response.headers.get("X-RateLimit-Reset-After"))
if reset_after is not None:
return reset_after
try:
payload = response.json()
except ValueError:
return None
return parse_float(payload.get("retry_after"))
def get_bucket_exhausted_delay(response: requests.Response) -> float | None:
remaining = response.headers.get("X-RateLimit-Remaining")
if remaining != "0":
return None
return parse_float(response.headers.get("X-RateLimit-Reset-After"))
def sleep_for_rate_limit(delay_seconds: float, reason: str) -> None:
if delay_seconds <= 0:
return
logging.info(
f"Waiting {delay_seconds:.3f}s for Discord rate limit reset ({reason}).")
time.sleep(delay_seconds)
def find_last_message_by_author(
headers: dict[str, str],
guild_id: str,
channel_id: str,
author_id: str,
) -> dict | None:
"""Find the newest indexed message for the author in the target guild channel."""
url = f"https://discord.com/api/v10/guilds/{guild_id}/messages/search"
params = [
("author_id", author_id),
("author_type", "webhook"),
("channel_id", channel_id),
("sort_by", "timestamp"),
("sort_order", "desc"),
("limit", "10"),
]
for _ in range(3):
try:
response = requests.get(
url, headers=headers, params=params, timeout=10)
except requests.RequestException as e:
logging.error(f"Error searching guild messages: {e}")
return None
if response.status_code == 202:
payload = response.json()
retry_after = float(payload.get("retry_after", 1) or 1)
logging.info(
f"Guild search index not ready. Retrying after {retry_after} seconds."
)
time.sleep(retry_after)
continue
if response.status_code == 429:
retry_after = get_rate_limit_retry_after(response) or 1.0
sleep_for_rate_limit(retry_after, "guild search")
continue
if response.status_code != 200:
logging.error(
f"Failed to search guild messages: {response.status_code} - {response.text}"
)
return None
message_groups = response.json().get("messages", [])
if not message_groups or not message_groups[0]:
return None
return message_groups[0][0]
logging.error("Guild search index did not become available in time.")
return None
def fetch_messages_to_delete(headers: dict, channel_id: str, webhook_id: str, author_id: str, cutoff: int, last_message_id: str | None = None) -> tuple[list[dict], str | None]:
"""
Fetch messages from the channel that are older than the cutoff timestamp and sent by the webhook.
Uses pagination with the 'before' parameter to resume from the last processed message.
Returns a tuple of (list of messages to delete, last message ID for pagination).
"""
url = f"https://discord.com/api/v10/channels/{channel_id}/messages"
params: dict[str, str | int] = {
"limit": 100, # Maximum allowed by Discord API
}
# Use the 'before' parameter for pagination if last_message_id is provided
if last_message_id:
params["before"] = last_message_id
try:
for _ in range(3):
response = requests.get(url, headers=headers,
params=params, timeout=10)
if response.status_code == 429:
retry_after = get_rate_limit_retry_after(response) or 1.0
sleep_for_rate_limit(retry_after, "channel message fetch")
continue
break
else:
logging.error(
"Failed to fetch messages after repeated rate limits.")
return [], last_message_id
if response.status_code == 200:
messages = response.json()
delete_list = []
new_last_message_id = None
for message in messages:
new_last_message_id = message.get("id")
if should_delete_message(
message,
webhook_id,
author_id,
cutoff,
):
delete_list.append(build_delete_entry(message))
# Limit the list to 100 items
if len(delete_list) >= 100:
break
return delete_list, new_last_message_id
else:
logging.error(
f"Failed to fetch messages: {response.status_code} - {response.text}")
return [], last_message_id
except requests.RequestException as e:
logging.error(f"Error fetching messages: {e}")
return [], last_message_id
def delete_message(headers: dict, channel_id: str, message_id: str) -> tuple[bool, float | None, bool]:
"""
Delete a single message from the channel.
Returns:
tuple[bool, float | None, bool]:
- whether the delete succeeded
- how long to wait before the next request, if any
- whether to abort the batch because further requests would be invalid
"""
delete_url = f"https://discord.com/api/v10/channels/{channel_id}/messages/{message_id}"
delete_response = requests.delete(delete_url, headers=headers, timeout=10)
if delete_response.status_code == 204:
return True, get_bucket_exhausted_delay(delete_response), False
if delete_response.status_code == 429:
retry_after = get_rate_limit_retry_after(delete_response) or 1.0
scope = delete_response.headers.get("X-RateLimit-Scope", "unknown")
is_global = delete_response.headers.get(
"X-RateLimit-Global", "false").lower() == "true"
logging.warning(
"Discord rate limit hit while deleting message %s: scope=%s global=%s retry_after=%.3fs",
message_id,
scope,
is_global,
retry_after,
)
return False, retry_after, False
if delete_response.status_code in {401, 403}:
logging.error(
"Failed to delete message %s: %s - %s. Stopping deletes to avoid invalid request spam.",
message_id,
delete_response.status_code,
delete_response.text,
)
return False, None, True
logging.error(
f"Failed to delete message {message_id}: {delete_response.status_code} - {delete_response.text}")
return False, None, False
def delete_old_messages(minutes: int = 6) -> None:
"""
Delete all messages sent by the webhook in the last `minutes` minutes.
Uses a dynamic slowdown to avoid hitting Discord API rate limits and pagination to fetch all messages.
"""
if not DISCORD_BOT_TOKEN or not DISCORD_CHANNEL_ID or not GUILD_ID:
logging.error(
"DISCORD_BOT_TOKEN, DISCORD_CHANNEL_ID, or DISCORD_GUILD_ID not set")
return
headers = get_discord_headers()
# Calculate the time `minutes` minutes ago
cutoff_timestamp = datetime.now() - timedelta(minutes=minutes)
cutoff = int(cutoff_timestamp.timestamp())
webhook_id = '1413817504194760766'
author_id = '1413817504194760766'
last_author_message = find_last_message_by_author(
headers,
GUILD_ID,
DISCORD_CHANNEL_ID,
author_id,
)
if last_author_message is None:
logging.info("No indexed messages found for the target author.")
return
last_message_id = last_author_message.get("id")
if not last_message_id:
logging.info("Search result did not contain a message id.")
return
deleted_count = 0
if should_delete_message(
last_author_message,
webhook_id,
author_id,
cutoff,
):
anchor_message = build_delete_entry(last_author_message)
deleted, wait_seconds, abort_batch = delete_message(
headers,
DISCORD_CHANNEL_ID,
anchor_message["id"],
)
if deleted:
deleted_count += 1
logging.info(
f"Deleted message {anchor_message['id']} from {anchor_message['timestamp'].isoformat()}"
)
elif abort_batch:
return
if wait_seconds is not None:
sleep_for_rate_limit(wait_seconds, "delete bucket")
while True:
# Fetch messages to delete with pagination
delete_list, next_last_message_id = fetch_messages_to_delete(
headers, DISCORD_CHANNEL_ID, webhook_id, author_id, cutoff, last_message_id
)
# Stop scanning when a page has no eligible messages to delete.
# Continuing pagination here can walk indefinitely through history.
if not delete_list:
if deleted_count == 0:
logging.info("No messages to delete.")
else:
logging.info("No more messages to delete.")
break
for message in delete_list:
message_id = message["id"]
message_time = message["timestamp"]
deleted, wait_seconds, abort_batch = delete_message(
headers,
DISCORD_CHANNEL_ID,
message_id,
)
if deleted:
deleted_count += 1
logging.info(
f"Deleted message {message_id} from {message_time.isoformat()}")
elif abort_batch:
logging.info(
"Stopping delete batch after an invalid Discord response.")
return
if wait_seconds is not None:
sleep_for_rate_limit(wait_seconds, "delete bucket")
if next_last_message_id is None or next_last_message_id == last_message_id:
break
last_message_id = next_last_message_id
logging.info(
f"Deleted {deleted_count} messages older than {minutes} minutes sent by the webhook.")
def send_notification(message: str) -> None: def send_notification(message: str) -> None:
""" """
Send a notification to the Discord webhook. Send a notification to the Discord webhook.
@@ -613,7 +160,7 @@ def send_notification(message: str) -> None:
tz_list: list[str] | None = None tz_list: list[str] | None = None
if message == "420": if message == "420":
try: try:
cache = TZDB_CACHE cache = get_tzdb_cache()
if cache is None: if cache is None:
tz_list = where_is_it_420(load_timezones(), load_countries()) tz_list = where_is_it_420(load_timezones(), load_countries())
else: else:
@@ -624,7 +171,7 @@ def send_notification(message: str) -> None:
tz_to_country_code=cache.get("tz_to_country_code"), tz_to_country_code=cache.get("tz_to_country_code"),
country_code_to_name=cache.get("country_code_to_name"), country_code_to_name=cache.get("country_code_to_name"),
) )
_update_state(last_locations=tz_list) _update_state(last_locations=tz_list or [])
except Exception as e: except Exception as e:
_update_state(last_locations=[], last_error=str(e)) _update_state(last_locations=[], last_error=str(e))
@@ -648,74 +195,6 @@ def send_notification(message: str) -> None:
_update_state(last_error=str(e)) _update_state(last_error=str(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],
tz_names: list[str] | None = None,
tz_to_country_code: dict[str, str] | None = None,
country_code_to_name: dict[str, str] | None = None,
) -> 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.
"""
# Build fast lookup dicts if not provided.
if tz_to_country_code is None:
tz_to_country_code = {}
for tz in timezones:
zone_name = tz.get("zone_name")
country_code = tz.get("country_code")
if isinstance(zone_name, str) and isinstance(country_code, str):
tz_to_country_code[zone_name] = country_code
if country_code_to_name is None:
country_code_to_name = {}
for c in countries:
code = c.get("country_code")
name = c.get("country_name")
if isinstance(code, str) and name is not None:
country_code_to_name[code] = str(name).strip().strip('"')
names_to_check = tz_names if tz_names is not None else pytz.all_timezones
results: list[str] = []
seen: set[str] = set()
for tz_name in names_to_check:
try:
tz_obj = pytz.timezone(tz_name)
except Exception:
continue
now = datetime.now(tz_obj)
if now.hour != 4 and now.hour != 16:
continue
country_code = tz_to_country_code.get(tz_name)
if not country_code:
continue
country_name = country_code_to_name.get(country_code)
if not country_name:
continue
if country_name in seen:
continue
seen.add(country_name)
results.append(country_name)
return results
def schedule_notification(interval: str, at: str, type: str) -> None: def schedule_notification(interval: str, at: str, type: str) -> None:
"""Example: schedule.every().hour.at(":15").do(send_notification, "reminder")""" """Example: schedule.every().hour.at(":15").do(send_notification, "reminder")"""
if interval == "hour": if interval == "hour":
maintenance.py
+343
View File
@@ -0,0 +1,343 @@
import logging
import os
import time
from datetime import datetime, timedelta
import requests
WEBHOOK_AUTHOR_ID = "1413817504194760766"
def get_discord_headers() -> dict[str, str]:
token = os.getenv("DISCORD_BOT_TOKEN")
return {
"Authorization": f"Bot {token}",
"Content-Type": "application/json",
}
def parse_message_timestamp(message: dict) -> datetime:
return datetime.fromisoformat(message["timestamp"].replace("Z", ""))
def build_delete_entry(message: dict) -> dict:
return {
"id": message.get("id"),
"timestamp": parse_message_timestamp(message),
}
def parse_float(value: str | int | float | None) -> float | None:
if value is None:
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def should_delete_message(
message: dict,
webhook_id: str,
author_id: str,
cutoff: int,
) -> bool:
message_timestamp = int(parse_message_timestamp(message).timestamp())
return (
message_timestamp <= cutoff
and message.get("webhook_id") == webhook_id
and message.get("author", {}).get("id") == author_id
)
def get_rate_limit_retry_after(response: requests.Response) -> float | None:
header_retry_after = parse_float(response.headers.get("Retry-After"))
if header_retry_after is not None:
return header_retry_after
reset_after = parse_float(response.headers.get("X-RateLimit-Reset-After"))
if reset_after is not None:
return reset_after
try:
payload = response.json()
except ValueError:
return None
return parse_float(payload.get("retry_after"))
def get_bucket_exhausted_delay(response: requests.Response) -> float | None:
remaining = response.headers.get("X-RateLimit-Remaining")
if remaining != "0":
return None
return parse_float(response.headers.get("X-RateLimit-Reset-After"))
def sleep_for_rate_limit(delay_seconds: float, reason: str) -> None:
if delay_seconds <= 0:
return
logging.info(
f"Waiting {delay_seconds:.3f}s for Discord rate limit reset ({reason}).")
time.sleep(delay_seconds)
def find_last_message_by_author(
headers: dict[str, str],
guild_id: str,
channel_id: str,
author_id: str,
) -> dict | None:
"""Find the newest indexed message for the author in the target guild channel."""
url = f"https://discord.com/api/v10/guilds/{guild_id}/messages/search"
params = [
("author_id", author_id),
("author_type", "webhook"),
("channel_id", channel_id),
("sort_by", "timestamp"),
("sort_order", "desc"),
("limit", "10"),
]
for _ in range(3):
try:
response = requests.get(
url, headers=headers, params=params, timeout=10)
except requests.RequestException as e:
logging.error(f"Error searching guild messages: {e}")
return None
if response.status_code == 202:
payload = response.json()
retry_after = float(payload.get("retry_after", 1) or 1)
logging.info(
f"Guild search index not ready. Retrying after {retry_after} seconds."
)
time.sleep(retry_after)
continue
if response.status_code == 429:
retry_after = get_rate_limit_retry_after(response) or 1.0
sleep_for_rate_limit(retry_after, "guild search")
continue
if response.status_code != 200:
logging.error(
f"Failed to search guild messages: {response.status_code} - {response.text}"
)
return None
message_groups = response.json().get("messages", [])
if not message_groups or not message_groups[0]:
return None
return message_groups[0][0]
logging.error("Guild search index did not become available in time.")
return None
def fetch_messages_to_delete(headers: dict, channel_id: str, webhook_id: str, author_id: str, cutoff: int, last_message_id: str | None = None) -> tuple[list[dict], str | None]:
"""
Fetch messages from the channel that are older than the cutoff timestamp and sent by the webhook.
Uses pagination with the 'before' parameter to resume from the last processed message.
Returns a tuple of (list of messages to delete, last message ID for pagination).
"""
url = f"https://discord.com/api/v10/channels/{channel_id}/messages"
params: dict[str, str | int] = {
"limit": 100,
}
if last_message_id:
params["before"] = last_message_id
try:
for _ in range(3):
response = requests.get(url, headers=headers,
params=params, timeout=10)
if response.status_code == 429:
retry_after = get_rate_limit_retry_after(response) or 1.0
sleep_for_rate_limit(retry_after, "channel message fetch")
continue
break
else:
logging.error(
"Failed to fetch messages after repeated rate limits.")
return [], last_message_id
if response.status_code == 200:
messages = response.json()
delete_list = []
new_last_message_id = None
for message in messages:
new_last_message_id = message.get("id")
if should_delete_message(
message,
webhook_id,
author_id,
cutoff,
):
delete_list.append(build_delete_entry(message))
if len(delete_list) >= 100:
break
return delete_list, new_last_message_id
logging.error(
f"Failed to fetch messages: {response.status_code} - {response.text}")
return [], last_message_id
except requests.RequestException as e:
logging.error(f"Error fetching messages: {e}")
return [], last_message_id
def delete_message(headers: dict, channel_id: str, message_id: str) -> tuple[bool, float | None, bool]:
"""
Delete a single message from the channel.
Returns:
tuple[bool, float | None, bool]:
- whether the delete succeeded
- how long to wait before the next request, if any
- whether to abort the batch because further requests would be invalid
"""
delete_url = f"https://discord.com/api/v10/channels/{channel_id}/messages/{message_id}"
delete_response = requests.delete(delete_url, headers=headers, timeout=10)
if delete_response.status_code == 204:
return True, get_bucket_exhausted_delay(delete_response), False
if delete_response.status_code == 429:
retry_after = get_rate_limit_retry_after(delete_response) or 1.0
scope = delete_response.headers.get("X-RateLimit-Scope", "unknown")
is_global = delete_response.headers.get(
"X-RateLimit-Global", "false").lower() == "true"
logging.warning(
"Discord rate limit hit while deleting message %s: scope=%s global=%s retry_after=%.3fs",
message_id,
scope,
is_global,
retry_after,
)
return False, retry_after, False
if delete_response.status_code in {401, 403}:
logging.error(
"Failed to delete message %s: %s - %s. Stopping deletes to avoid invalid request spam.",
message_id,
delete_response.status_code,
delete_response.text,
)
return False, None, True
logging.error(
f"Failed to delete message {message_id}: {delete_response.status_code} - {delete_response.text}")
return False, None, False
def delete_old_messages(minutes: int = 6) -> None:
"""
Delete all messages sent by the webhook in the last `minutes` minutes.
Uses a dynamic slowdown to avoid hitting Discord API rate limits and pagination to fetch all messages.
"""
discord_bot_token = os.getenv("DISCORD_BOT_TOKEN")
discord_channel_id = os.getenv("DISCORD_CHANNEL_ID")
guild_id = os.getenv("DISCORD_GUILD_ID")
if not discord_bot_token or not discord_channel_id or not guild_id:
logging.error(
"DISCORD_BOT_TOKEN, DISCORD_CHANNEL_ID, or DISCORD_GUILD_ID not set")
return
headers = get_discord_headers()
cutoff_timestamp = datetime.now() - timedelta(minutes=minutes)
cutoff = int(cutoff_timestamp.timestamp())
webhook_id = WEBHOOK_AUTHOR_ID
author_id = WEBHOOK_AUTHOR_ID
last_author_message = find_last_message_by_author(
headers,
guild_id,
discord_channel_id,
author_id,
)
if last_author_message is None:
logging.info("No indexed messages found for the target author.")
return
last_message_id = last_author_message.get("id")
if not last_message_id:
logging.info("Search result did not contain a message id.")
return
deleted_count = 0
if should_delete_message(
last_author_message,
webhook_id,
author_id,
cutoff,
):
anchor_message = build_delete_entry(last_author_message)
deleted, wait_seconds, abort_batch = delete_message(
headers,
discord_channel_id,
anchor_message["id"],
)
if deleted:
deleted_count += 1
logging.info(
f"Deleted message {anchor_message['id']} from {anchor_message['timestamp'].isoformat()}"
)
elif abort_batch:
return
if wait_seconds is not None:
sleep_for_rate_limit(wait_seconds, "delete bucket")
while True:
delete_list, next_last_message_id = fetch_messages_to_delete(
headers, discord_channel_id, webhook_id, author_id, cutoff, last_message_id
)
if not delete_list:
if deleted_count == 0:
logging.info("No messages to delete.")
else:
logging.info("No more messages to delete.")
break
for message in delete_list:
message_id = message["id"]
message_time = message["timestamp"]
deleted, wait_seconds, abort_batch = delete_message(
headers,
discord_channel_id,
message_id,
)
if deleted:
deleted_count += 1
logging.info(
f"Deleted message {message_id} from {message_time.isoformat()}")
elif abort_batch:
logging.info(
"Stopping delete batch after an invalid Discord response.")
return
if wait_seconds is not None:
sleep_for_rate_limit(wait_seconds, "delete bucket")
if next_last_message_id is None or next_last_message_id == last_message_id:
break
last_message_id = next_last_message_id
logging.info(
f"Deleted {deleted_count} messages older than {minutes} minutes sent by the webhook.")
tests/test_dashboard.py
+143
View File
@@ -0,0 +1,143 @@
from datetime import datetime
import dashboard
def _make_app():
return dashboard.create_app(
get_state=lambda: {
"running": True,
"started_at": datetime(2026, 1, 1, 10, 0, 0),
"last_type": "420",
"last_attempt_at": datetime(2026, 1, 1, 10, 15, 0),
"last_success_at": datetime(2026, 1, 1, 10, 20, 0),
"last_status_code": 204,
"last_error": None,
"last_locations": ["Nowhere"],
},
get_next_event=lambda: {
"type": "reminder",
"at": datetime(2026, 1, 1, 11, 15, 0),
},
)
def test_fmt_dt_none_and_datetime():
assert dashboard._fmt_dt(None) == "—"
assert dashboard._fmt_dt(
datetime(2026, 1, 1, 10, 0, 0)) == "2026-01-01T10:00:00"
def test_get_html_template_wraps_content():
html = dashboard.get_html_template("<p>hello</p>")
assert "<title>thc-webhook admin</title>" in html
assert "<p>hello</p>" in html
def test_index_route_renders_status_page():
app = _make_app()
client = app.test_client()
response = client.get("/")
assert response.status_code == 200
body = response.get_data(as_text=True)
assert "thc-webhook" in body
assert "last_type: 420" in body
assert "type: reminder" in body
assert "Nowhere" in body
def test_admin_get_renders_template_form(monkeypatch):
monkeypatch.setattr(
dashboard,
"load_templates",
lambda path: {
"420": {
"text": "Blaze",
"color": 3066993,
"image_url": "https://example.com/img.png",
}
},
)
app = _make_app()
app.config["TEMPLATES_PATH"] = "templates.json"
client = app.test_client()
response = client.get("/admin")
assert response.status_code == 200
body = response.get_data(as_text=True)
assert "Admin: templates" in body
assert "name='420__text'" in body
assert "name='420__color'" in body
assert "name='420__image_url'" in body
def test_admin_post_validation_error(monkeypatch):
monkeypatch.setattr(
dashboard,
"load_templates",
lambda path: {"420": {"text": "x", "color": 1}},
)
monkeypatch.setattr(dashboard, "parse_color", lambda raw: (
_ for _ in ()).throw(ValueError("bad color")))
save_called = {"value": False}
def _save_templates(path, updated):
save_called["value"] = True
monkeypatch.setattr(dashboard, "save_templates", _save_templates)
app = _make_app()
client = app.test_client()
response = client.post(
"/admin",
data={
"420__text": "Updated",
"420__color": "bad",
"420__image_url": "",
},
)
assert response.status_code == 400
assert "invalid color" in response.get_data(as_text=True)
assert save_called["value"] is False
def test_admin_post_success(monkeypatch):
monkeypatch.setattr(
dashboard,
"load_templates",
lambda path: {"420": {"text": "x", "color": 1}},
)
monkeypatch.setattr(dashboard, "parse_color", lambda raw: 123)
saved = {"path": None, "payload": None}
def _save_templates(path, updated):
saved["path"] = path
saved["payload"] = updated
monkeypatch.setattr(dashboard, "save_templates", _save_templates)
app = _make_app()
app.config["TEMPLATES_PATH"] = "custom_templates.json"
client = app.test_client()
response = client.post(
"/admin",
data={
"420__text": "Updated",
"420__color": "123",
"420__image_url": "",
},
)
assert response.status_code == 200
assert "Saved." in response.get_data(as_text=True)
assert saved["path"] == "custom_templates.json"
assert saved["payload"] == {"420": {"text": "Updated", "color": 123}}
tests/test_main.py
-135
View File
@@ -1,135 +0,0 @@
import io
import time
from unittest import mock
import pytest
import main
SAMPLE_TIMEZONE_CSV = """Etc/UTC,ZZ,UTC,0,0,0
America/New_York,US,EST,0,-18000,0
Europe/London,GB,BST,0,0,1
"""
SAMPLE_COUNTRY_CSV = """ZZ,Unknown
US,United States
GB,United Kingdom
"""
def test_load_timezones_and_countries(monkeypatch):
tzs = main.load_timezones()
countries = main.load_countries()
assert any(t['zone_name'] == 'America/New_York' for t in tzs)
assert any(c['country_code'] == 'US' for c in countries)
def test_get_tz_and_country_info():
timezones = [{'zone_name': 'A/B', 'country_code': 'US'}]
countries = [{'country_code': 'US', 'country_name': 'United States'}]
assert main.get_tz_info('A/B', timezones)['zone_name'] == 'A/B'
assert main.get_country_info('US', countries)[
'country_name'] == 'United States'
assert main.get_tz_info('X/Y', timezones) is None
assert main.get_country_info('XX', countries) is None
def test_create_embed_all_types(monkeypatch):
# Prevent create_embed from trying to read actual CSV files by patching loaders
monkeypatch.setattr(main, 'load_timezones', lambda: [
{'zone_name': 'Etc/UTC', 'country_code': 'ZZ'}])
monkeypatch.setattr(main, 'load_countries', lambda: [
{'country_code': 'ZZ', 'country_name': 'Nowhere'}])
# reminder
emb = main.create_embed('reminder')
assert emb['title'] == 'Reminder'
assert '5 minute' in emb['description']
assert emb['color'] == 0xe67e22
# reminder_halftime
emb = main.create_embed('reminder_halftime')
assert emb['title'] == 'Reminder halftime'
assert 'Half-time in 5 minutes' in emb['description']
# halftime (should include image)
monkeypatch.setattr(main, 'where_is_it_420', lambda tzs, cs, **kwargs: [])
emb = main.create_embed('halftime')
assert emb['title'] == 'Halftime'
assert emb['image'] is not None
# 420 (should include image and appended tz info string when list empty)
monkeypatch.setattr(main, 'where_is_it_420', lambda tzs, cs, **kwargs: [])
emb = main.create_embed('420')
assert emb['title'] == '420'
assert emb['image'] is not None
# unknown
emb = main.create_embed('nope')
assert emb['description'] == 'Unknown notification type'
def test_where_is_it_420(monkeypatch):
# Limit timezones to a predictable set
monkeypatch.setattr(main.pytz, 'all_timezones', ['Etc/UTC'])
tzs = [{'zone_name': 'Etc/UTC', 'country_code': 'ZZ'}]
countries = [{'country_code': 'ZZ', 'country_name': 'Nowhere'}]
class FakeDatetime:
@staticmethod
def now(tz):
class R:
hour = 4
return R()
monkeypatch.setattr(main, 'datetime', FakeDatetime)
monkeypatch.setattr(main, 'get_tz_info', lambda name,
t: tzs[0] if name == 'Etc/UTC' else None)
monkeypatch.setattr(main, 'get_country_info', lambda code,
c: countries[0] if code == 'ZZ' else None)
res = main.where_is_it_420(tzs, countries)
assert res == ['Nowhere']
def test_main_exits_quickly(monkeypatch):
# Patch send_notification so it doesn't perform network
monkeypatch.setattr(main, 'send_notification', lambda x: None)
# Don't start dashboard during this test
monkeypatch.setattr(main, 'start_dashboard', lambda: None)
# Make schedule.run_pending raise KeyboardInterrupt to exit loop
monkeypatch.setattr(main.schedule, 'run_pending', lambda: (
_ for _ in ()).throw(KeyboardInterrupt()))
# Patch time.sleep to no-op
monkeypatch.setattr(main.time, 'sleep', lambda s: None)
# Ensure WEBHOOK_URL present to avoid early return
monkeypatch.setenv('DISCORD_WEBHOOK_URL', 'http://example.com/webhook')
main.WEBHOOK_URL = 'http://example.com/webhook'
# Should exit quickly due to KeyboardInterrupt from run_pending
main.main()
def test_get_next_scheduled_event():
# 10:14 -> next is 10:15 reminder
now = main.datetime(2025, 1, 1, 10, 14, 30)
nxt = main.get_next_scheduled_event(now)
assert nxt["type"] == "reminder"
assert nxt["at"].hour == 10 and nxt["at"].minute == 15
# 10:50:01 -> next is 11:15 reminder
now = main.datetime(2025, 1, 1, 10, 50, 1)
nxt = main.get_next_scheduled_event(now)
assert nxt["type"] == "reminder"
assert nxt["at"].hour == 11 and nxt["at"].minute == 15
def test_split_tz_name():
assert main.split_tz_name("America/New_York") == ("America", "New_York")
assert main.split_tz_name("America/Argentina/Buenos_Aires") == (
"America",
"Argentina/Buenos_Aires",
)
assert main.split_tz_name("UTC") == ("UTC", "")
tests/test_main_embed.py
+32
View File
@@ -0,0 +1,32 @@
import main
def test_create_embed_all_types(monkeypatch):
monkeypatch.setattr(main, "load_timezones", lambda: [
{"zone_name": "Etc/UTC", "country_code": "ZZ"}
])
monkeypatch.setattr(main, "load_countries", lambda: [
{"country_code": "ZZ", "country_name": "Nowhere"}
])
emb = main.create_embed("reminder")
assert emb["title"] == "Reminder"
assert "5 minute" in emb["description"]
assert emb["color"] == 0xE67E22
emb = main.create_embed("reminder_halftime")
assert emb["title"] == "Reminder halftime"
assert "Half-time in 5 minutes" in emb["description"]
monkeypatch.setattr(main, "where_is_it_420", lambda tzs, cs, **kwargs: [])
emb = main.create_embed("halftime")
assert emb["title"] == "Halftime"
assert emb["image"] is not None
monkeypatch.setattr(main, "where_is_it_420", lambda tzs, cs, **kwargs: [])
emb = main.create_embed("420")
assert emb["title"] == "420"
assert emb["image"] is not None
emb = main.create_embed("nope")
assert emb["description"] == "Unknown notification type"
tests/test_main_scheduler.py
+25
View File
@@ -0,0 +1,25 @@
import main
def test_main_exits_quickly(monkeypatch):
monkeypatch.setattr(main, "send_notification", lambda x: None)
monkeypatch.setattr(main, "start_dashboard", lambda: None)
monkeypatch.setattr(main.schedule, "run_pending", lambda: (
_ for _ in ()).throw(KeyboardInterrupt()))
monkeypatch.setattr(main.time, "sleep", lambda s: None)
monkeypatch.setenv("DISCORD_WEBHOOK_URL", "http://example.com/webhook")
main.WEBHOOK_URL = "http://example.com/webhook"
main.main()
def test_get_next_scheduled_event():
now = main.datetime(2025, 1, 1, 10, 14, 30)
nxt = main.get_next_scheduled_event(now)
assert nxt["type"] == "reminder"
assert nxt["at"].hour == 10 and nxt["at"].minute == 15
now = main.datetime(2025, 1, 1, 10, 50, 1)
nxt = main.get_next_scheduled_event(now)
assert nxt["type"] == "reminder"
assert nxt["at"].hour == 11 and nxt["at"].minute == 15
tests/test_maintenance.py
+88
View File
@@ -0,0 +1,88 @@
from datetime import datetime, timezone
import maintenance
class DummyResponse:
def __init__(self, headers=None, payload=None):
self.headers = headers or {}
self._payload = payload
def json(self):
if self._payload is None:
raise ValueError("no json")
return self._payload
def test_parse_float():
assert maintenance.parse_float("1.5") == 1.5
assert maintenance.parse_float(2) == 2.0
assert maintenance.parse_float(None) is None
assert maintenance.parse_float("nope") is None
def test_parse_message_timestamp_and_build_delete_entry():
msg = {"id": "42", "timestamp": "2026-01-01T10:00:00Z"}
parsed = maintenance.parse_message_timestamp(msg)
assert parsed == datetime(2026, 1, 1, 10, 0, 0)
entry = maintenance.build_delete_entry(msg)
assert entry["id"] == "42"
assert entry["timestamp"] == parsed
def test_should_delete_message():
ts = int(datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc).timestamp())
message = {
"timestamp": "2026-01-01T10:00:00Z",
"webhook_id": "w",
"author": {"id": "a"},
}
assert maintenance.should_delete_message(
message,
webhook_id="w",
author_id="a",
cutoff=ts,
)
assert not maintenance.should_delete_message(
message,
webhook_id="x",
author_id="a",
cutoff=ts,
)
def test_get_rate_limit_retry_after_header_priority():
response = DummyResponse(
headers={
"Retry-After": "2.5",
"X-RateLimit-Reset-After": "10",
},
payload={"retry_after": "20"},
)
assert maintenance.get_rate_limit_retry_after(response) == 2.5
def test_get_rate_limit_retry_after_json_fallback():
response = DummyResponse(payload={"retry_after": "3"})
assert maintenance.get_rate_limit_retry_after(response) == 3.0
def test_get_bucket_exhausted_delay():
response = DummyResponse(
headers={
"X-RateLimit-Remaining": "0",
"X-RateLimit-Reset-After": "1.25",
}
)
assert maintenance.get_bucket_exhausted_delay(response) == 1.25
response_not_exhausted = DummyResponse(
headers={
"X-RateLimit-Remaining": "1",
"X-RateLimit-Reset-After": "1.25",
}
)
assert maintenance.get_bucket_exhausted_delay(
response_not_exhausted) is None
tests/test_thctime.py
+47
View File
@@ -0,0 +1,47 @@
import thctime
def test_load_timezones_and_countries():
tzs = thctime.load_timezones()
countries = thctime.load_countries()
assert any(t["zone_name"] == "America/New_York" for t in tzs)
assert any(c["country_code"] == "US" for c in countries)
def test_get_tz_and_country_info():
timezones = [{"zone_name": "A/B", "country_code": "US"}]
countries = [{"country_code": "US", "country_name": "United States"}]
assert thctime.get_tz_info("A/B", timezones)["zone_name"] == "A/B"
assert thctime.get_country_info("US", countries)[
"country_name"] == "United States"
assert thctime.get_tz_info("X/Y", timezones) is None
assert thctime.get_country_info("XX", countries) is None
def test_where_is_it_420(monkeypatch):
monkeypatch.setattr(thctime.pytz, "all_timezones", ["Etc/UTC"])
tzs = [{"zone_name": "Etc/UTC", "country_code": "ZZ"}]
countries = [{"country_code": "ZZ", "country_name": "Nowhere"}]
class FakeDatetime:
@staticmethod
def now(tz):
class Result:
hour = 4
return Result()
monkeypatch.setattr(thctime, "datetime", FakeDatetime)
res = thctime.where_is_it_420(tzs, countries)
assert res == ["Nowhere"]
def test_split_tz_name():
assert thctime.split_tz_name("America/New_York") == ("America", "New_York")
assert thctime.split_tz_name("America/Argentina/Buenos_Aires") == (
"America",
"Argentina/Buenos_Aires",
)
assert thctime.split_tz_name("UTC") == ("UTC", "")
thctime.py
+175
View File
@@ -0,0 +1,175 @@
from datetime import datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import pytz
TZDB_CACHE: dict | None = None
def get_tzdb_cache() -> dict | None:
return TZDB_CACHE
def init_tzdb_cache() -> dict:
"""Initialize a cached lookup structure for tzdb data."""
global TZDB_CACHE
if TZDB_CACHE is not None:
return TZDB_CACHE
timezones = load_timezones()
countries = load_countries()
tz_to_country_code: dict[str, str] = {}
tz_meta: dict[str, dict] = {}
for tz in timezones:
zone_name = tz.get("zone_name")
country_code = tz.get("country_code")
if not isinstance(zone_name, str) or not zone_name:
continue
if not isinstance(country_code, str) or not country_code:
continue
tz_to_country_code[zone_name] = country_code
region, city = split_tz_name(zone_name)
tz_meta[zone_name] = {
"zone_name": zone_name,
"country_code": country_code,
"region": region,
"city": city,
}
country_code_to_name: dict[str, str] = {}
for c in countries:
code = c.get("country_code")
name = c.get("country_name")
if code and name:
country_code_to_name[code] = str(name).strip().strip('"')
for zone_name, meta in tz_meta.items():
code = meta.get("country_code")
if isinstance(code, str):
meta["country_name"] = country_code_to_name.get(code)
tz_names: list[str] = []
for zone_name in tz_to_country_code.keys():
try:
ZoneInfo(zone_name)
except ZoneInfoNotFoundError:
continue
tz_names.append(zone_name)
TZDB_CACHE = {
"tz_to_country_code": tz_to_country_code,
"country_code_to_name": country_code_to_name,
"tz_names": tz_names,
"tz_meta": tz_meta,
}
return TZDB_CACHE
def split_tz_name(zone_name: str) -> tuple[str, str]:
"""Split an IANA timezone name into (region, city)."""
if "/" not in zone_name:
return zone_name, ""
region, rest = zone_name.split("/", 1)
return region, rest
def load_timezones() -> list[dict]:
"""Load timezones from csv file."""
with open("tzdb/TimeZoneDB.csv/time_zone.csv", "r", encoding="utf-8") as f:
lines = f.readlines()
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."""
with open("tzdb/TimeZoneDB.csv/country.csv", "r", encoding="utf-8") as f:
lines = f.readlines()
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
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],
tz_names: list[str] | None = None,
tz_to_country_code: dict[str, str] | None = None,
country_code_to_name: dict[str, str] | None = None,
) -> list[str]:
"""Get timezones where the current hour is 4 or 16."""
if tz_to_country_code is None:
tz_to_country_code = {}
for tz in timezones:
zone_name = tz.get("zone_name")
country_code = tz.get("country_code")
if isinstance(zone_name, str) and isinstance(country_code, str):
tz_to_country_code[zone_name] = country_code
if country_code_to_name is None:
country_code_to_name = {}
for c in countries:
code = c.get("country_code")
name = c.get("country_name")
if isinstance(code, str) and name is not None:
country_code_to_name[code] = str(name).strip().strip('"')
names_to_check = tz_names if tz_names is not None else pytz.all_timezones
results: list[str] = []
seen: set[str] = set()
for tz_name in names_to_check:
try:
tz_obj = pytz.timezone(tz_name)
except Exception:
continue
now = datetime.now(tz_obj)
if now.hour != 4 and now.hour != 16:
continue
country_code = tz_to_country_code.get(tz_name)
if not country_code:
continue
country_name = country_code_to_name.get(country_code)
if not country_name:
continue
if country_name in seen:
continue
seen.add(country_name)
results.append(country_name)
return results