tux.utils.hot_reload ¶

def _build_extension_map(self) -> None:
    """Build a map of file paths to extension names."""

    for extension in list(self.bot.extensions.keys()):
        if extension == "jishaku":
            continue

        path = path_from_extension(extension)

        if path.exists():
            self.path_to_extension[str(path)] = extension
        else:
            logger.warning(f"Could not find file for extension {extension}, expected at {path}")

def start(self) -> None:
    """Start watching for file changes."""

    self.observer.start()
    logger.info(f"Started watching {self.display_path} for changes (recursive={self.recursive})")

def stop(self) -> None:
    """Stop watching for file changes."""

    self.observer.stop()
    self.observer.join()

    # Cancel any pending tasks
    for task in self.pending_tasks:
        if not task.done():
            task.cancel()
    logger.info("Stopped watching for changes")

def on_modified(self, event: watchdog.events.FileSystemEvent) -> None:
    """
    Handle file modification events.

    Parameters
    ----------
    event : watchdog.events.FileSystemEvent
        The file system event.
    """

    # Skip non-Python files and directories
    if event.is_directory or not str(event.src_path).endswith(".py"):
        return

    file_path = Path(str(event.src_path))

    # Handle special cases first
    if self._handle_special_files(file_path):
        return

    # Handle regular extension files
    self._handle_extension_file(file_path)

def _handle_special_files(self, file_path: Path) -> bool:
    """
    Handle special files like help.py and __init__.py.

    Returns True if handled, False otherwise.
    """

    # Check if it's the help file
    if file_path == self.help_file_path:
        self._reload_help()
        return True

    # Special handling for __init__.py files
    if file_path.name == "__init__.py":
        self._handle_init_file_change(file_path)
        return True

    return False

def _handle_extension_file(self, file_path: Path) -> None:
    """Handle changes to regular extension files."""

    # Check direct mapping first
    if extension := self.path_to_extension.get(str(file_path)):
        self._reload_extension(extension)
        return

    # Try to infer extension name from path
    if possible_extension := get_extension_from_path(file_path, self.base_dir):
        # Try different variations of the extension name
        if self._try_reload_extension_variations(possible_extension, file_path):
            return

    else:
        logger.debug(f"Changed file {file_path} not mapped to any extension")

def _process_extension_reload(self, extension: str, file_path: Path | None = None) -> None:
    """
    Process extension reload with logging and path mapping.

    Parameters
    ----------
    extension : str
        The extension to reload
    file_path : Path, optional
        If provided, updates path mapping after reload
    """

    self._reload_extension(extension)

    if file_path:
        self.path_to_extension[str(file_path)] = extension

def _try_reload_extension_variations(self, extension: str, file_path: Path) -> bool:
    """
    Try to reload an extension with different name variations.

    Returns True if successful, False otherwise.
    """

    # Check exact match
    if extension in self.bot.extensions:
        self._process_extension_reload(extension, file_path)
        return True

    # Check if a shorter version of this extension is already loaded
    # This prevents duplicate loading of the same module
    parts = extension.split(".")
    for i in range(len(parts) - 1, 0, -1):
        shorter_ext = ".".join(parts[:i])
        if shorter_ext in self.bot.extensions:
            logger.warning(f"Skipping reload of {extension} as parent module {shorter_ext} already loaded")
            # Still update our path mapping
            self.path_to_extension[str(file_path)] = shorter_ext
            return True

    # Check parent modules
    parent_ext = extension
    while "." in parent_ext:
        parent_ext = parent_ext.rsplit(".", 1)[0]
        if parent_ext in self.bot.extensions:
            self._process_extension_reload(parent_ext, file_path)
            return True

    # Try without tux prefix
    if extension.startswith("tux."):
        no_prefix = extension[4:]  # Remove "tux."
        if no_prefix in self.bot.extensions:
            self._process_extension_reload(no_prefix, file_path)
            return True

    return False

def _handle_init_file_change(self, init_file_path: Path) -> None:
    """
    Handle changes to __init__.py files that may be used by multiple cogs.

    Parameters
    ----------
    init_file_path : Path
        Path to the __init__.py file that changed
    """

    # Get the directory containing this __init__.py file
    directory = init_file_path.parent
    package_path = directory.relative_to(self.base_dir)

    # Convert path to potential extension prefix
    package_name = str(package_path).replace(os.sep, ".")
    if not package_name.startswith("cogs."):
        return

    # Find all extensions that start with this package name
    full_package = f"tux.{package_name}"

    # Reload the modules themselves first
    reload_module_by_name(full_package)
    reload_module_by_name(package_name)  # Try the short version too

    if extensions_to_reload := self._collect_extensions_to_reload(full_package, package_name):
        logger.info(f"Reloading {len(extensions_to_reload)} extensions after __init__.py change")
        for ext in extensions_to_reload:
            self._process_extension_reload(ext)

def _collect_extensions_to_reload(self, full_package: str, short_package: str) -> list[str]:
    """Collect extensions that need to be reloaded based on package names."""

    extensions_to_reload: list[str] = [
        ext for ext in self.bot.extensions if ext.startswith(f"{full_package}.") or ext == full_package
    ]

    # Check for extensions with short package prefix (cogs.moderation)
    extensions_to_reload.extend(
        [ext for ext in self.bot.extensions if ext.startswith(f"{short_package}.") or ext == short_package],
    )

    return extensions_to_reload

def _reload_extension(self, extension: str) -> None:
    """
    Reload an extension.

    Parameters
    ----------
    extension : str
        The extension to reload.
    """

    try:
        # Add a small delay to ensure file write is complete
        time.sleep(0.1)

        # Use the stored main event loop instead of trying to get the current one
        # which doesn't exist in the watchdog thread
        asyncio.run_coroutine_threadsafe(self._async_reload_extension(extension), self.loop)

    except Exception as e:
        logger.error(f"Failed to schedule reload of extension {extension}: {e}")

def _reload_help(self) -> None:
    """Reload the help command."""

    try:
        # Add a small delay to ensure file write is complete
        time.sleep(0.1)

        # Use the stored main event loop
        asyncio.run_coroutine_threadsafe(self._async_reload_help(), self.loop)

    except Exception as e:
        logger.error(f"Failed to schedule reload of help command: {e}")
        import traceback

        logger.error(traceback.format_exc())

async def _async_reload_extension(self, extension: str) -> None:
    """
    Asynchronously reload an extension.

    Parameters
    ----------
    extension : str
        The extension to reload.
    """

    try:
        await self.bot.reload_extension(extension)
        logger.info(f"Reloaded extension {extension}")

    except commands.ExtensionNotLoaded:
        try:
            # Try to load it if it wasn't loaded before
            await self.bot.load_extension(extension)
            logger.info(f"Loaded new extension {extension}")

            # Update our map
            path = path_from_extension(extension)
            self.path_to_extension[str(path)] = extension

        except commands.ExtensionError as e:
            logger.error(f"Failed to load new extension {extension}: {e}")

    except commands.ExtensionError as e:
        logger.error(f"Failed to reload extension {extension}: {e}")

async def _async_reload_help(self) -> None:
    """Asynchronously reload the help command."""

    try:
        # Force reload of the help module
        if "tux.help" in sys.modules:
            importlib.reload(sys.modules["tux.help"])
        else:
            importlib.import_module("tux.help")

        try:
            # Create a dynamic import that doesn't happen at module load time
            # This breaks the circular import chain
            help_module = importlib.import_module("tux.help")
            TuxHelp = help_module.TuxHelp  # noqa: N806

            # Reset the help command with a new instance from the reloaded module
            self.bot.help_command = TuxHelp()
            logger.info("Reloaded help command")

        except (AttributeError, ImportError) as e:
            logger.error(f"Error accessing TuxHelp class: {e}")

    except Exception as e:
        logger.error(f"Failed to reload help command: {e}")
        import traceback

        logger.error(traceback.format_exc())