Rosie-Val/SeaCogs
1
0
Fork 0
Code Pull requests Activity

fix(updatechecker): more pylint fixes

Browse source
This commit is contained in:
Seaswimmer 2024-01-07 19:27:19 +00:00
parent bc57fb8eca
commit 45fcc00cdb
Signed by untrusted user: cswimr
GPG key ID: D74DDDDF420E13DF
1 changed files with 162 additions and 169 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

331
updatechecker/updatechecker.py
View file

@ -75,171 +75,169 @@ class UpdateChecker(commands.Cog):
@tasks.loop(minutes=1) @tasks.loop(minutes=1)
async def bg_task(self): async def bg_task(self):
while True: cog = self.bot.get_cog("Downloader")
cog = self.bot.get_cog("Downloader") if cog is not None:
if cog is not None: data = await self.conf.all()
data = await self.conf.all() repos = data["repos"]
repos = data["repos"] channel = data["gochannel"]
auto = data["auto"] use_embed = data["embed"]
channel = data["gochannel"] whitelist = data["whitelist"]
use_embed = data["embed"] blacklist = data["blacklist"]
whitelist = data["whitelist"] if channel:
blacklist = data["blacklist"] channel = self.bot.get_channel(channel)
if channel: if channel is None:
channel = self.bot.get_channel(channel) await self.bot.send_to_owners(
if channel is None: "[Update Checker] It appears that I am no longer allowed to send messages to the designated update channel. "
"From now on, it will DM you."
)
await self.conf.gochannel.set(0)
send = self.bot.send_to_owners
else:
use_embed = (
use_embed and channel.permissions_for(channel.guild.me).embed_links
)
send = channel.send
else:
send = self.bot.send_to_owners
all_repos = cog._repo_manager.get_all_repo_names() # pylint: disable=protected-access
for repo in all_repos:
if not (repo in list(repos.keys())):
repos[repo] = "--default--"
await self.conf.repos.set(repos)
saving_dict = {k: v for k, v in repos.items() if k in all_repos}
for repo_name, commit_saved in saving_dict.items():
repo = cog._repo_manager.get_repo(repo_name) # pylint: disable=protected-access
if not repo:
continue
parsed_url = urlparse(repo.url)
if parsed_url.netloc == "github.com":
url = repo.url + r"/commits/" + repo.branch + ".atom"
response = await self.fetch_feed(url)
try:
commit = response.entries[0]["id"][33:]
chash = "[" + commit + "](" + response.entries[0]["link"] + ")"
cn = response.entries[0]["title"] + " - " + response.entries[0]["author"]
image = response.entries[0]["media_thumbnail"][0]["url"].split("?")[0]
except AttributeError:
continue
else:
url = repo.url + r"/rss/branch/" + repo.branch
response = await self.fetch_feed(url)
try:
commit = response.entries[0]["id"][33:]
chash = "[" + commit + "](" + response.entries[0]["link"] + ")"
cn = response.entries[0]["title"] + " - " + response.entries[0]["author"]
image = await self.fetch_gitea_thumbnail(parsed_url.scheme + "://" + parsed_url.netloc + "/api/v1/repos" + parsed_url.path)
except AttributeError:
continue
saving_dict[repo_name] = commit
if whitelist:
if repo_name not in whitelist:
continue
if repo_name in blacklist:
continue
# CN is used here for backwards compatability, don't want people to get an
# update for each and every one of their cogs when updating this cog
if commit_saved not in (commit, cn, '--default--'):
if use_embed:
e = discord.Embed(
title="Update Checker",
description=f"Update available for repo: {repo.name}",
timestamp=datetime.utcnow(),
color=0x00FF00,
)
e.add_field(name="URL", value=repo.url)
e.add_field(name="Branch", value=repo.branch)
e.add_field(name="Commit", value=cn)
e.add_field(name="Hash", value=chash)
if image is not None:
e.set_thumbnail(url=image)
else:
e = box(
"[Update Checker]"
f" Repo: {repo.name}\n"
f" URL: {repo.url}\n"
f" Commit: {cn}\n"
f" Hash: {commit}\n"
f" Time: {datetime.utcnow()}",
'css'
)
try:
if use_embed:
await send(embed=e)
else:
await send(e)
except discord.Forbidden:
# send_to_owners suppresses Forbidden, logging it to console.
# As a result, this will only happen if a channel was set.
await self.bot.send_to_owners( await self.bot.send_to_owners(
"[Update Checker] It appears that I am no longer allowed to send messages to the designated update channel. " "[Update Checker] It appears that I am no longer allowed to send messages to the designated update channel. "
"From now on, it will DM you." "From now on, it will DM you."
) )
await self.conf.gochannel.set(0)
send = self.bot.send_to_owners
else:
use_embed = (
use_embed and channel.permissions_for(channel.guild.me).embed_links
)
send = channel.send
else:
send = self.bot.send_to_owners
all_repos = cog._repo_manager.get_all_repo_names() # pylint: disable=protected-access
for repo in all_repos:
if not (repo in list(repos.keys())):
repos[repo] = "--default--"
await self.conf.repos.set(repos)
saving_dict = {k: v for k, v in repos.items() if k in all_repos}
for repo_name, commit_saved in saving_dict.items():
repo = cog._repo_manager.get_repo(repo_name) # pylint: disable=protected-access
if not repo:
continue
parsed_url = urlparse(repo.url)
if parsed_url.netloc == "github.com":
url = repo.url + r"/commits/" + repo.branch + ".atom"
response = await self.fetch_feed(url)
try:
commit = response.entries[0]["id"][33:]
chash = "[" + commit + "](" + response.entries[0]["link"] + ")"
cn = response.entries[0]["title"] + " - " + response.entries[0]["author"]
image = response.entries[0]["media_thumbnail"][0]["url"].split("?")[0]
except AttributeError:
continue
else:
url = repo.url + r"/rss/branch/" + repo.branch
response = await self.fetch_feed(url)
try:
commit = response.entries[0]["id"][33:]
chash = "[" + commit + "](" + response.entries[0]["link"] + ")"
cn = response.entries[0]["title"] + " - " + response.entries[0]["author"]
image = await self.fetch_gitea_thumbnail(parsed_url.scheme + "://" + parsed_url.netloc + "/api/v1/repos" + parsed_url.path)
except AttributeError:
continue
saving_dict[repo_name] = commit
if whitelist:
if repo_name not in whitelist:
continue
if repo_name in blacklist:
continue
# CN is used here for backwards compatability, don't want people to get an
# update for each and every one of their cogs when updating this cog
if commit_saved not in (commit, cn, '--default--'):
if use_embed: if use_embed:
e = discord.Embed( await self.bot.send_to_owners(embed=e)
title="Update Checker",
description=f"Update available for repo: {repo.name}",
timestamp=datetime.utcnow(),
color=0x00FF00,
)
e.add_field(name="URL", value=repo.url)
e.add_field(name="Branch", value=repo.branch)
e.add_field(name="Commit", value=cn)
e.add_field(name="Hash", value=chash)
if image is not None:
e.set_thumbnail(url=image)
else: else:
e = box( await self.bot.send_to_owners(e)
"[Update Checker]" await self.conf.gochannel.set(0)
f" Repo: {repo.name}\n" # Was already inaccessible before I got here, so I'm just gonna leave it and look at it later -- Sea
f" URL: {repo.url}\n" # try:
f" Commit: {cn}\n" # await channel.send(
f" Hash: {commit}\n" # f"[Update Checker] Update found for repo: {repo.name}. Updating repos..."
f" Time: {datetime.utcnow()}", # )
'css' # except AttributeError:
) # owner = (await self.bot.application_info()).owner
try: # await owner.send(
if use_embed: # "[Update Checker] It appears that the channel for this cog has been deleted. From now on, it will DM you."
await send(embed=e) # )
else: # channel = owner
await send(e) # await self.conf.gochannel.set(0)
except discord.Forbidden: # except discord.errors.Forbidden:
# send_to_owners suppresses Forbidden, logging it to console. # owner = (await self.bot.application_info()).owner
# As a result, this will only happen if a channel was set. # await owner.send(
await self.bot.send_to_owners( # "[Update Checker] It appears that I am no longer allowed to send messages to the designated update channel. From now on, it will DM you."
"[Update Checker] It appears that I am no longer allowed to send messages to the designated update channel. " # )
"From now on, it will DM you." # channel = owner
) # await self.conf.gochannel.set(0)
if use_embed: # # Just a copy of `[p]cog update`, but without using ctx things
await self.bot.send_to_owners(embed=e) # try:
else: # installed_cogs = set(await cog.installed_cogs())
await self.bot.send_to_owners(e) # updated = await cog._repo_manager.update_all_repos()
await self.conf.gochannel.set(0) # updated_cogs = set(
# Was already inaccessible before I got here, so I'm just gonna leave it and look at it later -- Sea # cog for repo in updated for cog in repo.available_cogs
# try: # )
# await channel.send( # installed_and_updated = updated_cogs & installed_cogs
# f"[Update Checker] Update found for repo: {repo.name}. Updating repos..." # if installed_and_updated:
# ) # await cog._reinstall_requirements(installed_and_updated)
# except AttributeError: # await cog._reinstall_cogs(installed_and_updated)
# owner = (await self.bot.application_info()).owner # await cog._reinstall_libraries(installed_and_updated)
# await owner.send( # cognames = {c.name for c in installed_and_updated}
# "[Update Checker] It appears that the channel for this cog has been deleted. From now on, it will DM you." # message = humanize_list(tuple(map(inline, cognames)))
# ) # except Exception as error:
# channel = owner # exception_log = (
# await self.conf.gochannel.set(0) # "Exception while updating repos in Update Checker \n"
# except discord.errors.Forbidden: # )
# owner = (await self.bot.application_info()).owner # exception_log += "".join(
# await owner.send( # traceback.format_exception(
# "[Update Checker] It appears that I am no longer allowed to send messages to the designated update channel. From now on, it will DM you." # type(error), error, error.__traceback__
# ) # )
# channel = owner # )
# await self.conf.gochannel.set(0) # try:
# # Just a copy of `[p]cog update`, but without using ctx things # await channel.send(
# try: # f"[Update Checker]: Error while updating repos.\n\n{exception_log}"
# installed_cogs = set(await cog.installed_cogs()) # )
# updated = await cog._repo_manager.update_all_repos() # except discord.errors.Forbidden:
# updated_cogs = set( # pass
# cog for repo in updated for cog in repo.available_cogs # else:
# ) # try:
# installed_and_updated = updated_cogs & installed_cogs # await channel.send(
# if installed_and_updated: # f"[Update Checker]: Ran cog update. Updated cogs: {message}"
# await cog._reinstall_requirements(installed_and_updated) # )
# await cog._reinstall_cogs(installed_and_updated) # except discord.errors.Forbidden:
# await cog._reinstall_libraries(installed_and_updated) # pass
# cognames = {c.name for c in installed_and_updated} await asyncio.sleep(1)
# message = humanize_list(tuple(map(inline, cognames))) await self.conf.repos.set(saving_dict)
# except Exception as error:
# exception_log = (
# "Exception while updating repos in Update Checker \n"
# )
# exception_log += "".join(
# traceback.format_exception(
# type(error), error, error.__traceback__
# )
# )
# try:
# await channel.send(
# f"[Update Checker]: Error while updating repos.\n\n{exception_log}"
# )
# except discord.errors.Forbidden:
# pass
# else:
# try:
# await channel.send(
# f"[Update Checker]: Ran cog update. Updated cogs: {message}"
# )
# except discord.errors.Forbidden:
# pass
await asyncio.sleep(1)
await self.conf.repos.set(saving_dict)
async def fetch_feed(self, url: str): async def fetch_feed(self, url: str):
# Thank's to Sinbad's rss cog after which I copied this # Thank's to Sinbad's rss cog after which I copied this
@ -315,9 +313,8 @@ class UpdateChecker(commands.Cog):
if channel == 0: if channel == 0:
channel = "Direct Messages" channel = "Direct Messages"
else: else:
try: channel = self.bot.get_channel(channel).name
channel = self.bot.get_channel(channel).name if channel is None:
except:
channel = "Unknown" channel = "Unknown"
e.add_field(name="Update Channel", value=channel) e.add_field(name="Update Channel", value=channel)
await ctx.send(embed=e) await ctx.send(embed=e)
@ -325,9 +322,8 @@ class UpdateChecker(commands.Cog):
if channel == 0: if channel == 0:
channel = "Direct Messages" channel = "Direct Messages"
else: else:
try: channel = self.bot.get_channel(channel).name
channel = self.bot.get_channel(channel).name if channel is None:
except:
channel = "Unknown" channel = "Unknown"
message = ( message = (
"```css\n" "```css\n"
@ -363,7 +359,6 @@ class UpdateChecker(commands.Cog):
@whiteblacklist.group() @whiteblacklist.group()
async def whitelist(self, ctx): async def whitelist(self, ctx):
"""Whitelist certain repos from which to receive updates.""" """Whitelist certain repos from which to receive updates."""
pass
@whitelist.command(name="add") @whitelist.command(name="add")
async def whitelistadd(self, ctx, *repos: Repo): async def whitelistadd(self, ctx, *repos: Repo):
@ -396,7 +391,6 @@ class UpdateChecker(commands.Cog):
@whiteblacklist.group() @whiteblacklist.group()
async def blacklist(self, ctx): async def blacklist(self, ctx):
"""Blacklist certain repos from which to receive updates.""" """Blacklist certain repos from which to receive updates."""
pass
@blacklist.command(name="add") @blacklist.command(name="add")
async def blacklistadd(self, ctx, *repos: Repo): async def blacklistadd(self, ctx, *repos: Repo):
@ -430,7 +424,6 @@ class UpdateChecker(commands.Cog):
@update.group(name="task") @update.group(name="task")
async def _group_update_task(self, ctx): async def _group_update_task(self, ctx):
"""View the status of the task (the one checking for updates).""" """View the status of the task (the one checking for updates)."""
pass
@_group_update_task.command() @_group_update_task.command()
async def status(self, ctx): async def status(self, ctx):
@ -462,5 +455,5 @@ class UpdateChecker(commands.Cog):
message = "No error has been encountered." message = "No error has been encountered."
else: else:
ex = traceback.format_exception(type(e), e, e.__traceback__) ex = traceback.format_exception(type(e), e, e.__traceback__)
message = "An error has been encountered: ```py\n" + "".join(ex) + "```" message = "An error has been encountered:" + box("".join(ex), "py")
await ctx.send(message) await ctx.send(message)