We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Stop backing up lavalink logs
Lavalink logs are host specific, stop backing them up.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `redbot/core/utils/_internal_utils.py`
Content:
```
1 from __future__ import annotations
2 
3 import asyncio
4 import collections.abc
5 import json
6 import logging
7 import os
8 import re
9 import shutil
10 import tarfile
11 from datetime import datetime
12 from pathlib import Path
13 from typing import (
14     AsyncIterator,
15     Awaitable,
16     Callable,
17     Iterator,
18     List,
19     Optional,
20     Union,
21     TYPE_CHECKING,
22     Tuple,
23 )
24 
25 import aiohttp
26 import discord
27 import pkg_resources
28 from fuzzywuzzy import fuzz, process
29 from redbot import VersionInfo
30 
31 from redbot.core import data_manager
32 from redbot.core.utils.chat_formatting import box
33 
34 if TYPE_CHECKING:
35     from redbot.core.bot import Red
36     from redbot.core.commands import Command, Context
37 
38 main_log = logging.getLogger("red")
39 
40 __all__ = (
41     "safe_delete",
42     "fuzzy_command_search",
43     "format_fuzzy_results",
44     "create_backup",
45     "send_to_owners_with_preprocessor",
46     "send_to_owners_with_prefix_replaced",
47     "expected_version",
48     "fetch_latest_red_version_info",
49 )
50 
51 
52 def safe_delete(pth: Path):
53     if pth.exists():
54         for root, dirs, files in os.walk(str(pth)):
55             os.chmod(root, 0o700)
56 
57             for d in dirs:
58                 os.chmod(os.path.join(root, d), 0o700)
59 
60             for f in files:
61                 os.chmod(os.path.join(root, f), 0o700)
62 
63         shutil.rmtree(str(pth), ignore_errors=True)
64 
65 
66 def _fuzzy_log_filter(record):
67     return record.funcName != "extractWithoutOrder"
68 
69 
70 logging.getLogger().addFilter(_fuzzy_log_filter)
71 
72 
73 async def fuzzy_command_search(
74     ctx: Context,
75     term: Optional[str] = None,
76     *,
77     commands: Optional[Union[AsyncIterator[Command], Iterator[Command]]] = None,
78     min_score: int = 80,
79 ) -> Optional[List[Command]]:
80     """Search for commands which are similar in name to the one invoked.
81 
82     Returns a maximum of 5 commands which must all be at least matched
83     greater than ``min_score``.
84 
85     Parameters
86     ----------
87     ctx : `commands.Context <redbot.core.commands.Context>`
88         The command invocation context.
89     term : Optional[str]
90         The name of the invoked command. If ``None``,
91         `Context.invoked_with` will be used instead.
92     commands : Optional[Union[AsyncIterator[commands.Command], Iterator[commands.Command]]]
93         The commands available to choose from when doing a fuzzy match.
94         When omitted, `Bot.walk_commands` will be used instead.
95     min_score : int
96         The minimum score for matched commands to reach. Defaults to 80.
97 
98     Returns
99     -------
100     Optional[List[`commands.Command <redbot.core.commands.Command>`]]
101         A list of commands which were fuzzily matched with the invoked
102         command.
103 
104     """
105     if ctx.guild is not None:
106         enabled = await ctx.bot._config.guild(ctx.guild).fuzzy()
107     else:
108         enabled = await ctx.bot._config.fuzzy()
109 
110     if not enabled:
111         return None
112 
113     if term is None:
114         term = ctx.invoked_with
115 
116     # If the term is an alias or CC, we don't want to send a supplementary fuzzy search.
117     alias_cog = ctx.bot.get_cog("Alias")
118     if alias_cog is not None:
119         alias = await alias_cog._aliases.get_alias(ctx.guild, term)
120 
121         if alias:
122             return None
123     customcom_cog = ctx.bot.get_cog("CustomCommands")
124     if customcom_cog is not None:
125         cmd_obj = customcom_cog.commandobj
126 
127         try:
128             await cmd_obj.get(ctx.message, term)
129         except:
130             pass
131         else:
132             return None
133 
134     if commands is None:
135         choices = set(ctx.bot.walk_commands())
136     elif isinstance(commands, collections.abc.AsyncIterator):
137         choices = {c async for c in commands}
138     else:
139         choices = set(commands)
140 
141     # Do the scoring. `extracted` is a list of tuples in the form `(command, score)`
142     extracted = process.extract(term, choices, limit=5, scorer=fuzz.QRatio)
143     if not extracted:
144         return None
145 
146     # Filter through the fuzzy-matched commands.
147     matched_commands = []
148     for command, score in extracted:
149         if score < min_score:
150             # Since the list is in decreasing order of score, we can exit early.
151             break
152         if await command.can_see(ctx):
153             matched_commands.append(command)
154 
155     return matched_commands
156 
157 
158 async def format_fuzzy_results(
159     ctx: Context, matched_commands: List[Command], *, embed: Optional[bool] = None
160 ) -> Union[str, discord.Embed]:
161     """Format the result of a fuzzy command search.
162 
163     Parameters
164     ----------
165     ctx : `commands.Context <redbot.core.commands.Context>`
166         The context in which this result is being displayed.
167     matched_commands : List[`commands.Command <redbot.core.commands.Command>`]
168         A list of commands which have been matched by the fuzzy search, sorted
169         in order of decreasing similarity.
170     embed : bool
171         Whether or not the result should be an embed. If set to ``None``, this
172         will default to the result of `ctx.embed_requested`.
173 
174     Returns
175     -------
176     Union[str, discord.Embed]
177         The formatted results.
178 
179     """
180     if embed is not False and (embed is True or await ctx.embed_requested()):
181         lines = []
182         for cmd in matched_commands:
183             short_doc = cmd.format_shortdoc_for_context(ctx)
184             lines.append(f"**{ctx.clean_prefix}{cmd.qualified_name}** {short_doc}")
185         return discord.Embed(
186             title="Perhaps you wanted one of these?",
187             colour=await ctx.embed_colour(),
188             description="\n".join(lines),
189         )
190     else:
191         lines = []
192         for cmd in matched_commands:
193             short_doc = cmd.format_shortdoc_for_context(ctx)
194             lines.append(f"{ctx.clean_prefix}{cmd.qualified_name} -- {short_doc}")
195         return "Perhaps you wanted one of these? " + box("\n".join(lines), lang="vhdl")
196 
197 
198 async def create_backup(dest: Path = Path.home()) -> Optional[Path]:
199     data_path = Path(data_manager.core_data_path().parent)
200     if not data_path.exists():
201         return None
202 
203     dest.mkdir(parents=True, exist_ok=True)
204     timestr = datetime.utcnow().strftime("%Y-%m-%dT%H-%M-%S")
205     backup_fpath = dest / f"redv3_{data_manager.instance_name}_{timestr}.tar.gz"
206 
207     to_backup = []
208     exclusions = [
209         "__pycache__",
210         "Lavalink.jar",
211         os.path.join("Downloader", "lib"),
212         os.path.join("CogManager", "cogs"),
213         os.path.join("RepoManager", "repos"),
214     ]
215 
216     # Avoiding circular imports
217     from ...cogs.downloader.repo_manager import RepoManager
218 
219     repo_mgr = RepoManager()
220     await repo_mgr.initialize()
221     repo_output = []
222     for repo in repo_mgr.repos:
223         repo_output.append({"url": repo.url, "name": repo.name, "branch": repo.branch})
224     repos_file = data_path / "cogs" / "RepoManager" / "repos.json"
225     with repos_file.open("w") as fs:
226         json.dump(repo_output, fs, indent=4)
227     instance_file = data_path / "instance.json"
228     with instance_file.open("w") as fs:
229         json.dump({data_manager.instance_name: data_manager.basic_config}, fs, indent=4)
230     for f in data_path.glob("**/*"):
231         if not any(ex in str(f) for ex in exclusions) and f.is_file():
232             to_backup.append(f)
233 
234     with tarfile.open(str(backup_fpath), "w:gz") as tar:
235         for f in to_backup:
236             tar.add(str(f), arcname=str(f.relative_to(data_path)), recursive=False)
237     return backup_fpath
238 
239 
240 # this might be worth moving to `bot.send_to_owners` at later date
241 
242 
243 async def send_to_owners_with_preprocessor(
244     bot: Red,
245     content: str,
246     *,
247     content_preprocessor: Optional[
248         Callable[[Red, discord.abc.Messageable, str], Awaitable[str]]
249     ] = None,
250     **kwargs,
251 ):
252     """
253     This sends something to all owners and their configured extra destinations.
254 
255     This acts the same as `Red.send_to_owners`, with
256     one added keyword argument as detailed below in *Other Parameters*.
257 
258     Other Parameters
259     ----------------
260     content_preprocessor: Optional[Callable[[Red, discord.abc.Messageable, str], Awaitable[str]]]
261         Optional async function that takes
262         bot object, owner notification destination and message content
263         and returns the content that should be sent to given location.
264     """
265     destinations = await bot.get_owner_notification_destinations()
266 
267     async def wrapped_send(bot, location, content=None, preprocessor=None, **kwargs):
268         try:
269             if preprocessor is not None:
270                 content = await preprocessor(bot, location, content)
271             await location.send(content, **kwargs)
272         except Exception as _exc:
273             main_log.error(
274                 "I could not send an owner notification to %s (%s)",
275                 location,
276                 location.id,
277                 exc_info=_exc,
278             )
279 
280     sends = [wrapped_send(bot, d, content, content_preprocessor, **kwargs) for d in destinations]
281     await asyncio.gather(*sends)
282 
283 
284 async def send_to_owners_with_prefix_replaced(bot: Red, content: str, **kwargs):
285     """
286     This sends something to all owners and their configured extra destinations.
287 
288     This acts the same as `Red.send_to_owners`, with one addition - `[p]` in ``content`` argument
289     is replaced with a clean prefix for each specific destination.
290     """
291 
292     async def preprocessor(bot: Red, destination: discord.abc.Messageable, content: str) -> str:
293         prefixes = await bot.get_valid_prefixes(getattr(destination, "guild", None))
294         prefix = re.sub(
295             rf"<@!?{bot.user.id}>", f"@{bot.user.name}".replace("\\", r"\\"), prefixes[0]
296         )
297         return content.replace("[p]", prefix)
298 
299     await send_to_owners_with_preprocessor(bot, content, content_preprocessor=preprocessor)
300 
301 
302 def expected_version(current: str, expected: str) -> bool:
303     # `pkg_resources` needs a regular requirement string, so "x" serves as requirement's name here
304     return current in pkg_resources.Requirement.parse(f"x{expected}")
305 
306 
307 async def fetch_latest_red_version_info() -> Tuple[Optional[VersionInfo], Optional[str]]:
308     try:
309         async with aiohttp.ClientSession() as session:
310             async with session.get("https://pypi.org/pypi/Red-DiscordBot/json") as r:
311                 data = await r.json()
312     except (aiohttp.ClientError, asyncio.TimeoutError):
313         return None, None
314     else:
315         release = VersionInfo.from_str(data["info"]["version"])
316         required_python = data["info"]["requires_python"]
317 
318         return release, required_python
319
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/redbot/core/utils/_internal_utils.py b/redbot/core/utils/_internal_utils.py
--- a/redbot/core/utils/_internal_utils.py
+++ b/redbot/core/utils/_internal_utils.py
@@ -211,6 +211,7 @@
         os.path.join("Downloader", "lib"),
         os.path.join("CogManager", "cogs"),
         os.path.join("RepoManager", "repos"),
+        os.path.join("Audio", "logs"),
     ]
 
     # Avoiding circular imports

{"golden_diff": "diff --git a/redbot/core/utils/_internal_utils.py b/redbot/core/utils/_internal_utils.py\n--- a/redbot/core/utils/_internal_utils.py\n+++ b/redbot/core/utils/_internal_utils.py\n@@ -211,6 +211,7 @@\n         os.path.join(\"Downloader\", \"lib\"),\n         os.path.join(\"CogManager\", \"cogs\"),\n         os.path.join(\"RepoManager\", \"repos\"),\n+        os.path.join(\"Audio\", \"logs\"),\n     ]\n \n     # Avoiding circular imports\n", "issue": "Stop backing up lavalink logs\nLavalink logs are host specific, stop backing them up.\n", "before_files": [{"content": "from __future__ import annotations\n\nimport asyncio\nimport collections.abc\nimport json\nimport logging\nimport os\nimport re\nimport shutil\nimport tarfile\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import (\n    AsyncIterator,\n    Awaitable,\n    Callable,\n    Iterator,\n    List,\n    Optional,\n    Union,\n    TYPE_CHECKING,\n    Tuple,\n)\n\nimport aiohttp\nimport discord\nimport pkg_resources\nfrom fuzzywuzzy import fuzz, process\nfrom redbot import VersionInfo\n\nfrom redbot.core import data_manager\nfrom redbot.core.utils.chat_formatting import box\n\nif TYPE_CHECKING:\n    from redbot.core.bot import Red\n    from redbot.core.commands import Command, Context\n\nmain_log = logging.getLogger(\"red\")\n\n__all__ = (\n    \"safe_delete\",\n    \"fuzzy_command_search\",\n    \"format_fuzzy_results\",\n    \"create_backup\",\n    \"send_to_owners_with_preprocessor\",\n    \"send_to_owners_with_prefix_replaced\",\n    \"expected_version\",\n    \"fetch_latest_red_version_info\",\n)\n\n\ndef safe_delete(pth: Path):\n    if pth.exists():\n        for root, dirs, files in os.walk(str(pth)):\n            os.chmod(root, 0o700)\n\n            for d in dirs:\n                os.chmod(os.path.join(root, d), 0o700)\n\n            for f in files:\n                os.chmod(os.path.join(root, f), 0o700)\n\n        shutil.rmtree(str(pth), ignore_errors=True)\n\n\ndef _fuzzy_log_filter(record):\n    return record.funcName != \"extractWithoutOrder\"\n\n\nlogging.getLogger().addFilter(_fuzzy_log_filter)\n\n\nasync def fuzzy_command_search(\n    ctx: Context,\n    term: Optional[str] = None,\n    *,\n    commands: Optional[Union[AsyncIterator[Command], Iterator[Command]]] = None,\n    min_score: int = 80,\n) -> Optional[List[Command]]:\n    \"\"\"Search for commands which are similar in name to the one invoked.\n\n    Returns a maximum of 5 commands which must all be at least matched\n    greater than ``min_score``.\n\n    Parameters\n    ----------\n    ctx : `commands.Context <redbot.core.commands.Context>`\n        The command invocation context.\n    term : Optional[str]\n        The name of the invoked command. If ``None``,\n        `Context.invoked_with` will be used instead.\n    commands : Optional[Union[AsyncIterator[commands.Command], Iterator[commands.Command]]]\n        The commands available to choose from when doing a fuzzy match.\n        When omitted, `Bot.walk_commands` will be used instead.\n    min_score : int\n        The minimum score for matched commands to reach. Defaults to 80.\n\n    Returns\n    -------\n    Optional[List[`commands.Command <redbot.core.commands.Command>`]]\n        A list of commands which were fuzzily matched with the invoked\n        command.\n\n    \"\"\"\n    if ctx.guild is not None:\n        enabled = await ctx.bot._config.guild(ctx.guild).fuzzy()\n    else:\n        enabled = await ctx.bot._config.fuzzy()\n\n    if not enabled:\n        return None\n\n    if term is None:\n        term = ctx.invoked_with\n\n    # If the term is an alias or CC, we don't want to send a supplementary fuzzy search.\n    alias_cog = ctx.bot.get_cog(\"Alias\")\n    if alias_cog is not None:\n        alias = await alias_cog._aliases.get_alias(ctx.guild, term)\n\n        if alias:\n            return None\n    customcom_cog = ctx.bot.get_cog(\"CustomCommands\")\n    if customcom_cog is not None:\n        cmd_obj = customcom_cog.commandobj\n\n        try:\n            await cmd_obj.get(ctx.message, term)\n        except:\n            pass\n        else:\n            return None\n\n    if commands is None:\n        choices = set(ctx.bot.walk_commands())\n    elif isinstance(commands, collections.abc.AsyncIterator):\n        choices = {c async for c in commands}\n    else:\n        choices = set(commands)\n\n    # Do the scoring. `extracted` is a list of tuples in the form `(command, score)`\n    extracted = process.extract(term, choices, limit=5, scorer=fuzz.QRatio)\n    if not extracted:\n        return None\n\n    # Filter through the fuzzy-matched commands.\n    matched_commands = []\n    for command, score in extracted:\n        if score < min_score:\n            # Since the list is in decreasing order of score, we can exit early.\n            break\n        if await command.can_see(ctx):\n            matched_commands.append(command)\n\n    return matched_commands\n\n\nasync def format_fuzzy_results(\n    ctx: Context, matched_commands: List[Command], *, embed: Optional[bool] = None\n) -> Union[str, discord.Embed]:\n    \"\"\"Format the result of a fuzzy command search.\n\n    Parameters\n    ----------\n    ctx : `commands.Context <redbot.core.commands.Context>`\n        The context in which this result is being displayed.\n    matched_commands : List[`commands.Command <redbot.core.commands.Command>`]\n        A list of commands which have been matched by the fuzzy search, sorted\n        in order of decreasing similarity.\n    embed : bool\n        Whether or not the result should be an embed. If set to ``None``, this\n        will default to the result of `ctx.embed_requested`.\n\n    Returns\n    -------\n    Union[str, discord.Embed]\n        The formatted results.\n\n    \"\"\"\n    if embed is not False and (embed is True or await ctx.embed_requested()):\n        lines = []\n        for cmd in matched_commands:\n            short_doc = cmd.format_shortdoc_for_context(ctx)\n            lines.append(f\"**{ctx.clean_prefix}{cmd.qualified_name}** {short_doc}\")\n        return discord.Embed(\n            title=\"Perhaps you wanted one of these?\",\n            colour=await ctx.embed_colour(),\n            description=\"\\n\".join(lines),\n        )\n    else:\n        lines = []\n        for cmd in matched_commands:\n            short_doc = cmd.format_shortdoc_for_context(ctx)\n            lines.append(f\"{ctx.clean_prefix}{cmd.qualified_name} -- {short_doc}\")\n        return \"Perhaps you wanted one of these? \" + box(\"\\n\".join(lines), lang=\"vhdl\")\n\n\nasync def create_backup(dest: Path = Path.home()) -> Optional[Path]:\n    data_path = Path(data_manager.core_data_path().parent)\n    if not data_path.exists():\n        return None\n\n    dest.mkdir(parents=True, exist_ok=True)\n    timestr = datetime.utcnow().strftime(\"%Y-%m-%dT%H-%M-%S\")\n    backup_fpath = dest / f\"redv3_{data_manager.instance_name}_{timestr}.tar.gz\"\n\n    to_backup = []\n    exclusions = [\n        \"__pycache__\",\n        \"Lavalink.jar\",\n        os.path.join(\"Downloader\", \"lib\"),\n        os.path.join(\"CogManager\", \"cogs\"),\n        os.path.join(\"RepoManager\", \"repos\"),\n    ]\n\n    # Avoiding circular imports\n    from ...cogs.downloader.repo_manager import RepoManager\n\n    repo_mgr = RepoManager()\n    await repo_mgr.initialize()\n    repo_output = []\n    for repo in repo_mgr.repos:\n        repo_output.append({\"url\": repo.url, \"name\": repo.name, \"branch\": repo.branch})\n    repos_file = data_path / \"cogs\" / \"RepoManager\" / \"repos.json\"\n    with repos_file.open(\"w\") as fs:\n        json.dump(repo_output, fs, indent=4)\n    instance_file = data_path / \"instance.json\"\n    with instance_file.open(\"w\") as fs:\n        json.dump({data_manager.instance_name: data_manager.basic_config}, fs, indent=4)\n    for f in data_path.glob(\"**/*\"):\n        if not any(ex in str(f) for ex in exclusions) and f.is_file():\n            to_backup.append(f)\n\n    with tarfile.open(str(backup_fpath), \"w:gz\") as tar:\n        for f in to_backup:\n            tar.add(str(f), arcname=str(f.relative_to(data_path)), recursive=False)\n    return backup_fpath\n\n\n# this might be worth moving to `bot.send_to_owners` at later date\n\n\nasync def send_to_owners_with_preprocessor(\n    bot: Red,\n    content: str,\n    *,\n    content_preprocessor: Optional[\n        Callable[[Red, discord.abc.Messageable, str], Awaitable[str]]\n    ] = None,\n    **kwargs,\n):\n    \"\"\"\n    This sends something to all owners and their configured extra destinations.\n\n    This acts the same as `Red.send_to_owners`, with\n    one added keyword argument as detailed below in *Other Parameters*.\n\n    Other Parameters\n    ----------------\n    content_preprocessor: Optional[Callable[[Red, discord.abc.Messageable, str], Awaitable[str]]]\n        Optional async function that takes\n        bot object, owner notification destination and message content\n        and returns the content that should be sent to given location.\n    \"\"\"\n    destinations = await bot.get_owner_notification_destinations()\n\n    async def wrapped_send(bot, location, content=None, preprocessor=None, **kwargs):\n        try:\n            if preprocessor is not None:\n                content = await preprocessor(bot, location, content)\n            await location.send(content, **kwargs)\n        except Exception as _exc:\n            main_log.error(\n                \"I could not send an owner notification to %s (%s)\",\n                location,\n                location.id,\n                exc_info=_exc,\n            )\n\n    sends = [wrapped_send(bot, d, content, content_preprocessor, **kwargs) for d in destinations]\n    await asyncio.gather(*sends)\n\n\nasync def send_to_owners_with_prefix_replaced(bot: Red, content: str, **kwargs):\n    \"\"\"\n    This sends something to all owners and their configured extra destinations.\n\n    This acts the same as `Red.send_to_owners`, with one addition - `[p]` in ``content`` argument\n    is replaced with a clean prefix for each specific destination.\n    \"\"\"\n\n    async def preprocessor(bot: Red, destination: discord.abc.Messageable, content: str) -> str:\n        prefixes = await bot.get_valid_prefixes(getattr(destination, \"guild\", None))\n        prefix = re.sub(\n            rf\"<@!?{bot.user.id}>\", f\"@{bot.user.name}\".replace(\"\\\\\", r\"\\\\\"), prefixes[0]\n        )\n        return content.replace(\"[p]\", prefix)\n\n    await send_to_owners_with_preprocessor(bot, content, content_preprocessor=preprocessor)\n\n\ndef expected_version(current: str, expected: str) -> bool:\n    # `pkg_resources` needs a regular requirement string, so \"x\" serves as requirement's name here\n    return current in pkg_resources.Requirement.parse(f\"x{expected}\")\n\n\nasync def fetch_latest_red_version_info() -> Tuple[Optional[VersionInfo], Optional[str]]:\n    try:\n        async with aiohttp.ClientSession() as session:\n            async with session.get(\"https://pypi.org/pypi/Red-DiscordBot/json\") as r:\n                data = await r.json()\n    except (aiohttp.ClientError, asyncio.TimeoutError):\n        return None, None\n    else:\n        release = VersionInfo.from_str(data[\"info\"][\"version\"])\n        required_python = data[\"info\"][\"requires_python\"]\n\n        return release, required_python\n", "path": "redbot/core/utils/_internal_utils.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport asyncio\nimport collections.abc\nimport json\nimport logging\nimport os\nimport re\nimport shutil\nimport tarfile\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import (\n    AsyncIterator,\n    Awaitable,\n    Callable,\n    Iterator,\n    List,\n    Optional,\n    Union,\n    TYPE_CHECKING,\n    Tuple,\n)\n\nimport aiohttp\nimport discord\nimport pkg_resources\nfrom fuzzywuzzy import fuzz, process\nfrom redbot import VersionInfo\n\nfrom redbot.core import data_manager\nfrom redbot.core.utils.chat_formatting import box\n\nif TYPE_CHECKING:\n    from redbot.core.bot import Red\n    from redbot.core.commands import Command, Context\n\nmain_log = logging.getLogger(\"red\")\n\n__all__ = (\n    \"safe_delete\",\n    \"fuzzy_command_search\",\n    \"format_fuzzy_results\",\n    \"create_backup\",\n    \"send_to_owners_with_preprocessor\",\n    \"send_to_owners_with_prefix_replaced\",\n    \"expected_version\",\n    \"fetch_latest_red_version_info\",\n)\n\n\ndef safe_delete(pth: Path):\n    if pth.exists():\n        for root, dirs, files in os.walk(str(pth)):\n            os.chmod(root, 0o700)\n\n            for d in dirs:\n                os.chmod(os.path.join(root, d), 0o700)\n\n            for f in files:\n                os.chmod(os.path.join(root, f), 0o700)\n\n        shutil.rmtree(str(pth), ignore_errors=True)\n\n\ndef _fuzzy_log_filter(record):\n    return record.funcName != \"extractWithoutOrder\"\n\n\nlogging.getLogger().addFilter(_fuzzy_log_filter)\n\n\nasync def fuzzy_command_search(\n    ctx: Context,\n    term: Optional[str] = None,\n    *,\n    commands: Optional[Union[AsyncIterator[Command], Iterator[Command]]] = None,\n    min_score: int = 80,\n) -> Optional[List[Command]]:\n    \"\"\"Search for commands which are similar in name to the one invoked.\n\n    Returns a maximum of 5 commands which must all be at least matched\n    greater than ``min_score``.\n\n    Parameters\n    ----------\n    ctx : `commands.Context <redbot.core.commands.Context>`\n        The command invocation context.\n    term : Optional[str]\n        The name of the invoked command. If ``None``,\n        `Context.invoked_with` will be used instead.\n    commands : Optional[Union[AsyncIterator[commands.Command], Iterator[commands.Command]]]\n        The commands available to choose from when doing a fuzzy match.\n        When omitted, `Bot.walk_commands` will be used instead.\n    min_score : int\n        The minimum score for matched commands to reach. Defaults to 80.\n\n    Returns\n    -------\n    Optional[List[`commands.Command <redbot.core.commands.Command>`]]\n        A list of commands which were fuzzily matched with the invoked\n        command.\n\n    \"\"\"\n    if ctx.guild is not None:\n        enabled = await ctx.bot._config.guild(ctx.guild).fuzzy()\n    else:\n        enabled = await ctx.bot._config.fuzzy()\n\n    if not enabled:\n        return None\n\n    if term is None:\n        term = ctx.invoked_with\n\n    # If the term is an alias or CC, we don't want to send a supplementary fuzzy search.\n    alias_cog = ctx.bot.get_cog(\"Alias\")\n    if alias_cog is not None:\n        alias = await alias_cog._aliases.get_alias(ctx.guild, term)\n\n        if alias:\n            return None\n    customcom_cog = ctx.bot.get_cog(\"CustomCommands\")\n    if customcom_cog is not None:\n        cmd_obj = customcom_cog.commandobj\n\n        try:\n            await cmd_obj.get(ctx.message, term)\n        except:\n            pass\n        else:\n            return None\n\n    if commands is None:\n        choices = set(ctx.bot.walk_commands())\n    elif isinstance(commands, collections.abc.AsyncIterator):\n        choices = {c async for c in commands}\n    else:\n        choices = set(commands)\n\n    # Do the scoring. `extracted` is a list of tuples in the form `(command, score)`\n    extracted = process.extract(term, choices, limit=5, scorer=fuzz.QRatio)\n    if not extracted:\n        return None\n\n    # Filter through the fuzzy-matched commands.\n    matched_commands = []\n    for command, score in extracted:\n        if score < min_score:\n            # Since the list is in decreasing order of score, we can exit early.\n            break\n        if await command.can_see(ctx):\n            matched_commands.append(command)\n\n    return matched_commands\n\n\nasync def format_fuzzy_results(\n    ctx: Context, matched_commands: List[Command], *, embed: Optional[bool] = None\n) -> Union[str, discord.Embed]:\n    \"\"\"Format the result of a fuzzy command search.\n\n    Parameters\n    ----------\n    ctx : `commands.Context <redbot.core.commands.Context>`\n        The context in which this result is being displayed.\n    matched_commands : List[`commands.Command <redbot.core.commands.Command>`]\n        A list of commands which have been matched by the fuzzy search, sorted\n        in order of decreasing similarity.\n    embed : bool\n        Whether or not the result should be an embed. If set to ``None``, this\n        will default to the result of `ctx.embed_requested`.\n\n    Returns\n    -------\n    Union[str, discord.Embed]\n        The formatted results.\n\n    \"\"\"\n    if embed is not False and (embed is True or await ctx.embed_requested()):\n        lines = []\n        for cmd in matched_commands:\n            short_doc = cmd.format_shortdoc_for_context(ctx)\n            lines.append(f\"**{ctx.clean_prefix}{cmd.qualified_name}** {short_doc}\")\n        return discord.Embed(\n            title=\"Perhaps you wanted one of these?\",\n            colour=await ctx.embed_colour(),\n            description=\"\\n\".join(lines),\n        )\n    else:\n        lines = []\n        for cmd in matched_commands:\n            short_doc = cmd.format_shortdoc_for_context(ctx)\n            lines.append(f\"{ctx.clean_prefix}{cmd.qualified_name} -- {short_doc}\")\n        return \"Perhaps you wanted one of these? \" + box(\"\\n\".join(lines), lang=\"vhdl\")\n\n\nasync def create_backup(dest: Path = Path.home()) -> Optional[Path]:\n    data_path = Path(data_manager.core_data_path().parent)\n    if not data_path.exists():\n        return None\n\n    dest.mkdir(parents=True, exist_ok=True)\n    timestr = datetime.utcnow().strftime(\"%Y-%m-%dT%H-%M-%S\")\n    backup_fpath = dest / f\"redv3_{data_manager.instance_name}_{timestr}.tar.gz\"\n\n    to_backup = []\n    exclusions = [\n        \"__pycache__\",\n        \"Lavalink.jar\",\n        os.path.join(\"Downloader\", \"lib\"),\n        os.path.join(\"CogManager\", \"cogs\"),\n        os.path.join(\"RepoManager\", \"repos\"),\n        os.path.join(\"Audio\", \"logs\"),\n    ]\n\n    # Avoiding circular imports\n    from ...cogs.downloader.repo_manager import RepoManager\n\n    repo_mgr = RepoManager()\n    await repo_mgr.initialize()\n    repo_output = []\n    for repo in repo_mgr.repos:\n        repo_output.append({\"url\": repo.url, \"name\": repo.name, \"branch\": repo.branch})\n    repos_file = data_path / \"cogs\" / \"RepoManager\" / \"repos.json\"\n    with repos_file.open(\"w\") as fs:\n        json.dump(repo_output, fs, indent=4)\n    instance_file = data_path / \"instance.json\"\n    with instance_file.open(\"w\") as fs:\n        json.dump({data_manager.instance_name: data_manager.basic_config}, fs, indent=4)\n    for f in data_path.glob(\"**/*\"):\n        if not any(ex in str(f) for ex in exclusions) and f.is_file():\n            to_backup.append(f)\n\n    with tarfile.open(str(backup_fpath), \"w:gz\") as tar:\n        for f in to_backup:\n            tar.add(str(f), arcname=str(f.relative_to(data_path)), recursive=False)\n    return backup_fpath\n\n\n# this might be worth moving to `bot.send_to_owners` at later date\n\n\nasync def send_to_owners_with_preprocessor(\n    bot: Red,\n    content: str,\n    *,\n    content_preprocessor: Optional[\n        Callable[[Red, discord.abc.Messageable, str], Awaitable[str]]\n    ] = None,\n    **kwargs,\n):\n    \"\"\"\n    This sends something to all owners and their configured extra destinations.\n\n    This acts the same as `Red.send_to_owners`, with\n    one added keyword argument as detailed below in *Other Parameters*.\n\n    Other Parameters\n    ----------------\n    content_preprocessor: Optional[Callable[[Red, discord.abc.Messageable, str], Awaitable[str]]]\n        Optional async function that takes\n        bot object, owner notification destination and message content\n        and returns the content that should be sent to given location.\n    \"\"\"\n    destinations = await bot.get_owner_notification_destinations()\n\n    async def wrapped_send(bot, location, content=None, preprocessor=None, **kwargs):\n        try:\n            if preprocessor is not None:\n                content = await preprocessor(bot, location, content)\n            await location.send(content, **kwargs)\n        except Exception as _exc:\n            main_log.error(\n                \"I could not send an owner notification to %s (%s)\",\n                location,\n                location.id,\n                exc_info=_exc,\n            )\n\n    sends = [wrapped_send(bot, d, content, content_preprocessor, **kwargs) for d in destinations]\n    await asyncio.gather(*sends)\n\n\nasync def send_to_owners_with_prefix_replaced(bot: Red, content: str, **kwargs):\n    \"\"\"\n    This sends something to all owners and their configured extra destinations.\n\n    This acts the same as `Red.send_to_owners`, with one addition - `[p]` in ``content`` argument\n    is replaced with a clean prefix for each specific destination.\n    \"\"\"\n\n    async def preprocessor(bot: Red, destination: discord.abc.Messageable, content: str) -> str:\n        prefixes = await bot.get_valid_prefixes(getattr(destination, \"guild\", None))\n        prefix = re.sub(\n            rf\"<@!?{bot.user.id}>\", f\"@{bot.user.name}\".replace(\"\\\\\", r\"\\\\\"), prefixes[0]\n        )\n        return content.replace(\"[p]\", prefix)\n\n    await send_to_owners_with_preprocessor(bot, content, content_preprocessor=preprocessor)\n\n\ndef expected_version(current: str, expected: str) -> bool:\n    # `pkg_resources` needs a regular requirement string, so \"x\" serves as requirement's name here\n    return current in pkg_resources.Requirement.parse(f\"x{expected}\")\n\n\nasync def fetch_latest_red_version_info() -> Tuple[Optional[VersionInfo], Optional[str]]:\n    try:\n        async with aiohttp.ClientSession() as session:\n            async with session.get(\"https://pypi.org/pypi/Red-DiscordBot/json\") as r:\n                data = await r.json()\n    except (aiohttp.ClientError, asyncio.TimeoutError):\n        return None, None\n    else:\n        release = VersionInfo.from_str(data[\"info\"][\"version\"])\n        required_python = data[\"info\"][\"requires_python\"]\n\n        return release, required_python\n", "path": "redbot/core/utils/_internal_utils.py"}]}