Cleaned up numerous python scripts and fixed numerous minor errors

scripts/auto_library.py

@@ -91,7 +91,7 @@ def set_library_location(self):
 
     # Uses sql to update CW's app.db with the correct library location (config_calibre_dir in the settings table)
     def update_calibre_web_db(self):
-        if os.path.exists(self.metadb_path):
+        if os.path.exists(self.metadb_path): # type: ignore
             try:
                 print("[cwa-auto-library]: Updating Settings Database with library location...")
                 con = sqlite3.connect(self.app_db)

scripts/auto_zip.py

@@ -33,7 +33,7 @@ def last_mod_date(self, path_to_file) -> str:
         stat = os.stat(path_to_file)
         return datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d') #%H:%M:%S
 
-    def get_books_to_zip(self) -> dict[str:list[str]]:
+    def get_books_to_zip(self) -> dict[str,list[str]]:
         """ Returns a dictionary with the books that are to be zipped together in each dir """
         to_zip = {}
         for dir in self.archive_dirs:

scripts/convert_library.py

@@ -116,7 +116,7 @@ def convert_library(self):
             print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Converting {filename} from {file_extension} format to {self.target_format} format...")
 
             try: # Get Calibre Library Book ID
-                book_id = (re.search(r'\(\d*\)', file).group(0))[1:-1]
+                book_id = (re.search(r'\(\d*\)', file).group(0))[1:-1] # type: ignore
             except Exception as e:
                 print_and_log(f"[convert-library]: A Calibre Library Book ID could not be determined for {file}. Make sure the structure of your calibre library matches the following example:\n")
                 print_and_log("Terry Goodkind/")
@@ -132,7 +132,7 @@ def convert_library(self):
             if self.target_format == "kepub":
                 convert_successful, target_filepath = self.convert_to_kepub(filename, file_extension)
                 if not convert_successful:
-                    print_and_log(f"[convert-library]: Conversion of {os.path.basename(file)} was unsuccessful. See the following error:\n{e}")
+                    print_and_log(f"[convert-library]: Conversion of {os.path.basename(file)} was unsuccessful. Moving to next book...")
                     self.current_book += 1
                     continue
             else:
@@ -200,7 +200,7 @@ def convert_to_kepub(self, filepath:str ,import_format:str) -> tuple[bool, str]:
             print_and_log(f"[convert-library]: File in epub format, converting directly to kepub...")
 
             if self.cwa_settings['auto_backup_conversions']:
-                shutil.copyfile(file, f"/config/processed_books/converted/{os.path.basename(filepath)}")
+                shutil.copyfile(filepath, f"/config/processed_books/converted/{os.path.basename(filepath)}")
 
             epub_filepath = filepath
             epub_ready = True
@@ -211,7 +211,7 @@ def convert_to_kepub(self, filepath:str ,import_format:str) -> tuple[bool, str]:
                 subprocess.run(["ebook-convert", filepath, epub_filepath], check=True)
 
                 if self.cwa_settings['auto_backup_conversions']:
-                    shutil.copyfile(file, f"/config/processed_books/converted/{os.path.basename(filepath)}")
+                    shutil.copyfile(filepath, f"/config/processed_books/converted/{os.path.basename(filepath)}")
 
                 print_and_log(f"[convert-library]: Intermediate conversion of {os.path.basename(filepath)} to epub from {import_format} successful, now converting to kepub...")
                 epub_ready = True
@@ -259,6 +259,19 @@ def set_library_permissions(self):
         except subprocess.CalledProcessError as e:
             print_and_log(f"[convert-library] An error occurred while attempting to recursively set ownership of {self.library_dir} to abc:abc. See the following error:\n{e}")
 
+    # def process(self):
+    #     """ Allows LibraryConverter to be ran from an import  """
+    #     if len(self.to_convert) > 0:
+    #         self.convert_library()
+    #     else:
+    #         print_and_log("[convert-library] No books found in library without a copy in the target format. Exiting now...")
+    #         logging.info("FIN")
+    #         sys.exit(0)
+
+    #     print_and_log(f"\n[convert-library] Library conversion complete! {len(self.to_convert)} books converted! Exiting now...")
+    #     logging.info("FIN")
+    #     sys.exit(0)
+
 
 def main():
     # parser = argparse.ArgumentParser(

scripts/cover_enforcer.py

@@ -80,7 +80,7 @@ def get_new_metadata_path(self) -> str:
         temp_files = [os.path.join(dirpath,f) for (dirpath, dirnames, filenames) in os.walk(metadata_temp_dir) for f in filenames]
         return [f for f in temp_files if f.endswith('.opf')][0]
 
-    def export_as_dict(self) -> dict[str:str]:
+    def export_as_dict(self) -> dict[str,str | None]:
         return {"book_dir":self.book_dir,
                 "file_path":self.file_path,
                 "calibre_library":self.calibre_library,
@@ -167,7 +167,7 @@ def get_supported_files_from_dir(self, dir: str) -> list[str]:
 
         return supported_files
 
-    def enforce_cover(self, book_dir: str) -> bool | list[Book]:
+    def enforce_cover(self, book_dir: str) -> list:
         """Will force the Cover & Metadata to update for the supported book files in the given directory"""
         supported_files = self.get_supported_files_from_dir(book_dir)
         if supported_files:
@@ -179,20 +179,20 @@ def enforce_cover(self, book_dir: str) -> bool | list[Book]:
                 self.replace_old_metadata(book.old_metadata_path, book.new_metadata_path)
                 os.system(f'ebook-polish -c "{book.cover_path}" -o "{book.new_metadata_path}" -U "{file}" "{file}"')
                 self.empty_metadata_temp()
-                print(f"[cover-metadata-enforcer]: DONE: '{book.title_author}.{book.file_format}': Cover & Metadata updated", flash=True)
+                print(f"[cover-metadata-enforcer]: DONE: '{book.title_author}.{book.file_format}': Cover & Metadata updated", flush=True)
                 book_objects.append(book)
 
             return book_objects
         else:
             print(f"[cover-metadata-enforcer]: No supported file formats found in {book_dir}.", flush=True)
             print("[cover-metadata-enforcer]: *** NOTICE **** Only EPUB & AZW3 formats are currently supported.", flush=True)
-            return False
+            return []
 
     def enforce_all_covers(self) -> tuple[int, float, int] | tuple[bool, bool, bool]:
         """Will force the covers and metadata to be re-generated for all books in the library"""
         t_start = time.time()
 
-        supported_files = self.get_supported_files_from_book_dir(self.calibre_library)
+        supported_files = self.get_supported_files_from_dir(self.calibre_library)
         if supported_files:
             book_dirs = []
             for file in supported_files:
@@ -206,10 +206,11 @@ def enforce_all_covers(self) -> tuple[int, float, int] | tuple[bool, bool, bool]
             for book_dir in book_dirs:
                 try:
                     book_objects = self.enforce_cover(book_dir)
-                    book_dicts = []
-                    for book in book_objects:
-                        book_dicts.append(book.export_as_dict())
-                    self.db.enforce_add_entry_from_all(book_dicts)
+                    if book_objects:
+                        book_dicts = []
+                        for book in book_objects:
+                            book_dicts.append(book.export_as_dict())
+                        self.db.enforce_add_entry_from_all(book_dicts)
                 except Exception as e:
                     print(f"[cover-metadata-enforcer]: ERROR: {book_dir}")
                     print(f"[cover-metadata-enforcer]: Skipping book due to following error: {e}")
@@ -251,10 +252,11 @@ def check_for_other_logs(self):
                     log_info = self.read_log(auto=False, log_path=log)
                     book_dir = self.get_book_dir_from_log(log_info)
                     book_objects = self.enforce_cover(book_dir)
-                    for book in book_objects:
-                        book.log_info = log_info
-                        book.log_info['file_path'] = book.file_path
-                        self.db.enforce_add_entry_from_log(book.log_info)
+                    if book_objects:
+                        for book in book_objects:
+                            book.log_info = log_info
+                            book.log_info['file_path'] = book.file_path
+                            self.db.enforce_add_entry_from_log(book.log_info)
                     self.delete_log(auto=False, log_path=log)
 
 
@@ -310,10 +312,11 @@ def main():
             args.dir = args.dir[:-1]
         if os.path.isdir(args.dir):
             book_objects = enforcer.enforce_cover(args.dir)
-            book_dicts = []
-            for book in book_objects:
-                book_dicts.append(book.export_as_dict())
-            enforcer.db.enforce_add_entry_from_dir(book_dicts)
+            if book_objects:
+                book_dicts = []
+                for book in book_objects:
+                    book_dicts.append(book.export_as_dict())
+                enforcer.db.enforce_add_entry_from_dir(book_dicts)
         else:
             print(f"[cover-metadata-enforcer]: ERROR: '{args.dir}' is not a valid directory")
     elif args.log is not None and args.dir is None and args.all is False and args.list is False and args.history is False:
@@ -322,7 +325,7 @@ def main():
         book_dir = enforcer.get_book_dir_from_log(log_info)
         if enforcer.enforcer_on:
             book_objects = enforcer.enforce_cover(book_dir)
-            if book_objects == False:
+            if not book_objects:
                 print(f"[cover-metadata-enforcer] Metadata for '{log_info['book_title']}' not successfully enforced")
                 sys.exit(1)
             for book in book_objects:

scripts/cwa_db.py

@@ -14,7 +14,7 @@ def __init__(self, verbose=False):
 
         self.db_file = "cwa.db"
         self.db_path = "/config/"
-        self.con, self.cur = self.connect_to_db()
+        self.con, self.cur = self.connect_to_db() # type: ignore
 
         self.schema_path = "/app/calibre-web-automated/scripts/cwa_schema.sql"
 
@@ -140,7 +140,7 @@ def match_stat_table_columns_with_schema(self) -> None:
                     table_name = line[27:].replace('(', '')
                 elif line[:4] == "    ":
                     column_names.append(line.strip().split(' ')[0])
-            column_names_in_schema |= {table_name:column_names}
+            column_names_in_schema |= {table_name:column_names} # type: ignore
 
         for table in self.stats_tables:
             if len(current_column_names[table]) < len(column_names_in_schema[table]): # Adds new columns not yet in existing db
@@ -208,7 +208,7 @@ def set_default_settings(self, force=False) -> None:
             print("[cwa-db] CWA Settings loaded successfully")
 
 
-    def get_cwa_settings(self) -> dict[str:bool|str]:
+    def get_cwa_settings(self) -> dict:
         """Gets the current cwa_settings values from the table of the same name in cwa.db and returns them as a dict"""
         self.cur.execute("SELECT * FROM cwa_settings")
         headers = [header[0] for header in self.cur.description]
@@ -243,14 +243,14 @@ def enforce_add_entry_from_log(self, log_info: dict):
         self.con.commit()
 
 
-    def enforce_add_entry_from_dir(self, book_dicts: list[dict[str:str]]):
+    def enforce_add_entry_from_dir(self, book_dicts: list[dict[str,str]]):
         """Adds an entry to the db when cover_enforcer is ran with a directory"""
         for book in book_dicts:
             self.cur.execute("INSERT INTO cwa_enforcement(timestamp, book_id, book_title, author, file_path, trigger_type) VALUES (?, ?, ?, ?, ?, ?);", (book['timestamp'], book['book_id'], book['book_title'], book['author_name'], book['file_path'], 'manual -dir'))
             self.con.commit()
 
 
-    def enforce_add_entry_from_all(self, book_dicts: list[dict[str:str]]):
+    def enforce_add_entry_from_all(self, book_dicts: list[dict[str,str]]):
         """Adds an entry to the db when cover_enforcer is ran with the -all flag"""
         for book in book_dicts:
             self.cur.execute("INSERT INTO cwa_enforcement(timestamp, book_id, book_title, author, file_path, trigger_type) VALUES (?, ?, ?, ?, ?, ?);", (book['timestamp'], book['book_id'], book['book_title'], book['author_name'], book['file_path'], 'manual -all'))

scripts/ingest_processor.py

@@ -85,7 +85,7 @@ def can_convert_check(self) -> tuple[bool, str]:
         return can_convert, input_format
 
 
-    def convert_book(self, end_format: str=None) -> tuple[bool, str]:
+    def convert_book(self, end_format=None) -> tuple[bool, str]:
         """Uses the following terminal command to convert the books provided using the calibre converter tool:\n\n--- ebook-convert myfile.input_format myfile.output_format\n\nAnd then saves the resulting files to the calibre-web import folder."""
         print(f"\n[ingest-processor]: Starting conversion process for {self.filename}...", flush=True)
         print(f"[ingest-processor]: Converting file from {self.input_format} to {self.target_format} format...\n", flush=True)
@@ -120,7 +120,7 @@ def convert_book(self, end_format: str=None) -> tuple[bool, str]:
 
 
     # Kepubify can only convert EPUBs to Kepubs
-    def convert_to_kepub(self) -> None:
+    def convert_to_kepub(self) -> tuple[bool,str]:
         """Kepubify is limited in that it can only convert from epub to kepub, therefore any files not already in epub need to first be converted to epub, and then to kepub"""
         if self.input_format == "epub":
             print(f"[ingest-processor]: File in epub format, converting directly to kepub...", flush=True)
@@ -130,7 +130,7 @@ def convert_to_kepub(self) -> None:
             print("\n[ingest-processor]: *** NOTICE TO USER: Kepubify is limited in that it can only convert from epubs. To get around this, CWA will automatically convert other"
             "supported formats to epub using the Calibre's conversion tools & then use Kepubify to produce your desired kepubs. Obviously multi-step conversions aren't ideal"
             "so if you notice issues with your converted files, bare in mind starting with epubs will ensure the best possible results***\n", flush=True)
-            convert_successful, converted_filepath = self.convert_book(self.input_format, end_format="epub")
+            convert_successful, converted_filepath = self.convert_book(self.input_format, end_format="epub") # type: ignore
 
         if convert_successful:
             converted_filepath = Path(converted_filepath)
@@ -149,7 +149,7 @@ def convert_to_kepub(self) -> None:
 
             except subprocess.CalledProcessError as e:
                 print(f"[ingest-processor]: CON_ERROR: {self.filename} could not be converted to kepub due to the following error:\nEXIT/ERROR CODE: {e.returncode}\n{e.stderr}", flush=True)
-                shutil.copy2(converted_filepath, f"/config/processed_books/failed/{os.path.basename(original_filepath)}")
+                shutil.copy2(converted_filepath, f"/config/processed_books/failed/{os.path.basename(self.filepath)}")
                 return False, ""
         else:
             print(f"[ingest-processor]: An error occurred when converting the original {self.input_format} to epub. Cancelling kepub conversion...", flush=True)
@@ -240,7 +240,7 @@ def main(filepath=sys.argv[1]):
                     convert_successful, converted_filepath = nbp.convert_book()
 
                 if convert_successful: # If previous conversion process was successful, remove tmp files and import into library
-                    nbp.add_book_to_library(converted_filepath)
+                    nbp.add_book_to_library(converted_filepath) # type: ignore
 
             elif nbp.can_convert and not nbp.auto_convert_on: # Books not in target format but Auto-Converter is off so files are imported anyway
                 print(f"\n[ingest-processor]: {nbp.filename} not in target format but CWA Auto-Convert is deactivated so importing the file anyway...", flush=True)

scripts/kindle_epub_fixer.py

@@ -54,7 +54,7 @@ def fix_language(self):
 
                 if lang_tag is None:
                     metadata = root.find(".//{http://www.idpf.org/2007/opf}metadata")
-                    lang_tag = ET.SubElement(metadata, "{http://purl.org/dc/elements/1.1/}language")
+                    lang_tag = ET.SubElement(metadata, "{http://purl.org/dc/elements/1.1/}language") # type: ignore
                 lang_tag.text = new_lang
 
                 self.files[opf_file] = ET.tostring(root, encoding='unicode')