feat(hotreload): Channel Notifications (#51)
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 41s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 47s

# Channel Notifications

This PR adds the ability for HotReload to send messages to a configurable discord channel when reloading a cog. Messages are only sent after the cog is reloaded to prevent slowdowns.
<!-- Create a new issue, if it doesn't exist yet -->

- [x] By submitting this pull request, I permit [cswimr](https://www.coastalcommits.com/cswimr) to license my work under
  the [Mozilla Public License Version 2.0](https://www.coastalcommits.com/cswimr/SeaCogs/src/branch/main/LICENSE).

Reviewed-on: https://www.coastalcommits.com/cswimr/SeaCogs/pulls/51
This commit is contained in:
cswimr 2025-01-26 10:06:14 -05:00
parent ff9e20be91
commit ccf9389e13
Signed by: CoastalCommits
GPG key ID: 7E73189F651A553F

View file

@ -3,7 +3,7 @@ from pathlib import Path
from typing import Sequence
from red_commons.logging import RedTraceLogger, getLogger
from redbot.core import commands
from redbot.core import Config, checks, commands
from redbot.core.bot import Red
from redbot.core.core_commands import CoreLogic
from redbot.core.utils.chat_formatting import bold, humanize_list
@ -16,14 +16,16 @@ class HotReload(commands.Cog):
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.1.2"
__version__ = "1.2.0"
__documentation__ = "https://seacogs.coastalcommits.com/hotreload/"
def __init__(self, bot: Red) -> None:
super().__init__()
self.bot: Red = bot
self.config = Config.get_conf(self, identifier=294518358420750336, force_registration=True)
self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload")
self.observer = None
self.config.register_global(notify_channel=None)
watchdog_loggers = [getLogger(name="watchdog.observers.inotify_buffer")]
for watchdog_logger in watchdog_loggers:
watchdog_logger.setLevel("INFO") # SHUT UP!!!!
@ -61,17 +63,29 @@ class HotReload(commands.Cog):
self.observer = Observer()
paths = await self.get_paths()
for path in paths:
self.observer.schedule(event_handler=HotReloadHandler(bot=self.bot, path=path), path=path, recursive=True)
self.observer.schedule(event_handler=HotReloadHandler(cog=self, path=path), path=path, recursive=True)
self.observer.start()
self.logger.info("Started observer. Watching for file changes.")
@checks.is_owner()
@commands.group(name="hotreload")
async def hotreload_group(self, ctx: commands.Context) -> None:
"""HotReload configuration commands."""
pass
@hotreload_group.command(name="notifychannel")
async def hotreload_notifychannel(self, ctx: commands.Context, channel: commands.TextChannelConverter) -> None:
"""Set the channel to send notifications to."""
await self.config.notify_channel.set(channel.id)
await ctx.send(f"Notifications will be sent to {channel.mention}.")
class HotReloadHandler(RegexMatchingEventHandler):
"""Handler for file changes."""
def __init__(self, bot: Red, path: Path) -> None:
def __init__(self, cog: HotReload, path: Path) -> None:
super().__init__(regexes=[r".*\.py$"])
self.bot: Red = bot
self.cog: HotReload = cog
self.path: Path = path
self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload.Observer")
@ -99,11 +113,15 @@ class HotReloadHandler(RegexMatchingEventHandler):
self.logger.info("File %s has been %s%s.", event.src_path, event.event_type, dest)
run_coroutine_threadsafe(self.reload_cogs(cogs_to_reload), loop=self.bot.loop)
run_coroutine_threadsafe(self.reload_cogs(cogs_to_reload), loop=self.cog.bot.loop)
async def reload_cogs(self, cog_names: Sequence[str]) -> None:
"""Reload modified cog."""
core_logic = CoreLogic(bot=self.bot)
core_logic = CoreLogic(bot=self.cog.bot)
self.logger.info("Reloading cogs: %s", humanize_list(cog_names, style="unit"))
await core_logic._reload(pkg_names=cog_names) # noqa: SLF001 # We have to use this private method because there is no public API to reload other cogs
self.logger.info("Reloaded cogs: %s", humanize_list(cog_names, style="unit"))
channel = self.cog.bot.get_channel(await self.cog.config.notify_channel())
if channel:
await channel.send(f"Reloaded cogs: {humanize_list(cog_names, style='unit')}")