Atualizar m4b/cli.py
This commit is contained in:
+295
-2
@@ -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()
|
||||
sys.exit(cli())
|
||||
|
||||
|
||||
Reference in New Issue
Block a user