We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Reddit cog does not escape markdown in post titles
Discord markdown in reddit post titles is left unhandled and can sometimes break the links:
![redditpydisbot](https://i.imgur.com/ut0gzFQ.png)

For the basic markdown passing it through d.py's `escape_markdown` should work, but from a quick look I haven't found a way to escape brackets in post titles, which breaks the text links. A replacement with similar unicode chars is an option

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `bot/cogs/reddit.py`
Content:
```
1 import asyncio
2 import logging
3 import random
4 import textwrap
5 from collections import namedtuple
6 from datetime import datetime, timedelta
7 from typing import List
8 
9 from aiohttp import BasicAuth, ClientError
10 from discord import Colour, Embed, TextChannel
11 from discord.ext.commands import Cog, Context, group
12 from discord.ext.tasks import loop
13 
14 from bot.bot import Bot
15 from bot.constants import Channels, ERROR_REPLIES, Emojis, Reddit as RedditConfig, STAFF_ROLES, Webhooks
16 from bot.converters import Subreddit
17 from bot.decorators import with_role
18 from bot.pagination import LinePaginator
19 from bot.utils.messages import sub_clyde
20 
21 log = logging.getLogger(__name__)
22 
23 AccessToken = namedtuple("AccessToken", ["token", "expires_at"])
24 
25 
26 class Reddit(Cog):
27     """Track subreddit posts and show detailed statistics about them."""
28 
29     HEADERS = {"User-Agent": "python3:python-discord/bot:1.0.0 (by /u/PythonDiscord)"}
30     URL = "https://www.reddit.com"
31     OAUTH_URL = "https://oauth.reddit.com"
32     MAX_RETRIES = 3
33 
34     def __init__(self, bot: Bot):
35         self.bot = bot
36 
37         self.webhook = None
38         self.access_token = None
39         self.client_auth = BasicAuth(RedditConfig.client_id, RedditConfig.secret)
40 
41         bot.loop.create_task(self.init_reddit_ready())
42         self.auto_poster_loop.start()
43 
44     def cog_unload(self) -> None:
45         """Stop the loop task and revoke the access token when the cog is unloaded."""
46         self.auto_poster_loop.cancel()
47         if self.access_token and self.access_token.expires_at > datetime.utcnow():
48             asyncio.create_task(self.revoke_access_token())
49 
50     async def init_reddit_ready(self) -> None:
51         """Sets the reddit webhook when the cog is loaded."""
52         await self.bot.wait_until_guild_available()
53         if not self.webhook:
54             self.webhook = await self.bot.fetch_webhook(Webhooks.reddit)
55 
56     @property
57     def channel(self) -> TextChannel:
58         """Get the #reddit channel object from the bot's cache."""
59         return self.bot.get_channel(Channels.reddit)
60 
61     async def get_access_token(self) -> None:
62         """
63         Get a Reddit API OAuth2 access token and assign it to self.access_token.
64 
65         A token is valid for 1 hour. There will be MAX_RETRIES to get a token, after which the cog
66         will be unloaded and a ClientError raised if retrieval was still unsuccessful.
67         """
68         for i in range(1, self.MAX_RETRIES + 1):
69             response = await self.bot.http_session.post(
70                 url=f"{self.URL}/api/v1/access_token",
71                 headers=self.HEADERS,
72                 auth=self.client_auth,
73                 data={
74                     "grant_type": "client_credentials",
75                     "duration": "temporary"
76                 }
77             )
78 
79             if response.status == 200 and response.content_type == "application/json":
80                 content = await response.json()
81                 expiration = int(content["expires_in"]) - 60  # Subtract 1 minute for leeway.
82                 self.access_token = AccessToken(
83                     token=content["access_token"],
84                     expires_at=datetime.utcnow() + timedelta(seconds=expiration)
85                 )
86 
87                 log.debug(f"New token acquired; expires on UTC {self.access_token.expires_at}")
88                 return
89             else:
90                 log.debug(
91                     f"Failed to get an access token: "
92                     f"status {response.status} & content type {response.content_type}; "
93                     f"retrying ({i}/{self.MAX_RETRIES})"
94                 )
95 
96             await asyncio.sleep(3)
97 
98         self.bot.remove_cog(self.qualified_name)
99         raise ClientError("Authentication with the Reddit API failed. Unloading the cog.")
100 
101     async def revoke_access_token(self) -> None:
102         """
103         Revoke the OAuth2 access token for the Reddit API.
104 
105         For security reasons, it's good practice to revoke the token when it's no longer being used.
106         """
107         response = await self.bot.http_session.post(
108             url=f"{self.URL}/api/v1/revoke_token",
109             headers=self.HEADERS,
110             auth=self.client_auth,
111             data={
112                 "token": self.access_token.token,
113                 "token_type_hint": "access_token"
114             }
115         )
116 
117         if response.status == 204 and response.content_type == "application/json":
118             self.access_token = None
119         else:
120             log.warning(f"Unable to revoke access token: status {response.status}.")
121 
122     async def fetch_posts(self, route: str, *, amount: int = 25, params: dict = None) -> List[dict]:
123         """A helper method to fetch a certain amount of Reddit posts at a given route."""
124         # Reddit's JSON responses only provide 25 posts at most.
125         if not 25 >= amount > 0:
126             raise ValueError("Invalid amount of subreddit posts requested.")
127 
128         # Renew the token if necessary.
129         if not self.access_token or self.access_token.expires_at < datetime.utcnow():
130             await self.get_access_token()
131 
132         url = f"{self.OAUTH_URL}/{route}"
133         for _ in range(self.MAX_RETRIES):
134             response = await self.bot.http_session.get(
135                 url=url,
136                 headers={**self.HEADERS, "Authorization": f"bearer {self.access_token.token}"},
137                 params=params
138             )
139             if response.status == 200 and response.content_type == 'application/json':
140                 # Got appropriate response - process and return.
141                 content = await response.json()
142                 posts = content["data"]["children"]
143                 return posts[:amount]
144 
145             await asyncio.sleep(3)
146 
147         log.debug(f"Invalid response from: {url} - status code {response.status}, mimetype {response.content_type}")
148         return list()  # Failed to get appropriate response within allowed number of retries.
149 
150     async def get_top_posts(self, subreddit: Subreddit, time: str = "all", amount: int = 5) -> Embed:
151         """
152         Get the top amount of posts for a given subreddit within a specified timeframe.
153 
154         A time of "all" will get posts from all time, "day" will get top daily posts and "week" will get the top
155         weekly posts.
156 
157         The amount should be between 0 and 25 as Reddit's JSON requests only provide 25 posts at most.
158         """
159         embed = Embed(description="")
160 
161         posts = await self.fetch_posts(
162             route=f"{subreddit}/top",
163             amount=amount,
164             params={"t": time}
165         )
166 
167         if not posts:
168             embed.title = random.choice(ERROR_REPLIES)
169             embed.colour = Colour.red()
170             embed.description = (
171                 "Sorry! We couldn't find any posts from that subreddit. "
172                 "If this problem persists, please let us know."
173             )
174 
175             return embed
176 
177         for post in posts:
178             data = post["data"]
179 
180             text = data["selftext"]
181             if text:
182                 text = textwrap.shorten(text, width=128, placeholder="...")
183                 text += "\n"  # Add newline to separate embed info
184 
185             ups = data["ups"]
186             comments = data["num_comments"]
187             author = data["author"]
188 
189             title = textwrap.shorten(data["title"], width=64, placeholder="...")
190             link = self.URL + data["permalink"]
191 
192             embed.description += (
193                 f"**[{title}]({link})**\n"
194                 f"{text}"
195                 f"{Emojis.upvotes} {ups} {Emojis.comments} {comments} {Emojis.user} {author}\n\n"
196             )
197 
198         embed.colour = Colour.blurple()
199         return embed
200 
201     @loop()
202     async def auto_poster_loop(self) -> None:
203         """Post the top 5 posts daily, and the top 5 posts weekly."""
204         # once we upgrade to d.py 1.3 this can be removed and the loop can use the `time=datetime.time.min` parameter
205         now = datetime.utcnow()
206         tomorrow = now + timedelta(days=1)
207         midnight_tomorrow = tomorrow.replace(hour=0, minute=0, second=0)
208         seconds_until = (midnight_tomorrow - now).total_seconds()
209 
210         await asyncio.sleep(seconds_until)
211 
212         await self.bot.wait_until_guild_available()
213         if not self.webhook:
214             await self.bot.fetch_webhook(Webhooks.reddit)
215 
216         if datetime.utcnow().weekday() == 0:
217             await self.top_weekly_posts()
218             # if it's a monday send the top weekly posts
219 
220         for subreddit in RedditConfig.subreddits:
221             top_posts = await self.get_top_posts(subreddit=subreddit, time="day")
222             username = sub_clyde(f"{subreddit} Top Daily Posts")
223             message = await self.webhook.send(username=username, embed=top_posts, wait=True)
224 
225             if message.channel.is_news():
226                 await message.publish()
227 
228     async def top_weekly_posts(self) -> None:
229         """Post a summary of the top posts."""
230         for subreddit in RedditConfig.subreddits:
231             # Send and pin the new weekly posts.
232             top_posts = await self.get_top_posts(subreddit=subreddit, time="week")
233             username = sub_clyde(f"{subreddit} Top Weekly Posts")
234             message = await self.webhook.send(wait=True, username=username, embed=top_posts)
235 
236             if subreddit.lower() == "r/python":
237                 if not self.channel:
238                     log.warning("Failed to get #reddit channel to remove pins in the weekly loop.")
239                     return
240 
241                 # Remove the oldest pins so that only 12 remain at most.
242                 pins = await self.channel.pins()
243 
244                 while len(pins) >= 12:
245                     await pins[-1].unpin()
246                     del pins[-1]
247 
248                 await message.pin()
249 
250                 if message.channel.is_news():
251                     await message.publish()
252 
253     @group(name="reddit", invoke_without_command=True)
254     async def reddit_group(self, ctx: Context) -> None:
255         """View the top posts from various subreddits."""
256         await ctx.send_help(ctx.command)
257 
258     @reddit_group.command(name="top")
259     async def top_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
260         """Send the top posts of all time from a given subreddit."""
261         async with ctx.typing():
262             embed = await self.get_top_posts(subreddit=subreddit, time="all")
263 
264         await ctx.send(content=f"Here are the top {subreddit} posts of all time!", embed=embed)
265 
266     @reddit_group.command(name="daily")
267     async def daily_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
268         """Send the top posts of today from a given subreddit."""
269         async with ctx.typing():
270             embed = await self.get_top_posts(subreddit=subreddit, time="day")
271 
272         await ctx.send(content=f"Here are today's top {subreddit} posts!", embed=embed)
273 
274     @reddit_group.command(name="weekly")
275     async def weekly_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
276         """Send the top posts of this week from a given subreddit."""
277         async with ctx.typing():
278             embed = await self.get_top_posts(subreddit=subreddit, time="week")
279 
280         await ctx.send(content=f"Here are this week's top {subreddit} posts!", embed=embed)
281 
282     @with_role(*STAFF_ROLES)
283     @reddit_group.command(name="subreddits", aliases=("subs",))
284     async def subreddits_command(self, ctx: Context) -> None:
285         """Send a paginated embed of all the subreddits we're relaying."""
286         embed = Embed()
287         embed.title = "Relayed subreddits."
288         embed.colour = Colour.blurple()
289 
290         await LinePaginator.paginate(
291             RedditConfig.subreddits,
292             ctx, embed,
293             footer_text="Use the reddit commands along with these to view their posts.",
294             empty=False,
295             max_lines=15
296         )
297 
298 
299 def setup(bot: Bot) -> None:
300     """Load the Reddit cog."""
301     if not RedditConfig.secret or not RedditConfig.client_id:
302         log.error("Credentials not provided, cog not loaded.")
303         return
304     bot.add_cog(Reddit(bot))
305
```
--- 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/bot/cogs/reddit.py b/bot/cogs/reddit.py
--- a/bot/cogs/reddit.py
+++ b/bot/cogs/reddit.py
@@ -10,6 +10,7 @@
 from discord import Colour, Embed, TextChannel
 from discord.ext.commands import Cog, Context, group
 from discord.ext.tasks import loop
+from discord.utils import escape_markdown
 
 from bot.bot import Bot
 from bot.constants import Channels, ERROR_REPLIES, Emojis, Reddit as RedditConfig, STAFF_ROLES, Webhooks
@@ -187,6 +188,8 @@
             author = data["author"]
 
             title = textwrap.shorten(data["title"], width=64, placeholder="...")
+            # Normal brackets interfere with Markdown.
+            title = escape_markdown(title).replace("[", "⦋").replace("]", "⦌")
             link = self.URL + data["permalink"]
 
             embed.description += (

{"golden_diff": "diff --git a/bot/cogs/reddit.py b/bot/cogs/reddit.py\n--- a/bot/cogs/reddit.py\n+++ b/bot/cogs/reddit.py\n@@ -10,6 +10,7 @@\n from discord import Colour, Embed, TextChannel\n from discord.ext.commands import Cog, Context, group\n from discord.ext.tasks import loop\n+from discord.utils import escape_markdown\n \n from bot.bot import Bot\n from bot.constants import Channels, ERROR_REPLIES, Emojis, Reddit as RedditConfig, STAFF_ROLES, Webhooks\n@@ -187,6 +188,8 @@\n             author = data[\"author\"]\n \n             title = textwrap.shorten(data[\"title\"], width=64, placeholder=\"...\")\n+            # Normal brackets interfere with Markdown.\n+            title = escape_markdown(title).replace(\"[\", \"\u298b\").replace(\"]\", \"\u298c\")\n             link = self.URL + data[\"permalink\"]\n \n             embed.description += (\n", "issue": "Reddit cog does not escape markdown in post titles\nDiscord markdown in reddit post titles is left unhandled and can sometimes break the links:\r\n![redditpydisbot](https://i.imgur.com/ut0gzFQ.png)\r\n\r\nFor the basic markdown passing it through d.py's `escape_markdown` should work, but from a quick look I haven't found a way to escape brackets in post titles, which breaks the text links. A replacement with similar unicode chars is an option\n", "before_files": [{"content": "import asyncio\nimport logging\nimport random\nimport textwrap\nfrom collections import namedtuple\nfrom datetime import datetime, timedelta\nfrom typing import List\n\nfrom aiohttp import BasicAuth, ClientError\nfrom discord import Colour, Embed, TextChannel\nfrom discord.ext.commands import Cog, Context, group\nfrom discord.ext.tasks import loop\n\nfrom bot.bot import Bot\nfrom bot.constants import Channels, ERROR_REPLIES, Emojis, Reddit as RedditConfig, STAFF_ROLES, Webhooks\nfrom bot.converters import Subreddit\nfrom bot.decorators import with_role\nfrom bot.pagination import LinePaginator\nfrom bot.utils.messages import sub_clyde\n\nlog = logging.getLogger(__name__)\n\nAccessToken = namedtuple(\"AccessToken\", [\"token\", \"expires_at\"])\n\n\nclass Reddit(Cog):\n    \"\"\"Track subreddit posts and show detailed statistics about them.\"\"\"\n\n    HEADERS = {\"User-Agent\": \"python3:python-discord/bot:1.0.0 (by /u/PythonDiscord)\"}\n    URL = \"https://www.reddit.com\"\n    OAUTH_URL = \"https://oauth.reddit.com\"\n    MAX_RETRIES = 3\n\n    def __init__(self, bot: Bot):\n        self.bot = bot\n\n        self.webhook = None\n        self.access_token = None\n        self.client_auth = BasicAuth(RedditConfig.client_id, RedditConfig.secret)\n\n        bot.loop.create_task(self.init_reddit_ready())\n        self.auto_poster_loop.start()\n\n    def cog_unload(self) -> None:\n        \"\"\"Stop the loop task and revoke the access token when the cog is unloaded.\"\"\"\n        self.auto_poster_loop.cancel()\n        if self.access_token and self.access_token.expires_at > datetime.utcnow():\n            asyncio.create_task(self.revoke_access_token())\n\n    async def init_reddit_ready(self) -> None:\n        \"\"\"Sets the reddit webhook when the cog is loaded.\"\"\"\n        await self.bot.wait_until_guild_available()\n        if not self.webhook:\n            self.webhook = await self.bot.fetch_webhook(Webhooks.reddit)\n\n    @property\n    def channel(self) -> TextChannel:\n        \"\"\"Get the #reddit channel object from the bot's cache.\"\"\"\n        return self.bot.get_channel(Channels.reddit)\n\n    async def get_access_token(self) -> None:\n        \"\"\"\n        Get a Reddit API OAuth2 access token and assign it to self.access_token.\n\n        A token is valid for 1 hour. There will be MAX_RETRIES to get a token, after which the cog\n        will be unloaded and a ClientError raised if retrieval was still unsuccessful.\n        \"\"\"\n        for i in range(1, self.MAX_RETRIES + 1):\n            response = await self.bot.http_session.post(\n                url=f\"{self.URL}/api/v1/access_token\",\n                headers=self.HEADERS,\n                auth=self.client_auth,\n                data={\n                    \"grant_type\": \"client_credentials\",\n                    \"duration\": \"temporary\"\n                }\n            )\n\n            if response.status == 200 and response.content_type == \"application/json\":\n                content = await response.json()\n                expiration = int(content[\"expires_in\"]) - 60  # Subtract 1 minute for leeway.\n                self.access_token = AccessToken(\n                    token=content[\"access_token\"],\n                    expires_at=datetime.utcnow() + timedelta(seconds=expiration)\n                )\n\n                log.debug(f\"New token acquired; expires on UTC {self.access_token.expires_at}\")\n                return\n            else:\n                log.debug(\n                    f\"Failed to get an access token: \"\n                    f\"status {response.status} & content type {response.content_type}; \"\n                    f\"retrying ({i}/{self.MAX_RETRIES})\"\n                )\n\n            await asyncio.sleep(3)\n\n        self.bot.remove_cog(self.qualified_name)\n        raise ClientError(\"Authentication with the Reddit API failed. Unloading the cog.\")\n\n    async def revoke_access_token(self) -> None:\n        \"\"\"\n        Revoke the OAuth2 access token for the Reddit API.\n\n        For security reasons, it's good practice to revoke the token when it's no longer being used.\n        \"\"\"\n        response = await self.bot.http_session.post(\n            url=f\"{self.URL}/api/v1/revoke_token\",\n            headers=self.HEADERS,\n            auth=self.client_auth,\n            data={\n                \"token\": self.access_token.token,\n                \"token_type_hint\": \"access_token\"\n            }\n        )\n\n        if response.status == 204 and response.content_type == \"application/json\":\n            self.access_token = None\n        else:\n            log.warning(f\"Unable to revoke access token: status {response.status}.\")\n\n    async def fetch_posts(self, route: str, *, amount: int = 25, params: dict = None) -> List[dict]:\n        \"\"\"A helper method to fetch a certain amount of Reddit posts at a given route.\"\"\"\n        # Reddit's JSON responses only provide 25 posts at most.\n        if not 25 >= amount > 0:\n            raise ValueError(\"Invalid amount of subreddit posts requested.\")\n\n        # Renew the token if necessary.\n        if not self.access_token or self.access_token.expires_at < datetime.utcnow():\n            await self.get_access_token()\n\n        url = f\"{self.OAUTH_URL}/{route}\"\n        for _ in range(self.MAX_RETRIES):\n            response = await self.bot.http_session.get(\n                url=url,\n                headers={**self.HEADERS, \"Authorization\": f\"bearer {self.access_token.token}\"},\n                params=params\n            )\n            if response.status == 200 and response.content_type == 'application/json':\n                # Got appropriate response - process and return.\n                content = await response.json()\n                posts = content[\"data\"][\"children\"]\n                return posts[:amount]\n\n            await asyncio.sleep(3)\n\n        log.debug(f\"Invalid response from: {url} - status code {response.status}, mimetype {response.content_type}\")\n        return list()  # Failed to get appropriate response within allowed number of retries.\n\n    async def get_top_posts(self, subreddit: Subreddit, time: str = \"all\", amount: int = 5) -> Embed:\n        \"\"\"\n        Get the top amount of posts for a given subreddit within a specified timeframe.\n\n        A time of \"all\" will get posts from all time, \"day\" will get top daily posts and \"week\" will get the top\n        weekly posts.\n\n        The amount should be between 0 and 25 as Reddit's JSON requests only provide 25 posts at most.\n        \"\"\"\n        embed = Embed(description=\"\")\n\n        posts = await self.fetch_posts(\n            route=f\"{subreddit}/top\",\n            amount=amount,\n            params={\"t\": time}\n        )\n\n        if not posts:\n            embed.title = random.choice(ERROR_REPLIES)\n            embed.colour = Colour.red()\n            embed.description = (\n                \"Sorry! We couldn't find any posts from that subreddit. \"\n                \"If this problem persists, please let us know.\"\n            )\n\n            return embed\n\n        for post in posts:\n            data = post[\"data\"]\n\n            text = data[\"selftext\"]\n            if text:\n                text = textwrap.shorten(text, width=128, placeholder=\"...\")\n                text += \"\\n\"  # Add newline to separate embed info\n\n            ups = data[\"ups\"]\n            comments = data[\"num_comments\"]\n            author = data[\"author\"]\n\n            title = textwrap.shorten(data[\"title\"], width=64, placeholder=\"...\")\n            link = self.URL + data[\"permalink\"]\n\n            embed.description += (\n                f\"**[{title}]({link})**\\n\"\n                f\"{text}\"\n                f\"{Emojis.upvotes} {ups} {Emojis.comments} {comments} {Emojis.user} {author}\\n\\n\"\n            )\n\n        embed.colour = Colour.blurple()\n        return embed\n\n    @loop()\n    async def auto_poster_loop(self) -> None:\n        \"\"\"Post the top 5 posts daily, and the top 5 posts weekly.\"\"\"\n        # once we upgrade to d.py 1.3 this can be removed and the loop can use the `time=datetime.time.min` parameter\n        now = datetime.utcnow()\n        tomorrow = now + timedelta(days=1)\n        midnight_tomorrow = tomorrow.replace(hour=0, minute=0, second=0)\n        seconds_until = (midnight_tomorrow - now).total_seconds()\n\n        await asyncio.sleep(seconds_until)\n\n        await self.bot.wait_until_guild_available()\n        if not self.webhook:\n            await self.bot.fetch_webhook(Webhooks.reddit)\n\n        if datetime.utcnow().weekday() == 0:\n            await self.top_weekly_posts()\n            # if it's a monday send the top weekly posts\n\n        for subreddit in RedditConfig.subreddits:\n            top_posts = await self.get_top_posts(subreddit=subreddit, time=\"day\")\n            username = sub_clyde(f\"{subreddit} Top Daily Posts\")\n            message = await self.webhook.send(username=username, embed=top_posts, wait=True)\n\n            if message.channel.is_news():\n                await message.publish()\n\n    async def top_weekly_posts(self) -> None:\n        \"\"\"Post a summary of the top posts.\"\"\"\n        for subreddit in RedditConfig.subreddits:\n            # Send and pin the new weekly posts.\n            top_posts = await self.get_top_posts(subreddit=subreddit, time=\"week\")\n            username = sub_clyde(f\"{subreddit} Top Weekly Posts\")\n            message = await self.webhook.send(wait=True, username=username, embed=top_posts)\n\n            if subreddit.lower() == \"r/python\":\n                if not self.channel:\n                    log.warning(\"Failed to get #reddit channel to remove pins in the weekly loop.\")\n                    return\n\n                # Remove the oldest pins so that only 12 remain at most.\n                pins = await self.channel.pins()\n\n                while len(pins) >= 12:\n                    await pins[-1].unpin()\n                    del pins[-1]\n\n                await message.pin()\n\n                if message.channel.is_news():\n                    await message.publish()\n\n    @group(name=\"reddit\", invoke_without_command=True)\n    async def reddit_group(self, ctx: Context) -> None:\n        \"\"\"View the top posts from various subreddits.\"\"\"\n        await ctx.send_help(ctx.command)\n\n    @reddit_group.command(name=\"top\")\n    async def top_command(self, ctx: Context, subreddit: Subreddit = \"r/Python\") -> None:\n        \"\"\"Send the top posts of all time from a given subreddit.\"\"\"\n        async with ctx.typing():\n            embed = await self.get_top_posts(subreddit=subreddit, time=\"all\")\n\n        await ctx.send(content=f\"Here are the top {subreddit} posts of all time!\", embed=embed)\n\n    @reddit_group.command(name=\"daily\")\n    async def daily_command(self, ctx: Context, subreddit: Subreddit = \"r/Python\") -> None:\n        \"\"\"Send the top posts of today from a given subreddit.\"\"\"\n        async with ctx.typing():\n            embed = await self.get_top_posts(subreddit=subreddit, time=\"day\")\n\n        await ctx.send(content=f\"Here are today's top {subreddit} posts!\", embed=embed)\n\n    @reddit_group.command(name=\"weekly\")\n    async def weekly_command(self, ctx: Context, subreddit: Subreddit = \"r/Python\") -> None:\n        \"\"\"Send the top posts of this week from a given subreddit.\"\"\"\n        async with ctx.typing():\n            embed = await self.get_top_posts(subreddit=subreddit, time=\"week\")\n\n        await ctx.send(content=f\"Here are this week's top {subreddit} posts!\", embed=embed)\n\n    @with_role(*STAFF_ROLES)\n    @reddit_group.command(name=\"subreddits\", aliases=(\"subs\",))\n    async def subreddits_command(self, ctx: Context) -> None:\n        \"\"\"Send a paginated embed of all the subreddits we're relaying.\"\"\"\n        embed = Embed()\n        embed.title = \"Relayed subreddits.\"\n        embed.colour = Colour.blurple()\n\n        await LinePaginator.paginate(\n            RedditConfig.subreddits,\n            ctx, embed,\n            footer_text=\"Use the reddit commands along with these to view their posts.\",\n            empty=False,\n            max_lines=15\n        )\n\n\ndef setup(bot: Bot) -> None:\n    \"\"\"Load the Reddit cog.\"\"\"\n    if not RedditConfig.secret or not RedditConfig.client_id:\n        log.error(\"Credentials not provided, cog not loaded.\")\n        return\n    bot.add_cog(Reddit(bot))\n", "path": "bot/cogs/reddit.py"}], "after_files": [{"content": "import asyncio\nimport logging\nimport random\nimport textwrap\nfrom collections import namedtuple\nfrom datetime import datetime, timedelta\nfrom typing import List\n\nfrom aiohttp import BasicAuth, ClientError\nfrom discord import Colour, Embed, TextChannel\nfrom discord.ext.commands import Cog, Context, group\nfrom discord.ext.tasks import loop\nfrom discord.utils import escape_markdown\n\nfrom bot.bot import Bot\nfrom bot.constants import Channels, ERROR_REPLIES, Emojis, Reddit as RedditConfig, STAFF_ROLES, Webhooks\nfrom bot.converters import Subreddit\nfrom bot.decorators import with_role\nfrom bot.pagination import LinePaginator\nfrom bot.utils.messages import sub_clyde\n\nlog = logging.getLogger(__name__)\n\nAccessToken = namedtuple(\"AccessToken\", [\"token\", \"expires_at\"])\n\n\nclass Reddit(Cog):\n    \"\"\"Track subreddit posts and show detailed statistics about them.\"\"\"\n\n    HEADERS = {\"User-Agent\": \"python3:python-discord/bot:1.0.0 (by /u/PythonDiscord)\"}\n    URL = \"https://www.reddit.com\"\n    OAUTH_URL = \"https://oauth.reddit.com\"\n    MAX_RETRIES = 3\n\n    def __init__(self, bot: Bot):\n        self.bot = bot\n\n        self.webhook = None\n        self.access_token = None\n        self.client_auth = BasicAuth(RedditConfig.client_id, RedditConfig.secret)\n\n        bot.loop.create_task(self.init_reddit_ready())\n        self.auto_poster_loop.start()\n\n    def cog_unload(self) -> None:\n        \"\"\"Stop the loop task and revoke the access token when the cog is unloaded.\"\"\"\n        self.auto_poster_loop.cancel()\n        if self.access_token and self.access_token.expires_at > datetime.utcnow():\n            asyncio.create_task(self.revoke_access_token())\n\n    async def init_reddit_ready(self) -> None:\n        \"\"\"Sets the reddit webhook when the cog is loaded.\"\"\"\n        await self.bot.wait_until_guild_available()\n        if not self.webhook:\n            self.webhook = await self.bot.fetch_webhook(Webhooks.reddit)\n\n    @property\n    def channel(self) -> TextChannel:\n        \"\"\"Get the #reddit channel object from the bot's cache.\"\"\"\n        return self.bot.get_channel(Channels.reddit)\n\n    async def get_access_token(self) -> None:\n        \"\"\"\n        Get a Reddit API OAuth2 access token and assign it to self.access_token.\n\n        A token is valid for 1 hour. There will be MAX_RETRIES to get a token, after which the cog\n        will be unloaded and a ClientError raised if retrieval was still unsuccessful.\n        \"\"\"\n        for i in range(1, self.MAX_RETRIES + 1):\n            response = await self.bot.http_session.post(\n                url=f\"{self.URL}/api/v1/access_token\",\n                headers=self.HEADERS,\n                auth=self.client_auth,\n                data={\n                    \"grant_type\": \"client_credentials\",\n                    \"duration\": \"temporary\"\n                }\n            )\n\n            if response.status == 200 and response.content_type == \"application/json\":\n                content = await response.json()\n                expiration = int(content[\"expires_in\"]) - 60  # Subtract 1 minute for leeway.\n                self.access_token = AccessToken(\n                    token=content[\"access_token\"],\n                    expires_at=datetime.utcnow() + timedelta(seconds=expiration)\n                )\n\n                log.debug(f\"New token acquired; expires on UTC {self.access_token.expires_at}\")\n                return\n            else:\n                log.debug(\n                    f\"Failed to get an access token: \"\n                    f\"status {response.status} & content type {response.content_type}; \"\n                    f\"retrying ({i}/{self.MAX_RETRIES})\"\n                )\n\n            await asyncio.sleep(3)\n\n        self.bot.remove_cog(self.qualified_name)\n        raise ClientError(\"Authentication with the Reddit API failed. Unloading the cog.\")\n\n    async def revoke_access_token(self) -> None:\n        \"\"\"\n        Revoke the OAuth2 access token for the Reddit API.\n\n        For security reasons, it's good practice to revoke the token when it's no longer being used.\n        \"\"\"\n        response = await self.bot.http_session.post(\n            url=f\"{self.URL}/api/v1/revoke_token\",\n            headers=self.HEADERS,\n            auth=self.client_auth,\n            data={\n                \"token\": self.access_token.token,\n                \"token_type_hint\": \"access_token\"\n            }\n        )\n\n        if response.status == 204 and response.content_type == \"application/json\":\n            self.access_token = None\n        else:\n            log.warning(f\"Unable to revoke access token: status {response.status}.\")\n\n    async def fetch_posts(self, route: str, *, amount: int = 25, params: dict = None) -> List[dict]:\n        \"\"\"A helper method to fetch a certain amount of Reddit posts at a given route.\"\"\"\n        # Reddit's JSON responses only provide 25 posts at most.\n        if not 25 >= amount > 0:\n            raise ValueError(\"Invalid amount of subreddit posts requested.\")\n\n        # Renew the token if necessary.\n        if not self.access_token or self.access_token.expires_at < datetime.utcnow():\n            await self.get_access_token()\n\n        url = f\"{self.OAUTH_URL}/{route}\"\n        for _ in range(self.MAX_RETRIES):\n            response = await self.bot.http_session.get(\n                url=url,\n                headers={**self.HEADERS, \"Authorization\": f\"bearer {self.access_token.token}\"},\n                params=params\n            )\n            if response.status == 200 and response.content_type == 'application/json':\n                # Got appropriate response - process and return.\n                content = await response.json()\n                posts = content[\"data\"][\"children\"]\n                return posts[:amount]\n\n            await asyncio.sleep(3)\n\n        log.debug(f\"Invalid response from: {url} - status code {response.status}, mimetype {response.content_type}\")\n        return list()  # Failed to get appropriate response within allowed number of retries.\n\n    async def get_top_posts(self, subreddit: Subreddit, time: str = \"all\", amount: int = 5) -> Embed:\n        \"\"\"\n        Get the top amount of posts for a given subreddit within a specified timeframe.\n\n        A time of \"all\" will get posts from all time, \"day\" will get top daily posts and \"week\" will get the top\n        weekly posts.\n\n        The amount should be between 0 and 25 as Reddit's JSON requests only provide 25 posts at most.\n        \"\"\"\n        embed = Embed(description=\"\")\n\n        posts = await self.fetch_posts(\n            route=f\"{subreddit}/top\",\n            amount=amount,\n            params={\"t\": time}\n        )\n\n        if not posts:\n            embed.title = random.choice(ERROR_REPLIES)\n            embed.colour = Colour.red()\n            embed.description = (\n                \"Sorry! We couldn't find any posts from that subreddit. \"\n                \"If this problem persists, please let us know.\"\n            )\n\n            return embed\n\n        for post in posts:\n            data = post[\"data\"]\n\n            text = data[\"selftext\"]\n            if text:\n                text = textwrap.shorten(text, width=128, placeholder=\"...\")\n                text += \"\\n\"  # Add newline to separate embed info\n\n            ups = data[\"ups\"]\n            comments = data[\"num_comments\"]\n            author = data[\"author\"]\n\n            title = textwrap.shorten(data[\"title\"], width=64, placeholder=\"...\")\n            # Normal brackets interfere with Markdown.\n            title = escape_markdown(title).replace(\"[\", \"\u298b\").replace(\"]\", \"\u298c\")\n            link = self.URL + data[\"permalink\"]\n\n            embed.description += (\n                f\"**[{title}]({link})**\\n\"\n                f\"{text}\"\n                f\"{Emojis.upvotes} {ups} {Emojis.comments} {comments} {Emojis.user} {author}\\n\\n\"\n            )\n\n        embed.colour = Colour.blurple()\n        return embed\n\n    @loop()\n    async def auto_poster_loop(self) -> None:\n        \"\"\"Post the top 5 posts daily, and the top 5 posts weekly.\"\"\"\n        # once we upgrade to d.py 1.3 this can be removed and the loop can use the `time=datetime.time.min` parameter\n        now = datetime.utcnow()\n        tomorrow = now + timedelta(days=1)\n        midnight_tomorrow = tomorrow.replace(hour=0, minute=0, second=0)\n        seconds_until = (midnight_tomorrow - now).total_seconds()\n\n        await asyncio.sleep(seconds_until)\n\n        await self.bot.wait_until_guild_available()\n        if not self.webhook:\n            await self.bot.fetch_webhook(Webhooks.reddit)\n\n        if datetime.utcnow().weekday() == 0:\n            await self.top_weekly_posts()\n            # if it's a monday send the top weekly posts\n\n        for subreddit in RedditConfig.subreddits:\n            top_posts = await self.get_top_posts(subreddit=subreddit, time=\"day\")\n            username = sub_clyde(f\"{subreddit} Top Daily Posts\")\n            message = await self.webhook.send(username=username, embed=top_posts, wait=True)\n\n            if message.channel.is_news():\n                await message.publish()\n\n    async def top_weekly_posts(self) -> None:\n        \"\"\"Post a summary of the top posts.\"\"\"\n        for subreddit in RedditConfig.subreddits:\n            # Send and pin the new weekly posts.\n            top_posts = await self.get_top_posts(subreddit=subreddit, time=\"week\")\n            username = sub_clyde(f\"{subreddit} Top Weekly Posts\")\n            message = await self.webhook.send(wait=True, username=username, embed=top_posts)\n\n            if subreddit.lower() == \"r/python\":\n                if not self.channel:\n                    log.warning(\"Failed to get #reddit channel to remove pins in the weekly loop.\")\n                    return\n\n                # Remove the oldest pins so that only 12 remain at most.\n                pins = await self.channel.pins()\n\n                while len(pins) >= 12:\n                    await pins[-1].unpin()\n                    del pins[-1]\n\n                await message.pin()\n\n                if message.channel.is_news():\n                    await message.publish()\n\n    @group(name=\"reddit\", invoke_without_command=True)\n    async def reddit_group(self, ctx: Context) -> None:\n        \"\"\"View the top posts from various subreddits.\"\"\"\n        await ctx.send_help(ctx.command)\n\n    @reddit_group.command(name=\"top\")\n    async def top_command(self, ctx: Context, subreddit: Subreddit = \"r/Python\") -> None:\n        \"\"\"Send the top posts of all time from a given subreddit.\"\"\"\n        async with ctx.typing():\n            embed = await self.get_top_posts(subreddit=subreddit, time=\"all\")\n\n        await ctx.send(content=f\"Here are the top {subreddit} posts of all time!\", embed=embed)\n\n    @reddit_group.command(name=\"daily\")\n    async def daily_command(self, ctx: Context, subreddit: Subreddit = \"r/Python\") -> None:\n        \"\"\"Send the top posts of today from a given subreddit.\"\"\"\n        async with ctx.typing():\n            embed = await self.get_top_posts(subreddit=subreddit, time=\"day\")\n\n        await ctx.send(content=f\"Here are today's top {subreddit} posts!\", embed=embed)\n\n    @reddit_group.command(name=\"weekly\")\n    async def weekly_command(self, ctx: Context, subreddit: Subreddit = \"r/Python\") -> None:\n        \"\"\"Send the top posts of this week from a given subreddit.\"\"\"\n        async with ctx.typing():\n            embed = await self.get_top_posts(subreddit=subreddit, time=\"week\")\n\n        await ctx.send(content=f\"Here are this week's top {subreddit} posts!\", embed=embed)\n\n    @with_role(*STAFF_ROLES)\n    @reddit_group.command(name=\"subreddits\", aliases=(\"subs\",))\n    async def subreddits_command(self, ctx: Context) -> None:\n        \"\"\"Send a paginated embed of all the subreddits we're relaying.\"\"\"\n        embed = Embed()\n        embed.title = \"Relayed subreddits.\"\n        embed.colour = Colour.blurple()\n\n        await LinePaginator.paginate(\n            RedditConfig.subreddits,\n            ctx, embed,\n            footer_text=\"Use the reddit commands along with these to view their posts.\",\n            empty=False,\n            max_lines=15\n        )\n\n\ndef setup(bot: Bot) -> None:\n    \"\"\"Load the Reddit cog.\"\"\"\n    if not RedditConfig.secret or not RedditConfig.client_id:\n        log.error(\"Credentials not provided, cog not loaded.\")\n        return\n    bot.add_cog(Reddit(bot))\n", "path": "bot/cogs/reddit.py"}]}