diff --git a/changelog.d/3240.removal.rst b/changelog.d/3240.removal.rst new file mode 100644 index 000000000..75d57841b --- /dev/null +++ b/changelog.d/3240.removal.rst @@ -0,0 +1 @@ +Removed ``safe_delete``, ``fuzzy_command_search``, ``format_fuzzy_results`` and ``create_backup`` functions from ``redbot.core.utils``. \ No newline at end of file diff --git a/redbot/cogs/downloader/repo_manager.py b/redbot/cogs/downloader/repo_manager.py index 0edf29fa6..9c28a7a76 100644 --- a/redbot/cogs/downloader/repo_manager.py +++ b/redbot/cogs/downloader/repo_manager.py @@ -28,7 +28,7 @@ from typing import ( import discord from redbot.core import data_manager, commands, Config -from redbot.core.utils import safe_delete +from redbot.core.utils._internal_utils import safe_delete from redbot.core.i18n import Translator from . import errors diff --git a/redbot/core/commands/help.py b/redbot/core/commands/help.py index 2f2d0b359..1de10cfb6 100644 --- a/redbot/core/commands/help.py +++ b/redbot/core/commands/help.py @@ -43,7 +43,8 @@ from discord.ext import commands as dpy_commands from . import commands from .context import Context from ..i18n import Translator -from ..utils import menus, fuzzy_command_search, format_fuzzy_results +from ..utils import menus +from ..utils._internal_utils import fuzzy_command_search, format_fuzzy_results from ..utils.chat_formatting import box, pagify __all__ = ["red_help", "RedHelpFormatter", "HelpSettings"] diff --git a/redbot/core/core_commands.py b/redbot/core/core_commands.py index c634db6ec..fef089b07 100644 --- a/redbot/core/core_commands.py +++ b/redbot/core/core_commands.py @@ -31,7 +31,7 @@ from . import ( i18n, config, ) -from .utils import create_backup +from .utils._internal_utils import create_backup from .utils.predicates import MessagePredicate from .utils.chat_formatting import ( box, diff --git a/redbot/core/events.py b/redbot/core/events.py index ec48f26c0..44e5a9e11 100644 --- a/redbot/core/events.py +++ b/redbot/core/events.py @@ -16,8 +16,8 @@ from redbot.core.commands import RedHelpFormatter from .. import __version__ as red_version, version_info as red_version_info, VersionInfo from . import commands from .config import get_latest_confs +from .utils._internal_utils import fuzzy_command_search, format_fuzzy_results from .utils.chat_formatting import inline, bordered, format_perms_list, humanize_timedelta -from .utils import fuzzy_command_search, format_fuzzy_results log = logging.getLogger("red") init() diff --git a/redbot/core/utils/__init__.py b/redbot/core/utils/__init__.py index cb21c4355..dfdab23b0 100644 --- a/redbot/core/utils/__init__.py +++ b/redbot/core/utils/__init__.py @@ -1,13 +1,7 @@ import asyncio -import json -import logging -import os -import shutil -import tarfile from asyncio import AbstractEventLoop, as_completed, Semaphore from asyncio.futures import isfuture from itertools import chain -from pathlib import Path from typing import ( Any, AsyncIterator, @@ -25,24 +19,7 @@ from typing import ( TYPE_CHECKING, ) -import discord -from datetime import datetime -from fuzzywuzzy import fuzz, process - -from .. import commands, data_manager -from .chat_formatting import box - -if TYPE_CHECKING: - from ..commands import Command, Context - -__all__ = [ - "bounded_gather", - "safe_delete", - "fuzzy_command_search", - "format_fuzzy_results", - "deduplicate_iterables", - "create_backup", -] +__all__ = ("bounded_gather", "deduplicate_iterables") _T = TypeVar("_T") @@ -57,27 +34,6 @@ def deduplicate_iterables(*iterables): return list(dict.fromkeys(chain.from_iterable(iterables))) -def _fuzzy_log_filter(record): - return record.funcName != "extractWithoutOrder" - - -logging.getLogger().addFilter(_fuzzy_log_filter) - - -def safe_delete(pth: Path): - if pth.exists(): - for root, dirs, files in os.walk(str(pth)): - os.chmod(root, 0o700) - - for d in dirs: - os.chmod(os.path.join(root, d), 0o700) - - for f in files: - os.chmod(os.path.join(root, f), 0o700) - - shutil.rmtree(str(pth), ignore_errors=True) - - # https://github.com/PyCQA/pylint/issues/2717 class AsyncFilter(AsyncIterator[_T], Awaitable[List[_T]]): # pylint: disable=duplicate-bases """Class returned by `async_filter`. See that function for details. @@ -188,124 +144,6 @@ async def async_enumerate( start += 1 -async def fuzzy_command_search( - ctx: "Context", - term: Optional[str] = None, - *, - commands: Optional[Set["Command"]] = None, - min_score: int = 80, -) -> Optional[List["Command"]]: - """Search for commands which are similar in name to the one invoked. - - Returns a maximum of 5 commands which must all be at least matched - greater than ``min_score``. - - Parameters - ---------- - ctx : `commands.Context ` - The command invocation context. - term : Optional[str] - The name of the invoked command. If ``None``, - `Context.invoked_with` will be used instead. - commands : Optional[Set[commands.Command]] - The commands available to choose from when doing a fuzzy match. - When omitted, `Bot.walk_commands` will be used instead. - min_score : int - The minimum score for matched commands to reach. Defaults to 80. - - Returns - ------- - Optional[List[`commands.Command `]] - A list of commands which were fuzzily matched with the invoked - command. - - """ - if ctx.guild is not None: - enabled = await ctx.bot._config.guild(ctx.guild).fuzzy() - else: - enabled = await ctx.bot._config.fuzzy() - - if not enabled: - return - - if term is None: - term = ctx.invoked_with - - # If the term is an alias or CC, we don't want to send a supplementary fuzzy search. - alias_cog = ctx.bot.get_cog("Alias") - if alias_cog is not None: - is_alias, alias = await alias_cog.is_alias(ctx.guild, term) - - if is_alias: - return - customcom_cog = ctx.bot.get_cog("CustomCommands") - if customcom_cog is not None: - cmd_obj = customcom_cog.commandobj - - try: - await cmd_obj.get(ctx.message, term) - except: - pass - else: - return - - # Do the scoring. `extracted` is a list of tuples in the form `(command, score)` - extracted = process.extract( - term, (commands or set(ctx.bot.walk_commands())), limit=5, scorer=fuzz.QRatio - ) - if not extracted: - return - - # Filter through the fuzzy-matched commands. - matched_commands = [] - for command, score in extracted: - if score < min_score: - # Since the list is in decreasing order of score, we can exit early. - break - if await command.can_see(ctx): - matched_commands.append(command) - - return matched_commands - - -async def format_fuzzy_results( - ctx: "Context", matched_commands: List["Command"], *, embed: Optional[bool] = None -) -> Union[str, discord.Embed]: - """Format the result of a fuzzy command search. - - Parameters - ---------- - ctx : `commands.Context ` - The context in which this result is being displayed. - matched_commands : List[`commands.Command `] - A list of commands which have been matched by the fuzzy search, sorted - in order of decreasing similarity. - embed : bool - Whether or not the result should be an embed. If set to ``None``, this - will default to the result of `ctx.embed_requested`. - - Returns - ------- - Union[str, discord.Embed] - The formatted results. - - """ - if embed is not False and (embed is True or await ctx.embed_requested()): - lines = [] - for cmd in matched_commands: - lines.append(f"**{ctx.clean_prefix}{cmd.qualified_name}** {cmd.short_doc}") - return discord.Embed( - title="Perhaps you wanted one of these?", - colour=await ctx.embed_colour(), - description="\n".join(lines), - ) - else: - lines = [] - for cmd in matched_commands: - lines.append(f"{ctx.clean_prefix}{cmd.qualified_name} -- {cmd.short_doc}") - return "Perhaps you wanted one of these? " + box("\n".join(lines), lang="vhdl") - - async def _sem_wrapper(sem, task): async with sem: return await task @@ -402,45 +240,3 @@ def bounded_gather( tasks = (_sem_wrapper(semaphore, task) for task in coros_or_futures) return asyncio.gather(*tasks, loop=loop, return_exceptions=return_exceptions) - - -async def create_backup(dest: Path = Path.home()) -> Optional[Path]: - data_path = Path(data_manager.core_data_path().parent) - if not data_path.exists(): - return - - dest.mkdir(parents=True, exist_ok=True) - timestr = datetime.utcnow().strftime("%Y-%m-%dT%H-%M-%S") - backup_fpath = dest / f"redv3_{data_manager.instance_name}_{timestr}.tar.gz" - - to_backup = [] - exclusions = [ - "__pycache__", - "Lavalink.jar", - os.path.join("Downloader", "lib"), - os.path.join("CogManager", "cogs"), - os.path.join("RepoManager", "repos"), - ] - - # Avoiding circular imports - from ...cogs.downloader.repo_manager import RepoManager - - repo_mgr = RepoManager() - await repo_mgr.initialize() - repo_output = [] - for repo in repo_mgr.repos: - repo_output.append({"url": repo.url, "name": repo.name, "branch": repo.branch}) - repos_file = data_path / "cogs" / "RepoManager" / "repos.json" - with repos_file.open("w") as fs: - json.dump(repo_output, fs, indent=4) - instance_file = data_path / "instance.json" - with instance_file.open("w") as fs: - json.dump({data_manager.instance_name: data_manager.basic_config}, fs, indent=4) - for f in data_path.glob("**/*"): - if not any(ex in str(f) for ex in exclusions) and f.is_file(): - to_backup.append(f) - - with tarfile.open(str(backup_fpath), "w:gz") as tar: - for f in to_backup: - tar.add(str(f), arcname=f.relative_to(data_path), recursive=False) - return backup_fpath diff --git a/redbot/core/utils/_internal_utils.py b/redbot/core/utils/_internal_utils.py new file mode 100644 index 000000000..63fd23d33 --- /dev/null +++ b/redbot/core/utils/_internal_utils.py @@ -0,0 +1,202 @@ +from __future__ import annotations + +import json +import logging +import os +import shutil +import tarfile +from datetime import datetime +from pathlib import Path +from typing import List, Optional, Set, Union, TYPE_CHECKING + +import discord +from fuzzywuzzy import fuzz, process + +from redbot.core import data_manager +from redbot.core.utils.chat_formatting import box + +if TYPE_CHECKING: + from redbot.core.commands import Command, Context + +__all__ = ("safe_delete", "fuzzy_command_search", "format_fuzzy_results", "create_backup") + + +def safe_delete(pth: Path): + if pth.exists(): + for root, dirs, files in os.walk(str(pth)): + os.chmod(root, 0o700) + + for d in dirs: + os.chmod(os.path.join(root, d), 0o700) + + for f in files: + os.chmod(os.path.join(root, f), 0o700) + + shutil.rmtree(str(pth), ignore_errors=True) + + +def _fuzzy_log_filter(record): + return record.funcName != "extractWithoutOrder" + + +logging.getLogger().addFilter(_fuzzy_log_filter) + + +async def fuzzy_command_search( + ctx: Context, + term: Optional[str] = None, + *, + commands: Optional[Set[Command]] = None, + min_score: int = 80, +) -> Optional[List[Command]]: + """Search for commands which are similar in name to the one invoked. + + Returns a maximum of 5 commands which must all be at least matched + greater than ``min_score``. + + Parameters + ---------- + ctx : `commands.Context ` + The command invocation context. + term : Optional[str] + The name of the invoked command. If ``None``, + `Context.invoked_with` will be used instead. + commands : Optional[Set[commands.Command]] + The commands available to choose from when doing a fuzzy match. + When omitted, `Bot.walk_commands` will be used instead. + min_score : int + The minimum score for matched commands to reach. Defaults to 80. + + Returns + ------- + Optional[List[`commands.Command `]] + A list of commands which were fuzzily matched with the invoked + command. + + """ + if ctx.guild is not None: + enabled = await ctx.bot._config.guild(ctx.guild).fuzzy() + else: + enabled = await ctx.bot._config.fuzzy() + + if not enabled: + return None + + if term is None: + term = ctx.invoked_with + + # If the term is an alias or CC, we don't want to send a supplementary fuzzy search. + alias_cog = ctx.bot.get_cog("Alias") + if alias_cog is not None: + is_alias, alias = await alias_cog.is_alias(ctx.guild, term) + + if is_alias: + return None + customcom_cog = ctx.bot.get_cog("CustomCommands") + if customcom_cog is not None: + cmd_obj = customcom_cog.commandobj + + try: + await cmd_obj.get(ctx.message, term) + except: + pass + else: + return None + + # Do the scoring. `extracted` is a list of tuples in the form `(command, score)` + extracted = process.extract( + term, (commands or set(ctx.bot.walk_commands())), limit=5, scorer=fuzz.QRatio + ) + if not extracted: + return None + + # Filter through the fuzzy-matched commands. + matched_commands = [] + for command, score in extracted: + if score < min_score: + # Since the list is in decreasing order of score, we can exit early. + break + if await command.can_see(ctx): + matched_commands.append(command) + + return matched_commands + + +async def format_fuzzy_results( + ctx: Context, matched_commands: List[Command], *, embed: Optional[bool] = None +) -> Union[str, discord.Embed]: + """Format the result of a fuzzy command search. + + Parameters + ---------- + ctx : `commands.Context ` + The context in which this result is being displayed. + matched_commands : List[`commands.Command `] + A list of commands which have been matched by the fuzzy search, sorted + in order of decreasing similarity. + embed : bool + Whether or not the result should be an embed. If set to ``None``, this + will default to the result of `ctx.embed_requested`. + + Returns + ------- + Union[str, discord.Embed] + The formatted results. + + """ + if embed is not False and (embed is True or await ctx.embed_requested()): + lines = [] + for cmd in matched_commands: + lines.append(f"**{ctx.clean_prefix}{cmd.qualified_name}** {cmd.short_doc}") + return discord.Embed( + title="Perhaps you wanted one of these?", + colour=await ctx.embed_colour(), + description="\n".join(lines), + ) + else: + lines = [] + for cmd in matched_commands: + lines.append(f"{ctx.clean_prefix}{cmd.qualified_name} -- {cmd.short_doc}") + return "Perhaps you wanted one of these? " + box("\n".join(lines), lang="vhdl") + + +async def create_backup(dest: Path = Path.home()) -> Optional[Path]: + data_path = Path(data_manager.core_data_path().parent) + if not data_path.exists(): + return None + + dest.mkdir(parents=True, exist_ok=True) + timestr = datetime.utcnow().strftime("%Y-%m-%dT%H-%M-%S") + backup_fpath = dest / f"redv3_{data_manager.instance_name}_{timestr}.tar.gz" + + to_backup = [] + exclusions = [ + "__pycache__", + "Lavalink.jar", + os.path.join("Downloader", "lib"), + os.path.join("CogManager", "cogs"), + os.path.join("RepoManager", "repos"), + ] + + # Avoiding circular imports + from ...cogs.downloader.repo_manager import RepoManager + + repo_mgr = RepoManager() + await repo_mgr.initialize() + repo_output = [] + for repo in repo_mgr.repos: + repo_output.append({"url": repo.url, "name": repo.name, "branch": repo.branch}) + repos_file = data_path / "cogs" / "RepoManager" / "repos.json" + with repos_file.open("w") as fs: + json.dump(repo_output, fs, indent=4) + instance_file = data_path / "instance.json" + with instance_file.open("w") as fs: + json.dump({data_manager.instance_name: data_manager.basic_config}, fs, indent=4) + for f in data_path.glob("**/*"): + if not any(ex in str(f) for ex in exclusions) and f.is_file(): + to_backup.append(f) + + with tarfile.open(str(backup_fpath), "w:gz") as tar: + for f in to_backup: + tar.add(str(f), arcname=str(f.relative_to(data_path)), recursive=False) + return backup_fpath diff --git a/redbot/setup.py b/redbot/setup.py index 9259340a9..1ab706bff 100644 --- a/redbot/setup.py +++ b/redbot/setup.py @@ -12,7 +12,7 @@ import appdirs import click import redbot.logging -from redbot.core.utils import safe_delete, create_backup as red_create_backup +from redbot.core.utils._internal_utils import safe_delete, create_backup as red_create_backup from redbot.core import config, data_manager, drivers from redbot.core.drivers import BackendType, IdentifierData