blizzthewolf/SeaCogs
1
0
Fork 1
forked from cswimr/SeaCogs
Code Pull requests Activity

fix(updatechecker): pylint fixes

Browse source
This commit is contained in:
Seaswimmer 2024-01-07 19:20:21 +00:00
parent 77a4001fed
commit bc57fb8eca
Signed by untrusted user: cswimr
GPG key ID: D74DDDDF420E13DF
1 changed files with 109 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

225
updatechecker/updatechecker.py
View file

@ -62,14 +62,14 @@ class UpdateChecker(commands.Cog):
self.conf.register_global(**default_global) self.conf.register_global(**default_global)
self.task = self.bot.loop.create_task(self.bg_task()) self.task = self.bot.loop.create_task(self.bg_task())
def cog_unload(self): async def cog_unload(self):
self.__unload() self.__unload()
def __unload(self): def __unload(self):
self.task.cancel() self.task.cancel()
self.session.detach() self.session.detach()
async def red_delete_data_for_user(self, **kwargs): async def red_delete_data_for_user(self, **kwargs): # pylint: disable=unused-argument
"""This cog does not store user data""" """This cog does not store user data"""
return return
@ -102,7 +102,7 @@ class UpdateChecker(commands.Cog):
else: else:
send = self.bot.send_to_owners send = self.bot.send_to_owners
all_repos = cog._repo_manager.get_all_repo_names() all_repos = cog._repo_manager.get_all_repo_names() # pylint: disable=protected-access
for repo in all_repos: for repo in all_repos:
if not (repo in list(repos.keys())): if not (repo in list(repos.keys())):
repos[repo] = "--default--" repos[repo] = "--default--"
@ -110,7 +110,7 @@ class UpdateChecker(commands.Cog):
saving_dict = {k: v for k, v in repos.items() if k in all_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(): for repo_name, commit_saved in saving_dict.items():
repo = cog._repo_manager.get_repo(repo_name) repo = cog._repo_manager.get_repo(repo_name) # pylint: disable=protected-access
if not repo: if not repo:
continue continue
parsed_url = urlparse(repo.url) parsed_url = urlparse(repo.url)
@ -119,7 +119,7 @@ class UpdateChecker(commands.Cog):
response = await self.fetch_feed(url) response = await self.fetch_feed(url)
try: try:
commit = response.entries[0]["id"][33:] commit = response.entries[0]["id"][33:]
hash = "[" + commit + "](" + response.entries[0]["link"] + ")" chash = "[" + commit + "](" + response.entries[0]["link"] + ")"
cn = response.entries[0]["title"] + " - " + response.entries[0]["author"] cn = response.entries[0]["title"] + " - " + response.entries[0]["author"]
image = response.entries[0]["media_thumbnail"][0]["url"].split("?")[0] image = response.entries[0]["media_thumbnail"][0]["url"].split("?")[0]
except AttributeError: except AttributeError:
@ -129,7 +129,7 @@ class UpdateChecker(commands.Cog):
response = await self.fetch_feed(url) response = await self.fetch_feed(url)
try: try:
commit = response.entries[0]["id"][33:] commit = response.entries[0]["id"][33:]
hash = "[" + commit + "](" + response.entries[0]["link"] + ")" chash = "[" + commit + "](" + response.entries[0]["link"] + ")"
cn = response.entries[0]["title"] + " - " + response.entries[0]["author"] 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) image = await self.fetch_gitea_thumbnail(parsed_url.scheme + "://" + parsed_url.netloc + "/api/v1/repos" + parsed_url.path)
except AttributeError: except AttributeError:
@ -142,107 +142,102 @@ class UpdateChecker(commands.Cog):
continue continue
# CN is used here for backwards compatability, don't want people to get an # 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 # update for each and every one of their cogs when updating this cog
if ( if commit_saved not in (commit, cn, '--default--'):
commit != commit_saved if use_embed:
and cn != commit_saved e = discord.Embed(
and commit_saved != "--default--" title="Update Checker",
): description=f"Update available for repo: {repo.name}",
if True: # KACHOW timestamp=datetime.utcnow(),
if use_embed: color=0x00FF00,
e = discord.Embed( )
title="Update Checker", e.add_field(name="URL", value=repo.url)
description=f"Update available for repo: {repo.name}", e.add_field(name="Branch", value=repo.branch)
timestamp=datetime.utcnow(), e.add_field(name="Commit", value=cn)
color=0x00FF00, e.add_field(name="Hash", value=chash)
) if image is not None:
e.add_field(name="URL", value=repo.url) e.set_thumbnail(url=image)
e.add_field(name="Branch", value=repo.branch)
e.add_field(name="Commit", value=cn)
e.add_field(name="Hash", value=hash)
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(
"[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."
)
if use_embed:
await self.bot.send_to_owners(embed=e)
else:
await self.bot.send_to_owners(e)
await self.conf.gochannel.set(0)
else: else:
try: e = box(
await channel.send( "[Update Checker]"
f"[Update Checker] Update found for repo: {repo.name}. Updating repos..." f" Repo: {repo.name}\n"
) f" URL: {repo.url}\n"
except AttributeError: f" Commit: {cn}\n"
owner = (await self.bot.application_info()).owner f" Hash: {commit}\n"
await owner.send( f" Time: {datetime.utcnow()}",
"[Update Checker] It appears that the channel for this cog has been deleted. From now on, it will DM you." 'css'
) )
channel = owner try:
await self.conf.gochannel.set(0) if use_embed:
except discord.errors.Forbidden: await send(embed=e)
owner = (await self.bot.application_info()).owner
await owner.send(
"[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)
# Just a copy of `[p]cog update`, but without using ctx things
try:
installed_cogs = set(await cog.installed_cogs())
updated = await cog._repo_manager.update_all_repos()
updated_cogs = set(
cog for repo in updated for cog in repo.available_cogs
)
installed_and_updated = updated_cogs & installed_cogs
if installed_and_updated:
await cog._reinstall_requirements(installed_and_updated)
await cog._reinstall_cogs(installed_and_updated)
await cog._reinstall_libraries(installed_and_updated)
cognames = {c.name for c in installed_and_updated}
message = humanize_list(tuple(map(inline, cognames)))
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: else:
try: await send(e)
await channel.send( except discord.Forbidden:
f"[Update Checker]: Ran cog update. Updated cogs: {message}" # send_to_owners suppresses Forbidden, logging it to console.
) # As a result, this will only happen if a channel was set.
except discord.errors.Forbidden: await self.bot.send_to_owners(
pass "[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."
)
if use_embed:
await self.bot.send_to_owners(embed=e)
else:
await self.bot.send_to_owners(e)
await self.conf.gochannel.set(0)
# Was already inaccessible before I got here, so I'm just gonna leave it and look at it later -- Sea
# try:
# await channel.send(
# f"[Update Checker] Update found for repo: {repo.name}. Updating repos..."
# )
# except AttributeError:
# owner = (await self.bot.application_info()).owner
# await owner.send(
# "[Update Checker] It appears that the channel for this cog has been deleted. From now on, it will DM you."
# )
# channel = owner
# await self.conf.gochannel.set(0)
# except discord.errors.Forbidden:
# owner = (await self.bot.application_info()).owner
# await owner.send(
# "[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)
# # Just a copy of `[p]cog update`, but without using ctx things
# try:
# installed_cogs = set(await cog.installed_cogs())
# updated = await cog._repo_manager.update_all_repos()
# updated_cogs = set(
# cog for repo in updated for cog in repo.available_cogs
# )
# installed_and_updated = updated_cogs & installed_cogs
# if installed_and_updated:
# await cog._reinstall_requirements(installed_and_updated)
# await cog._reinstall_cogs(installed_and_updated)
# await cog._reinstall_libraries(installed_and_updated)
# cognames = {c.name for c in installed_and_updated}
# message = humanize_list(tuple(map(inline, cognames)))
# 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 asyncio.sleep(1)
await self.conf.repos.set(saving_dict) await self.conf.repos.set(saving_dict)
@ -274,21 +269,19 @@ class UpdateChecker(commands.Cog):
@commands.group(name="cogupdater", aliases=["cu"]) @commands.group(name="cogupdater", aliases=["cu"])
async def update(self, ctx): async def update(self, ctx):
"""Group command for controlling the update checker cog.""" """Group command for controlling the update checker cog."""
pass
@commands.is_owner() @commands.is_owner()
@update.command() @update.command()
async def auto(self, ctx): async def auto(self, ctx):
"""Changes automatic cog updates to the opposite setting.""" """Changes automatic cog updates to the opposite setting."""
if False: # KACHOW # Was already inaccessible before I got here, so I'm just gonna leave it and look at it later -- Sea
auto = await self.conf.auto() # auto = await self.conf.auto()
await self.conf.auto.set(not auto) # await self.conf.auto.set(not auto)
status = "disabled" if auto else "enabled" # status = "disabled" if auto else "enabled"
await ctx.send(f"Auto cog updates are now {status}") # await ctx.send(f"Auto cog updates are now {status}")
else: await ctx.send(
await ctx.send( "This command is disabled for the time being. Cog updates will not run automatically, however notifications will still send."
"This command is disabled for the time being. Cog updates will not run automatically, however notifications will still send." )
)
@commands.is_owner() @commands.is_owner()
@update.command() @update.command()
@ -457,7 +450,7 @@ class UpdateChecker(commands.Cog):
except asyncio.exceptions.InvalidStateError: except asyncio.exceptions.InvalidStateError:
message += " No error has been encountered." message += " No error has been encountered."
else: else:
message += " An error has been encountered. Please run `[p]cogupdater task error` and report it to Neuro Assassin on the help server." message += " An error has been encountered. Please run `[p]cogupdater task error` and report it to SeaswimmerTheFsh (.seasw) on the help server."
await ctx.send(message) await ctx.send(message)
@_group_update_task.command() @_group_update_task.command()