Yamraaj (Sourcery refactored)

bot/helper/mirror_utils/download_utils/youtube_dl_download_helper.py

@@ -19,7 +19,7 @@ def debug(self, msg):
         # Hack to fix changing changing extension
         match = re.search(r'.Merger..Merging formats into..(.*?).$', msg)
         if match and not self.obj.is_playlist:
-            newname = match.group(1)
+            newname = match[1]
             newname = newname.split("/")
             newname = newname[-1]
             self.obj.name = newname
@@ -159,10 +159,12 @@ def add_download(self, link, path, qual, name):
           self.opts['postprocessors'] = [{'key': 'FFmpegExtractAudio','preferredcodec': 'mp3','preferredquality': '320',}]
         else:
           self.opts['format'] = qual
-        if not self.is_playlist:
-            self.opts['outtmpl'] = f"{path}/{self.name}"
-        else:
-            self.opts['outtmpl'] = f"{path}/{self.name}/%(title)s.%(ext)s"
+        self.opts['outtmpl'] = (
+            f"{path}/{self.name}/%(title)s.%(ext)s"
+            if self.is_playlist
+            else f"{path}/{self.name}"
+        )
+
         self.__download(link)
 
     def cancel_download(self):

bot/helper/mirror_utils/upload_utils/gdriveTools.py

@@ -104,7 +104,7 @@ def getIdFromUrl(link: str):
             res = re.search(regex,link)
             if res is None:
                 raise IndexError("G-Drive ID not found.")
-            return res.group(5)
+            return res[5]
         parsed = urlparse.urlparse(link)
         return parse_qs(parsed.query)['id'][0]
 
@@ -197,9 +197,7 @@ def upload_file(self, file_path, file_name, mime_type, parent_id):
         drive_file = self.__service.files().create(supportsTeamDrives=True,
                                                    body=file_metadata, media_body=media_body)
         response = None
-        while response is None:
-            if self.is_cancelled:
-                break
+        while response is None and not self.is_cancelled:
             try:
                 self.status, response = drive_file.next_chunk()
             except HttpError as err:
@@ -236,7 +234,7 @@ def upload(self, file_name: str):
         file_dir = f"{DOWNLOAD_DIR}{self.__listener.message.message_id}"
         file_path = f"{file_dir}/{file_name}"
         size = get_readable_file_size(get_path_size(file_path))
-        LOGGER.info("Uploading File: " + file_path)
+        LOGGER.info(f"Uploading File: {file_path}")
         self.updater = setInterval(self.update_interval, self._on_upload_progress)
         if os.path.isfile(file_path):
             try:
@@ -246,7 +244,7 @@ def upload(self, file_name: str):
                     return
                 if link is None:
                     raise Exception('Upload has been manually cancelled')
-                LOGGER.info("Uploaded To G-Drive: " + file_path)
+                LOGGER.info(f"Uploaded To G-Drive: {file_path}")
             except Exception as e:
                 if isinstance(e, RetryError):
                     LOGGER.info(f"Total Attempts: {e.last_attempt.attempt_number}")
@@ -272,7 +270,7 @@ def upload(self, file_name: str):
                     msg = self.deletefile(link)
                     LOGGER.info(f"{msg}")
                     return
-                LOGGER.info("Uploaded To G-Drive: " + file_name)
+                LOGGER.info(f"Uploaded To G-Drive: {file_name}")
             except Exception as e:
                 if isinstance(e, RetryError):
                     LOGGER.info(f"Total Attempts: {e.last_attempt.attempt_number}")
@@ -439,9 +437,9 @@ def clone(self, link):
                 err = err.last_attempt.exception()
             err = str(err).replace('>', '').replace('<', '')
             LOGGER.error(err)
-            if "User rate limit exceeded" in str(err):
+            if "User rate limit exceeded" in err:
                 msg = "User rate limit exceeded."
-            elif "File not found" in str(err):
+            elif "File not found" in err:
                 token_service = self.alt_authorize()
                 if token_service is not None:
                     self.__service = token_service
@@ -488,7 +486,10 @@ def create_directory(self, directory_name, parent_id):
         file_id = file.get("id")
         if not IS_TEAM_DRIVE:
             self.__set_permission(file_id)
-        LOGGER.info("♻️ Created G-Drive Folder:\nName: {}\nID: {} ".format(file.get("name"), file_id))
+        LOGGER.info(
+            f'♻️ Created G-Drive Folder:\nName: {file.get("name")}\nID: {file_id} '
+        )
+
         return file_id
 
     def upload_dir(self, input_directory, parent_id):
@@ -724,7 +725,7 @@ def drive_list(self, fileName, stopDup=False, noMulti=False):
                             msg += f' <b>| <a href="{url}">Index Link</a></b>'
                 elif file.get('mimeType') == 'application/vnd.google-apps.shortcut':
                     msg += f"⁍<a href='https://drive.google.com/drive/folders/{file.get('id')}'>{file.get('name')}" \
-                        f"</a> (shortcut)"
+                            f"</a> (shortcut)"
                     # Excluded index link as indexes cant download or open these shortcuts
                 else:
                     furl = f"https://drive.google.com/uc?id={file.get('id')}&export=download"
@@ -826,12 +827,12 @@ def count(self, link):
         except Exception as err:
             err = str(err).replace('>', '').replace('<', '')
             LOGGER.error(err)
-            if "File not found" in str(err):
+            if "File not found" in err:
                 token_service = self.alt_authorize()
                 if token_service is not None:
                     self.__service = token_service
                     return self.count(link)
-                msg = "File not found." 
+                msg = "File not found."
             else:
                 msg = f"Error.\n{err}"
         return msg
@@ -880,12 +881,12 @@ def helper(self, link):
         except Exception as err:
             err = str(err).replace('>', '').replace('<', '')
             LOGGER.error(err)
-            if "File not found" in str(err):
+            if "File not found" in err:
                 token_service = self.alt_authorize()
                 if token_service is not None:
                     self.__service = token_service
                     return self.helper(link)
-                msg = "File not found."  
+                msg = "File not found."
             else:
                 msg = f"Error.\n{err}"
             return msg, "", "", ""
@@ -909,9 +910,9 @@ def download(self, link):
                 err = err.last_attempt.exception()
             err = str(err).replace('>', '').replace('<', '')
             LOGGER.error(err)
-            if "downloadQuotaExceeded" in str(err):
+            if "downloadQuotaExceeded" in err:
                 err = "Download Quota Exceeded."
-            elif "File not found" in str(err):
+            elif "File not found" in err:
                 token_service = self.alt_authorize()
                 if token_service is not None:
                     self.__service = token_service
@@ -928,7 +929,7 @@ def download(self, link):
     def download_folder(self, folder_id, path, folder_name):
         if not os.path.exists(path + folder_name):
             os.makedirs(path + folder_name)
-        path += folder_name + '/'
+        path += f'{folder_name}/'
         result = []
         page_token = None
         while True:
@@ -965,7 +966,7 @@ def download_folder(self, folder_id, path, folder_name):
     def download_file(self, file_id, path, filename, mime_type):
         request = self.__service.files().get_media(fileId=file_id)
         filename = filename.replace('/', '')
-        fh = io.FileIO('{}{}'.format(path, filename), 'wb')
+        fh = io.FileIO(f'{path}{filename}', 'wb')
         downloader = MediaIoBaseDownload(fh, request, chunksize = 65 * 1024 * 1024)
         done = False
         while not done:

bot/modules/mirror.py

@@ -84,7 +84,7 @@ def onDownloadComplete(self):
                     download_dict[self.uid] = TarStatus(name, m_path, size)
                 if self.isZip:
                     pswd = self.pswd
-                    path = m_path + ".zip"
+                    path = f"{m_path}.zip"
                     LOGGER.info(f'Zip: orig_path: {m_path}, zip_path: {path}')
                     if pswd is not None:
                         subprocess.run(["7z", "a", "-mx=0", f"-p{pswd}", path, m_path])
@@ -112,8 +112,8 @@ def onDownloadComplete(self):
                     for dirpath, subdir, files in os.walk(m_path, topdown=False):
                         for filee in files:
                             if re.search(r'\.part0*1.rar$', filee) or re.search(r'\.7z.0*1$', filee) \
-                               or (filee.endswith(".rar") and not re.search(r'\.part\d+.rar$', filee)) \
-                               or re.search(r'\.zip.0*1$', filee):
+                                   or (filee.endswith(".rar") and not re.search(r'\.part\d+.rar$', filee)) \
+                                   or re.search(r'\.zip.0*1$', filee):
                                 m_path = os.path.join(dirpath, filee)
                                 if pswd is not None:
                                     result = subprocess.run(["7z", "x", f"-p{pswd}", m_path, f"-o{dirpath}"])
@@ -124,7 +124,7 @@ def onDownloadComplete(self):
                                 break
                         for filee in files:
                             if filee.endswith(".rar") or re.search(r'\.r\d+$', filee) \
-                               or re.search(r'\.7z.\d+$', filee) or re.search(r'\.zip.\d+$', filee):
+                                   or re.search(r'\.7z.\d+$', filee) or re.search(r'\.zip.\d+$', filee):
                                 del_path = os.path.join(dirpath, filee)
                                 os.remove(del_path)
                     path = f'{DOWNLOAD_DIR}{self.uid}/{name}'
@@ -251,10 +251,10 @@ def onUploadComplete(self, link: str, size, files, folders, typ):
             return
         with download_dict_lock:
             msg = f'<b>🗂 𝗙𝗶𝗹𝗲𝗡𝗮𝗺𝗲 : </b><code>{download_dict[self.uid].name()}</code>\n\n<b>Size: </b>{size}\n' \
-                  f' \n' \
-                  f'🔥 𝙔𝙖𝙢𝙧𝙖𝙖𝙟 𝙈𝙞𝙧𝙧𝙤𝙧 𝙕𝙤𝙣𝙀\n' \
-                  f' \n' \
-                  f'🔥 𝙂𝙧𝙤𝙪𝙥 : @mirrorclouds\n'
+                      f' \n' \
+                      f'🔥 𝙔𝙖𝙢𝙧𝙖𝙖𝙟 𝙈𝙞𝙧𝙧𝙤𝙧 𝙕𝙤𝙣𝙀\n' \
+                      f' \n' \
+                      f'🔥 𝙂𝙧𝙤𝙪𝙥 : @mirrorclouds\n'
             if os.path.isdir(f'{DOWNLOAD_DIR}/{self.uid}/{download_dict[self.uid].name()}'):
                 msg += '\n\n<b>Type: </b>Folder'
                 msg += f'\n<b>SubFolders: </b>{folders}'
@@ -374,12 +374,8 @@ def _mirror(bot, update, isTar=False, extract=False, isZip=False, isQbit=False,
     link = link.strip()
     reply_to = update.message.reply_to_message
     if reply_to is not None:
-        file = None
         media_array = [reply_to.document, reply_to.video, reply_to.audio]
-        for i in media_array:
-            if i is not None:
-                file = i
-                break
+        file = next((i for i in media_array if i is not None), None)
         if (
             not bot_utils.is_url(link)
             and not bot_utils.is_magnet(link)