Created
February 20, 2017 10:51
-
-
Save cgspeck/210c3f32e3c5e1e98b27b9e205fd45bb to your computer and use it in GitHub Desktop.
Quick and dirty script to do a batched migration of books to a new Calibre library running in Docker
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #! /usr/bin/python3 | |
| import logging | |
| import glob | |
| import os | |
| import subprocess | |
| from pathlib import Path, PosixPath | |
| logger = logging.getLogger(__name__) | |
| logger.setLevel(logging.INFO) | |
| # create console handler and set level to debug | |
| ch = logging.StreamHandler() | |
| ch.setLevel(logging.DEBUG) | |
| # create formatter | |
| formatter = logging.Formatter( | |
| '%(asctime)s - %(name)s - %(levelname)s - %(message)s' | |
| ) | |
| # add formatter to ch | |
| ch.setFormatter(formatter) | |
| # add ch to logger | |
| logger.addHandler(ch) | |
| MIGRATE_DIR = '/media/mm/books-migrate' | |
| MIGRATE_DIR_POSIX = PosixPath(MIGRATE_DIR) | |
| MIGRATE_DIR_PARTS = len(MIGRATE_DIR_POSIX.parts) | |
| IMPORT_DIR = '/media/mm/books-import' | |
| FAILED_DIR = '/media/mm/books-failed' | |
| BATCH_SIZE = 30 | |
| migrate_path = Path(MIGRATE_DIR) | |
| immediate_subdirs = [x for x in migrate_path.iterdir() if x.is_dir] | |
| immediate_subdir_count = len(immediate_subdirs) | |
| logger.info("{count} immediate sub folders to process".format(count=immediate_subdir_count)) | |
| BOOK_FORMAT_WHITELIST = [ | |
| 'azw', | |
| 'chm', | |
| 'doc', | |
| 'epub', | |
| 'mobi', | |
| 'pdf', | |
| 'prc', | |
| 'rar', | |
| 'rtf', | |
| 'zip' | |
| ] | |
| i = 0 | |
| def run_docker(): | |
| logger.info('Launching Docker.') | |
| subprocess.run('docker exec -it calibre calibredb add /import --with-library /books', shell=True, check=True) | |
| logger.info('Docker run complete.') | |
| logger.info('Cleaning up.') | |
| subprocess.run("find {IMPORT_DIR} -type f -delete".format(IMPORT_DIR=IMPORT_DIR), shell=True, check=True) | |
| for migrate_path in immediate_subdirs: | |
| logger.info("Searching for media in %s" % migrate_path) | |
| paths = [PosixPath(g) for g in glob.glob(os.path.join(migrate_path, '**'), recursive=True)] | |
| file_paths = [pf for pf in paths if pf.is_file()] | |
| candidate_files = [fp for fp in file_paths if fp.parts[-1].lower().split('.')[-1] in BOOK_FORMAT_WHITELIST] | |
| for candidate_file in candidate_files: | |
| MovedPath = PosixPath(os.path.join(IMPORT_DIR, candidate_file.parts[-1])) | |
| print("Moving {src} to {dst}".format(src=candidate_file, dst=MovedPath)) | |
| os.rename(candidate_file, MovedPath) | |
| i += 1 | |
| if i > 0 and i % BATCH_SIZE == 0: | |
| run_docker() | |
| logger.info("{i} candidate files processed, {remaining} to go".format(i=i, remaining=immediate_subdir_count-i)) | |
| run_docker() | |
| logger.info("{i} candidate files processed, {remaining} to go".format(i=i, remaining=immediate_subdir_count-i)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment