From 9cc3b1ecc1d92adb69de00f8eed9c67627268372 Mon Sep 17 00:00:00 2001 From: Kingston-SCYD Date: Mon, 7 Sep 2026 15:45:03 -0500 Subject: [PATCH] Add deadman switch cog --- README.md | 32 ++++++ deadman/__init__.py | 16 +++ deadman/deadman.py | 238 ++++++++++++++++++++++++++++++++++++++++++++ deadman/info.json | 24 +++++ 4 files changed, 310 insertions(+) create mode 100644 deadman/__init__.py create mode 100644 deadman/deadman.py create mode 100644 deadman/info.json diff --git a/README.md b/README.md index da7758c..fcadf67 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,9 @@ Homepage: https://homepage.kingstons-scrapyard.net/ ## Table of Contents +**Deadman** +- [Deadman](#deadman) + **AutoRoom** - [Installation](#installation) - [AutoRoom](#autoroom) @@ -49,6 +52,35 @@ Homepage: https://homepage.kingstons-scrapyard.net/ [p]load autoroom ``` +Deadman (bot owner only): + +``` +[p]cog install scrapyard deadman +[p]load deadman +``` + +--- + +## Deadman + +A member of the Scrapyard Cogworks package. A deadman switch that alerts listed users if not disabled before it expires. + +- **Owner only**: Only the bot owner can arm, disable, or check status. +- **Arm**: `[p]deadman arm @user1 @user2 [channel]` +- **Disable**: `[p]deadman disable` +- **Status**: `[p]deadman status` + +**Duration format** (flexible string): +- `30s` = 30 seconds +- `6h` = 6 hours +- `2h30m` = 2 hours 30 minutes +- `1d` = 1 day +- `1d12h` = 1 day 12 hours + +Supported units: `s`, `m`, `h`, `d`, `w` (and their full names). + +When the switch is armed, the bot stores the expiry time and alerts users by DM. If an optional channel is provided, it also posts the alert there. If `[p]deadman disable` is not run before expiry, the listed users are DM'd and pinged that "the deadman switch has been TRIGGERED". + --- ## AutoRoom diff --git a/deadman/__init__.py b/deadman/__init__.py new file mode 100644 index 0000000..ac2dd25 --- /dev/null +++ b/deadman/__init__.py @@ -0,0 +1,16 @@ +"""Package for Deadman cog.""" + +import json +from pathlib import Path + +from redbot.core.bot import Red + +from .deadman import Deadman + +with Path(__file__).parent.joinpath("info.json").open() as fp: + __red_end_user_data_statement__ = json.load(fp)["end_user_data_statement"] + + +async def setup(bot: Red) -> None: + """Load Deadman cog.""" + await bot.add_cog(Deadman(bot)) diff --git a/deadman/deadman.py b/deadman/deadman.py new file mode 100644 index 0000000..b1cd418 --- /dev/null +++ b/deadman/deadman.py @@ -0,0 +1,238 @@ +import asyncio +from datetime import datetime, timedelta +from typing import Optional + +import discord +from redbot.core import Config, commands + + +def parse_duration(text: str) -> Optional[timedelta]: + """Parse a flexible duration string like ``6h``, ``2h30m``, ``90s`` or ``1d``.""" + text = text.strip().lower() + if not text: + return None + + total = timedelta() + num = "" + unit = "" + seen_unit = False + + for char in text: + if char.isdigit(): + if seen_unit: + # A digit after a unit means a new component, e.g. "1d12h" + delta = _unit_to_timedelta(int(num), unit) + if delta is None: + return None + total += delta + num = char + unit = "" + seen_unit = False + else: + num += char + else: + if not num: + return None + unit += char + seen_unit = True + + if unit: + delta = _unit_to_timedelta(int(num), unit) + if delta is None: + return None + total += delta + elif num: + # Trailing digits with no unit, e.g. "5m30" - invalid. + return None + + if total <= timedelta(0): + return None + + return total + + +def _unit_to_timedelta(amount: int, unit: str) -> Optional[timedelta]: + if not amount or not isinstance(amount, int): + return None + if unit and unit.isdigit(): + return None + if unit in ("s", "sec", "secs", "second", "seconds"): + return timedelta(seconds=amount) + if unit in ("m", "min", "mins", "minute", "minutes"): + return timedelta(minutes=amount) + if unit in ("h", "hr", "hrs", "hour", "hours"): + return timedelta(hours=amount) + if unit in ("d", "day", "days"): + return timedelta(days=amount) + if unit in ("w", "week", "weeks"): + return timedelta(weeks=amount) + return None + + +class Deadman(commands.Cog): + """Deadman switch - pings listed users if [p]deadman disable isn't run before expiration.""" + + def __init__(self, bot): + self.bot = bot + self.config = Config.get_conf(self, identifier=918273645) + self.config.register_global( + armed=False, + expires_at=None, + users=[], + trigger_channel=None, + ) + self._task = None + self._triggered = False + + async def red_delete_data_for_user(self, **kwargs): + return + + def cog_unload(self): + if self._task: + self._task.cancel() + + async def cog_load(self): + self._task = asyncio.create_task(self._monitor()) + + async def _monitor(self): + """Background task that checks every 30 seconds whether the switch has expired.""" + while True: + try: + await asyncio.sleep(30) + await self._check_expiry() + except asyncio.CancelledError: + break + except Exception: + pass + + async def _check_expiry(self): + if not await self.config.armed(): + self._triggered = False + return + + expires_at = await self.config.expires_at() + if expires_at is None: + return + + if datetime.utcnow() >= datetime.fromisoformat(expires_at): + if not self._triggered: + await self._trigger() + # Keep armed/stale so repeated pings don't spam; disable clears state. + + async def _trigger(self): + self._triggered = True + users = await self.config.users() + channel_id = await self.config.trigger_channel() + + victims: list[discord.User] = [] + for uid in users: + user = self.bot.get_user(uid) + if user is None: + try: + user = await self.bot.fetch_user(uid) + except Exception: + continue + if user is not None: + victims.append(user) + + for user in victims: + try: + await user.send( + "**GENERAL ALERT**\nThe deadman switch has been TRIGGERED. " + "Please contact the operator immediately." + ) + except discord.Forbidden: + pass + except Exception: + pass + + if channel_id: + channel = self.bot.get_channel(channel_id) + if channel: + mentions = " ".join( + f"<@{uid}>" for uid in users if self.bot.get_user(uid) + ) + try: + await channel.send( + f"**GENERAL ALERT**\nThe deadman switch has been TRIGGERED. " + f"{mentions} Please contact the operator immediately." + ) + except discord.Forbidden: + pass + except Exception: + pass + + @commands.group() + @commands.is_owner() + async def deadman(self, ctx): + """Deadman switch control.""" + + @deadman.command(name="arm") + async def deadman_arm(self, ctx, duration: str, users: commands.Greedy[discord.User], trigger_channel: Optional[discord.TextChannel] = None): + """Arm the deadman switch for a duration, pinging listed users if not disabled in time. + + Duration format: a flexible string like ``6h``, ``2h30m``, ``90s``, ``1d``, or a combination. + Users: mention the users to alert (e.g. ``@user1 @user2``). + Optional: a channel where the alert should also be posted. + """ + if not users: + await ctx.send("You must provide at least one user to alert.") + return + + duration_td = parse_duration(duration) + if duration_td is None: + await ctx.send( + "Invalid duration. Examples: `6h`, `2h30m`, `90s`, `1d`, `1d12h`." + ) + return + + expires_at = datetime.utcnow() + duration_td + await self.config.armed.set(True) + await self.config.expires_at.set(expires_at.isoformat()) + await self.config.users.set([u.id for u in users]) + await self.config.trigger_channel.set(trigger_channel.id if trigger_channel else None) + self._triggered = False + + expires_str = expires_at.strftime("%Y-%m-%d %H:%M UTC") + await ctx.send( + f"Deadman switch ARMED.\n" + f"Expires: **{expires_str}** (in {duration_td})\n" + f"Users notified: {', '.join(u.mention for u in users)}\n" + + (f"Alert channel: {trigger_channel.mention}\n" if trigger_channel else "\n") + + "Run `[p]deadman disable` before that time to cancel." + ) + + @deadman.command(name="disable") + async def deadman_disable(self, ctx): + """Disable the currently armed deadman switch.""" + armed = await self.config.armed() + if not armed: + await ctx.send("The deadman switch is not currently armed.") + return + + await self.config.armed.set(False) + await self.config.expires_at.set(None) + await self.config.users.set([]) + await self.config.trigger_channel.set(None) + self._triggered = False + await ctx.send("Deadman switch disabled.") + + @deadman.command(name="status") + async def deadman_status(self, ctx): + """Show the current deadman switch status.""" + armed = await self.config.armed() + if not armed: + await ctx.send("The deadman switch is currently **disarmed**.") + return + + expires_at = await self.config.expires_at() + users = await self.config.users() + expires_dt = datetime.fromisoformat(expires_at) + remaining = expires_dt - datetime.utcnow() + mentions = ", ".join(f"<@{uid}>" for uid in users) if users else "None" + await ctx.send( + f"Deadman switch is **ARMED**.\n" + f"Expires at: {expires_dt.strftime('%Y-%m-%d %H:%M UTC')}\n" + f"Time remaining: {remaining}\n" + f"Users to alert: {mentions}" + ) \ No newline at end of file diff --git a/deadman/info.json b/deadman/info.json new file mode 100644 index 0000000..d08132e --- /dev/null +++ b/deadman/info.json @@ -0,0 +1,24 @@ +{ + "name": "Deadman", + "author": [ + "Scrapyard Cogworks" + ], + "short": "Deadman switch that pings listed users if not disabled in time.", + "description": "Arm a deadman switch with a duration. If [p]deadman disable is not run before it expires, the bot DMs and pings all listed users that the switch has been triggered. Bot owner only.", + "install_msg": "Thanks for installing Deadman from Scrapyard Cogworks! Visit us at https://homepage.kingstons-scrapyard.net/", + "requirements": [], + "tags": [ + "deadman", + "switch", + "alert", + "safety", + "ping" + ], + "min_bot_version": "3.5.0", + "min_python_version": [ + 3, + 11, + 0 + ], + "end_user_data_statement": "This cog stores Discord user and channel IDs of configured participants. It does not store message content or profane data." +} \ No newline at end of file