# Calibre-Web Automated – fork of Calibre-Web # Copyright (C) 2018-2025 Calibre-Web contributors # Copyright (C) 2024-2025 Calibre-Web Automated contributors # SPDX-License-Identifier: GPL-3.0-or-later # See CONTRIBUTORS for full list of authors. import argparse import json import logging import os import re import sys import shutil from pathlib import Path import subprocess import tempfile import atexit from datetime import datetime import sqlite3 import pwd import grp from cwa_db import CWA_DB from kindle_epub_fixer import EPUBFixer ### Global Variables convert_library_log_file = "/config/convert-library.log" # Define the logger logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # Set the logging level # Create a FileHandler file_handler = logging.FileHandler(convert_library_log_file, mode='w', encoding='utf-8') # Create a Formatter and set it for the handler LOG_FORMAT = '%(message)s' formatter = logging.Formatter(LOG_FORMAT) file_handler.setFormatter(formatter) # Add the handler to the logger logger.addHandler(file_handler) # Define user and group USER_NAME = "abc" GROUP_NAME = "abc" # Get UID and GID uid = pwd.getpwnam(USER_NAME).pw_uid gid = grp.getgrnam(GROUP_NAME).gr_gid # Set permissions for log file (skip on network shares) try: nsm = os.getenv("NETWORK_SHARE_MODE", "false").strip().lower() in ("1", "true", "yes", "on") if not nsm: subprocess.run(["chown", f"{uid}:{gid}", convert_library_log_file], check=True) else: print(f"[convert-library] NETWORK_SHARE_MODE=true detected; skipping chown of {convert_library_log_file}", flush=True) except subprocess.CalledProcessError as e: print(f"[convert-library] An error occurred while attempting to set ownership of {convert_library_log_file} to abc:abc. See the following error:\n{e}", flush=True) def print_and_log(string) -> None: """ Ensures the provided string is passed to STDOUT and stored in the runs log file """ logger.info(string) print(string) # Creates a lock file unless one already exists meaning an instance of the script is # already running, then the script is closed, the user is notified and the program # exits with code 2 try: lock = open(tempfile.gettempdir() + '/convert_library.lock', 'x') lock.close() except FileExistsError: print_and_log("[convert-library]: CANCELLING... convert-library was initiated but is already running") logger.info(f"\nCWA Convert Library Service - Run Cancelled: {datetime.now()}") sys.exit(2) # Defining function to delete the lock on script exit def removeLock(): try: os.remove(tempfile.gettempdir() + '/convert_library.lock') except FileNotFoundError: ... # Will automatically run when the script exits atexit.register(removeLock) backup_destinations = { entry.name: entry.path for entry in os.scandir("/config/processed_books") if entry.is_dir() } class LibraryConverter: def __init__(self, args) -> None: self.args = args self.verbose = args.verbose self.db = CWA_DB() self.cwa_settings = self.db.cwa_settings self.target_format = self.cwa_settings['auto_convert_target_format'] self.convert_ignored_formats = self.cwa_settings['auto_convert_ignored_formats'] self.kindle_epub_fixer = self.cwa_settings['kindle_epub_fixer'] self.supported_book_formats = {'acsm', 'azw', 'azw3', 'azw4', 'cbz', 'cbr', 'cb7', 'cbc', 'chm', 'djvu', 'docx', 'epub', 'fb2', 'fbz', 'html', 'htmlz', 'lit', 'lrf', 'mobi', 'odt', 'pdf', 'prc', 'pdb', 'pml', 'rb', 'rtf', 'snb', 'tcr', 'txt', 'txtz'} self.hierarchy_of_success = {'epub', 'lit', 'mobi', 'azw', 'azw3', 'fb2', 'fbz', 'azw4', 'prc', 'odt', 'lrf', 'pdb', 'cbz', 'pml', 'rb', 'cbr', 'cb7', 'cbc', 'chm', 'djvu', 'snb', 'tcr', 'pdf', 'docx', 'rtf', 'html', 'htmlz', 'txtz', 'txt'} self.current_book = 1 self.ingest_folder, self.library_dir, self.tmp_conversion_dir = self.get_dirs('/app/calibre-web-automated/dirs.json') self.to_convert = self.get_books_to_convert() self.calibre_env = os.environ.copy() # Enables Calibre plugins to be used from /config/plugins self.calibre_env["HOME"] = "/config" # Gets split library info from app.db and sets library dir to the split dir if split library is enabled self.split_library = self.get_split_library() if self.split_library: self.library_dir = self.split_library["split_path"] self.calibre_env['CALIBRE_OVERRIDE_DATABASE_PATH'] = os.path.join(self.split_library["db_path"], "metadata.db") def get_split_library(self) -> dict[str, str] | None: """Checks whether or not the user has split library enabled. Returns None if they don't and the path of the Split Library location if True.""" con = sqlite3.connect("/config/app.db", timeout=30) cur = con.cursor() split_library = cur.execute('SELECT config_calibre_split FROM settings;').fetchone()[0] if split_library: split_path = cur.execute('SELECT config_calibre_split_dir FROM settings;').fetchone()[0] db_path = cur.execute('SELECT config_calibre_dir FROM settings;').fetchone()[0] con.close() return { "split_path":split_path, "db_path":db_path } else: con.close() return None def get_dirs(self, dirs_json_path: str) -> tuple[str, str, str]: dirs = {} with open(dirs_json_path, 'r') as f: dirs: dict[str, str] = json.load(f) ingest_folder = f"{dirs['ingest_folder']}/" library_dir = f"{dirs['calibre_library_dir']}/" tmp_conversion_dir = f"{dirs['tmp_conversion_dir']}/" return ingest_folder, library_dir, tmp_conversion_dir def get_books_to_convert(self): library_files = [os.path.join(dirpath,f) for (dirpath, dirnames, filenames) in os.walk(self.library_dir) for f in filenames] exclusion_list = [] # If multiple formats for a book exist, only the one with the highest success rate will be converted and the rest will be left alone files_already_in_target_format = [f for f in library_files if f.endswith(f'.{self.target_format}')] for file in files_already_in_target_format: filename, file_extension = os.path.splitext(file) exclusion_list.append(filename) # Adding books with a file already in the target format to the exclusion list to_convert = [] # Will only contain a single filepath for each book without an existing file in the target format in the format with the highest available conversion success rate, where that filepath is allow to be converted for format in self.hierarchy_of_success: if format in self.convert_ignored_formats: print_and_log(f"{format} in list of user-defined ignored formats for conversion. To change this, navigate to the CWA Settings panel from the Settings page in the Web UI.") continue files_in_format = [f for f in library_files if f.endswith(f'.{format}')] if len(files_in_format) > 0: for file in files_in_format: filename, file_extension = os.path.splitext(file) if filename not in exclusion_list: to_convert.append(file) exclusion_list.append(filename) return to_convert def backup(self, input_file, backup_type): try: output_path = backup_destinations[backup_type] shutil.copy2(input_file, output_path) except Exception as e: print_and_log(f"[convert-library]: ERROR - The following error occurred when trying to copy {input_file} to {output_path}:\n{e}") def convert_library(self): for file in self.to_convert: filename = os.path.basename(file) file_extension = Path(file).suffix 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 from the immediate book folder (e.g., "Title (6120)") book_folder = os.path.basename(os.path.dirname(file)) m = re.search(r"\((\d+)\)$", book_folder) book_id = m.group(1) # type: ignore[attr-defined] except Exception as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) 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/") print_and_log("└── Wizard's First Rule (6120)") print_and_log(" ├── cover.jpg") print_and_log(" ├── metadata.opf") print_and_log(" └── Wizard's First Rule - Terry Goodkind.epub") self.backup(file, backup_type="failed") self.current_book += 1 continue if self.target_format == "kepub": convert_successful, target_filepath = self.convert_to_kepub(file, file_extension) if not convert_successful: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Conversion of {os.path.basename(file)} was unsuccessful. Moving to next book...") self.current_book += 1 continue else: try: # Convert Book to target format (target is not kepub) target_filepath = f"{self.tmp_conversion_dir}{Path(file).stem}.{self.target_format}" with subprocess.Popen( ["ebook-convert", file, target_filepath], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=self.calibre_env, text=True, encoding='utf-8' ) as process: for line in process.stdout: # Read from the combined stdout (which includes stderr) if self.verbose: print_and_log(line) else: print(line) if self.cwa_settings['auto_backup_conversions']: self.backup(file, backup_type="converted") self.db.conversion_add_entry(os.path.basename(target_filepath), Path(file).suffix, self.target_format, str(self.cwa_settings["auto_backup_conversions"])) print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Conversion of {os.path.basename(file)} to {self.target_format} format successful!") # Removed as of V3.0.0 - Removing old version from library... except subprocess.CalledProcessError as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Conversion of {os.path.basename(file)} was unsuccessful. See the following error:\n{e}") self.current_book += 1 continue if self.target_format == "epub" and self.kindle_epub_fixer: try: EPUBFixer().process(input_path=target_filepath) print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Resulting EPUB file successfully processed by CWA-EPUB-Fixer!") except Exception as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) An error occurred while processing {os.path.basename(target_filepath)} with the kindle-epub-fixer. See the following error:\n{e}") try: # Import converted book to library. As of V3.0.0, "add_format" is used instead of "add" with subprocess.Popen( ["calibredb", "add_format", book_id, target_filepath, f"--library-path={self.library_dir}"], env=self.calibre_env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding='utf-8' ) as process: for line in process.stdout: # Read from the combined stdout (which includes stderr) if self.verbose: print_and_log(line) else: print(line) if self.cwa_settings['auto_backup_imports']: self.backup(target_filepath, backup_type="imported") self.db.import_add_entry(os.path.basename(target_filepath), str(self.cwa_settings["auto_backup_imports"])) print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Import of {os.path.basename(target_filepath)} successfully completed!") except subprocess.CalledProcessError as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Import of {os.path.basename(target_filepath)} was not successfully completed. Converted file moved to /config/processed_books/failed/{os.path.basename(target_filepath)}. See the following error:\n{e}") try: output_path = f"/config/processed_books/failed/{os.path.basename(target_filepath)}" shutil.move(target_filepath, output_path) except Exception as e: print_and_log(f"[convert-library]: ERROR - The following error occurred when trying to copy {file} to {output_path}:\n{e}") self.current_book += 1 continue self.set_library_permissions() self.empty_tmp_con_dir() self.current_book += 1 continue def convert_to_kepub(self, filepath:str ,import_format:str) -> 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 import_format == "epub": print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) File already in epub format, converting directly to kepub...") if self.cwa_settings['auto_backup_conversions']: self.backup(filepath, backup_type="converted") epub_filepath = filepath epub_ready = True else: print_and_log(f"\n[convert-library]: ({self.current_book}/{len(self.to_convert)}) *** 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") try: # Convert book to epub format so it can then be converted to kepub epub_filepath = f"{self.tmp_conversion_dir}{Path(filepath).stem}.epub" with subprocess.Popen( ["ebook-convert", filepath, epub_filepath], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=self.calibre_env, text=True ) as process: for line in process.stdout: # Read from the combined stdout (which includes stderr) if self.verbose: print_and_log(line) else: print(line) if self.cwa_settings['auto_backup_conversions']: self.backup(filepath, backup_type="converted") print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Intermediate conversion of {os.path.basename(filepath)} to epub from {import_format} successful, now converting to kepub...") epub_ready = True except subprocess.CalledProcessError as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Intermediate conversion of {os.path.basename(filepath)} to epub was unsuccessful. Cancelling kepub conversion and moving on to next file. See the following error:\n{e}") return False, "" if epub_ready: epub_filepath = Path(epub_filepath) target_filepath = f"{self.tmp_conversion_dir}{epub_filepath.stem}.kepub" try: with subprocess.Popen( ['kepubify', '--inplace', '--calibre', '--output', self.tmp_conversion_dir, epub_filepath], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding='utf-8' ) as process: for line in process.stdout: # Read from the combined stdout (which includes stderr) if self.verbose: print_and_log(line) else: print(line) if self.cwa_settings['auto_backup_conversions']: self.backup(filepath, backup_type="converted") self.db.conversion_add_entry(epub_filepath.stem, import_format, self.target_format, str(self.cwa_settings["auto_backup_conversions"])) return True, target_filepath except subprocess.CalledProcessError as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) CON_ERROR: {os.path.basename(filepath)} could not be converted to kepub due to the following error:\nEXIT/ERROR CODE: {e.returncode}\n{e.stderr}") self.backup(epub_filepath, backup_type="failed") return False, "" else: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) An error occurred when converting the original {import_format} to epub. Cancelling kepub conversion and moving on to next file...") return False, "" def empty_tmp_con_dir(self): try: files = os.listdir(self.tmp_conversion_dir) for file in files: file_path = os.path.join(self.tmp_conversion_dir, file) if os.path.isfile(file_path): os.remove(file_path) except OSError: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) An error occurred while emptying {self.tmp_conversion_dir}.") def set_library_permissions(self): try: nsm = os.getenv("NETWORK_SHARE_MODE", "false").strip().lower() in ("1", "true", "yes", "on") if not nsm: subprocess.run(["chown", "-R", "abc:abc", self.library_dir], check=True) print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) Successfully set ownership of new files in {self.library_dir} to abc:abc.") else: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) NETWORK_SHARE_MODE=true detected; skipping chown of {self.library_dir}") except subprocess.CalledProcessError as e: print_and_log(f"[convert-library]: ({self.current_book}/{len(self.to_convert)}) An error occurred while attempting to recursively set ownership of {self.library_dir} to abc:abc. See the following error:\n{e}") def main(): parser = argparse.ArgumentParser( prog='convert-library', description='Made for the purpose of converting ebooks in a calibre library to the users specified target format (default epub)' ) parser.add_argument('--verbose', '-v', action='store_true', required=False, dest='verbose', help='When passed, the output from the ebook-convert command will be included in what is shown to the user in the Web UI', default=False) args = parser.parse_args() logger.info(f"CWA Convert Library Service - Run Started: {datetime.now()}\n") converter = LibraryConverter(args) if len(converter.to_convert) > 0: converter.convert_library() else: print_and_log("[convert-library]: No books found in library without a copy in the target format. Exiting now...") logger.info(f"\nCWA Convert Library Service - Run Ended: {datetime.now()}") sys.exit(0) print_and_log(f"\n[convert-library]: Library conversion complete! {len(converter.to_convert)} books converted! Exiting now...") logger.info(f"\nCWA Convert Library Service - Run Ended: {datetime.now()}") sys.exit(0) if __name__ == "__main__": main()