diff --git a/m4b/cli.py b/m4b/cli.py index ed18f67..ab95d6b 100644 --- a/m4b/cli.py +++ b/m4b/cli.py @@ -1,8 +1,301 @@ +#!/usr/bin/env python3 +import os, sys, subprocess, shutil, logging, argparse, dataclasses, collections, pprint, typing, pathlib, zipfile, mimetypes +import xml.etree.ElementTree as ET +from datetime import datetime as dt + +import mutagen + +@dataclasses.dataclass +class Config: + #input_path: pathlib.Path = None + #output_path: pathlib.Path = None + #merge_input: str = None + #merge_output: str = None + + # Default Fields + Tool = collections.namedtuple("Tool", ["name", "path", "flags"]) + m4b = Tool("m4b", "m4b-tool", + [ + # "--add-silence=1000", + "--use-filenames-as-chapters", + # "-vv", + ]) + timeout: int = 30 + logger: logging.Logger = None + logstream: typing.TextIO = sys.stdout + name: str = __name__ + logfmt: str = '[%(name)s:%(funcName)s][%(filename)s:%(lineno)04d][%(relativeCreated).3f][%(levelname)4s] %(message)s' + loglevel: int = logging.INFO + logfile: [str | Path] = None + exit: typing.Callable = sys.exit + # Audio types supported in epub files (media overlay) + # See browser support: https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/MIME_types/Common_types + # See epub specs: https://www.w3.org/TR/epub/#sec-core-media-types + AudioType = collections.namedtuple("AudioType",["type","suffix"]) + epub_audio_types = ("audio/mp3", "audio/mp4", "audio/ogg") + epub_audio_suffixes: tuple = (".mp3", ".mp4", ".m4a", ".ogg",".m4b") + tmp_dir: str = None + tmp_suffix: str = "_workdir" + + + def __post_init__(self): + self.create_logger() + + for tool_name in [f.name for f in dataclasses.fields(self) if isinstance(f.type, Config.Tool)]: + tool_path = self.gettattr(tool_name) + if not shutil.which(tool_path): + self.error("%s (%s) not found in $PATH", tool_name, tool_path) + self.exit(1) + + p = argparse.ArgumentParser() + p.add_argument("input", type=str, help="input directory") + p.add_argument("-o","--output", type=str, help="output directory") + p.add_argument("--debug", dest="loglevel", action="store_const",const="DEBUG", help="Set log level to DEBUG") + p.add_argument("--extract", action="store_true", help="Extract file if compressed archive") + p.add_argument("--chapters", action="store_true", help="Parse chapters from epub file") + self.args = vars(p.parse_args()) + #for a in self.args: + # setattr(self, a, self.args[a]) + + self.info(self.args) + + # Input + input_path = pathlib.Path(self.args['input']) + if not input_path.exists(): + self.error("Invalid input dir: '%s'", self.args['input']) + self.exit(1) + if input_path.is_dir(): + self.merge_input = input_path.resolve() + elif zipfile.is_zipfile(input_path): + self.tmp_dir = input_path.with_name(input_path.name + self.tmp_suffix) + if self.args['extract']: + self.extract_input(input_path) + self.merge_input = self.tmp_dir + + if self.args['chapters']: + self.parse_chapters(self.tmp_dir) + + else: + self.error("Invalid input dir: '%s'", self.args['input']) + self.exit(1) + + # Output + if not self.args['output']: + output_path = input_path.parent + output_name = input_path.with_suffix(".m4b").name + else: + output_path = pathlib.Path(self.args['output']) + if output_path.suffix == ".m4b": + output_name = output_path.name + output_path = output_path.parent + + self.merge_output = output_path.resolve() / output_name + + def extract_input(self, input_path): + self.debug("Input is compressed archive: %s", input_path) + tmp_dir = pathlib.Path(self.tmp_dir).resolve() if self.tmp_dir else input_path.with_name(input_path.name + self.tmp_suffix).resolve() + os.makedirs(tmp_dir, exist_ok=True) + if not tmp_dir.exists(): + self.error("Could not create temporary dir to extract files: %s", tmp_dir) + self.exit(1) + self.generate_epub_audio_suffixes() + with zipfile.ZipFile(input_path) as compressed: + compressed.extractall(tmp_dir) + + # audio_files = [] + # #for file in compressed.namelist(): + # for file_info in compressed.infolist(): + # filename = pathlib.Path(file_info.filename) + # if not filename.suffix in self.epub_audio_suffixes: + # self.debug("Ignoring file with unsupported suffix: %s", filename) + # continue + # with compressed.open(file_info, 'r') as src: + # audio_file = mutagen.File(src) + # #guessed_type, _ = mimetypes.guess_file_type(src) + # guessed_type = audio_file.mime[0] + # if guessed_type in self.epub_audio_types: + # target = tmp_dir / filename.name + # self.debug("Extracting %s (%s) to %s", filename.name, guessed_type, target) + # with open(target, 'wb') as dst: + # shutil.copyfileobj(src, dst) + # #dst.write(src) + # audio_files.append(target) + # else: + # self.error("Guessed mimetype is not supported: %s", guessed_type) + # if len(audio_files)==0: + # self.error("No supported audio files in compressed archive: %s", input_path) + # self.exit(1) + # else: + # for file in audio_files: + # if not file.exists(): + # self.error("Error extracting (file does not exist): %s", file) + # if not mutagen.File(file).mime[0] in self.epub_audio_types: + # self.error("Error extracting (wrong mimetype): %s", file) + # if not file.parent == tmp_dir: + # self.error("Error extracting (wrong target path): %s", file) + # self.info("Verified %s", file) + return tmp_dir + + + def generate_epub_audio_suffixes(self): + if not self.epub_audio_suffixes: + # Possible epub with media overlay inside + suffix_to_mimetype = {} + for mimetype in self.epub_audio_types: + for suffix in mimetypes.guess_all_extensions(mimetype): + suffix_to_mimetype[suffix] = mimetype + self.epub_audio_suffixes = tuple(suffix_to_mimetype) + self.write("Generated audio suffixes: %s", str(self.epub_audio_suffixes)) + + def create_logger(self): + #logging.basicConfig(format = self.logfmt) + + # Get Logger + self.logger = logging.getLogger(self.name) + + # Log Level + if "--debug" in sys.argv: + self.loglevel = logging.DEBUG + if isinstance(self.loglevel, str): + self.loglevel = getattr(logging, self.loglevel, Config.loglevel) + self.logger.setLevel(self.loglevel) + + # Log Format + + #self.formatter = logging.Formatter(self.logfmt) + class Formatter(logging.Formatter): + def format(self, record): + # Calculate seconds directly onto the record object + record.relativeCreated = record.relativeCreated / 1000.0 + return super().format(record) + self.formatter = Formatter(self.logfmt) + + # File Logger + if self.logfile: + self.filehandler = logging.FileHandler(self.logfile) + self.filehandler.setLevel(self.loglevel) + self.filehandler.setFormatter(self.formatter) + self.logger.addHandler(self.filehandler) + + # Console Logger + self.streamhandler = logging.StreamHandler(self.logstream) + self.streamhandler.setLevel(self.loglevel) + self.streamhandler.setFormatter(self.formatter) + self.logger.addHandler(self.streamhandler) + + def log(self, loglevel, msg, *args): + if not msg: + return + if not loglevel: + loglevel = self.loglevel + if args: + self.logger.log(loglevel, msg, *args, stacklevel=3) + elif isinstance(msg, str): + for line in msg.splitlines(): + self.logger.log(loglevel, line, stacklevel=3) + else: + self.logger.log(loglevel, msg, stacklevel=3) + + def error(self, msg, *args): + self.log(logging.ERROR, msg, *args) + + def write (self, msg, *args): + self.log(logging.INFO, msg, *args) + + def info(self, msg, *args): + self.log(logging.INFO, msg, *args) + + def warning(self, msg, *args): + self.log(logging.WARNING, msg, *args) + + def debug(self, msg, *args): + self.log(logging.DEBUG, msg, *args) + + + def run(self, cmd): + if not cmd: + self.error("Empty cmd") + return 1 + self.info(cmd) + proc = subprocess.Popen(cmd, text=True, + stdout = subprocess.PIPE, + stderr = subprocess.PIPE) + try: + outs,errs = proc.communicate(timeout=self.timeout) + self.write(outs) + self.error(errs) + except subprocess.TimeoutExpired: + proc.kill() + outs,errs = proc.communicate() + self.write(outs) + self.error(errs) + return 1 + + return proc.returncode + + def parse_chapters(self, path): + + # Content + content = ET.parse(path/'content.opf').getroot() + xmlns = {"opf": "http://www.idpf.org/2007/opf", + "dc": "http://purl.org/dc/elements/1.1/", + "dcterms": "http://purl.org/dc/terms/", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "calibre": "http://calibre.kovidgoyal.net/2009/metadata", + "xhtml": "http://www.w3.org/1999/xhtml", + "epub": "http://www.idpf.org/2007/ops", + "ncx": "http://www.daisy.org/z3986/2005/ncx/", + } + + @dataclasses.dataclass + class Item: + id: str + href: pathlib.Path + mimetype: str + order: int = -1 + title: str = '' + + items: dict[str,Item] = {} + for item in content.findall(f"./manifest/item", {'':xmlns['opf']}): + parsed = Item(item.attrib['id'], pathlib.Path(item.attrib['href']), item.attrib['media-type']) + items[parsed.id] = parsed + self.info(items[parsed.id]) + + # Spine + for i,itemref in enumerate(content.findall(f"./spine/itemref", {'':xmlns['opf']})): + idref = itemref.attrib['idref'] + if idref in items: + items[idref].order = i + + # Nav + nav = ET.parse(path/items['nav'].href).getroot() + for a in nav.find(f".//nav[@id='toc']", {'':xmlns['xhtml']}).findall(f".//a", {'':xmlns['xhtml']}): + href = pathlib.Path(a.attrib['href']) + title = a.text + if href.name in items: + items[href.name].title = title + + for item in sorted(items.values(), key=lambda i: i.order): + self.debug("%03d. [%s](%s)", item.order, item.title, item.href) + + # Audio + for k in [k for k,v in items.items() if v.mimetype.startswith("audio/")]: + target = items[k.removesuffix("-audio")] + order, title = target.order, target.title + items[k].order = order + items[k].title = title + audiopath = path/items[k].href + newpath = audiopath.with_stem(" ".join([f"{order:04d}", title[:32], f"[{audiopath.stem}]"])) + self.debug("Renaming '%s' -> '%s'", audiopath.name, newpath.name) + audiopath.move(newpath) def cli(): - pass + config = Config() + return config.run([config.m4b.path] + config.m4b.flags + + ["merge", "-o", config.merge_output, config.merge_input]) if __name__ == "__main__": - cli() \ No newline at end of file + sys.exit(cli()) +