feat(audiobook_library): functional write_metadata, create_m4b
shebang python venv get all metadata tags kid3 command corrections, add "" to fields sort self.files add changed flag to fix_... cmds fix series numbering
This commit is contained in:
Lucas Vivo committed
1 parent
bcc927db73
commit
815cd5a551
1 file changed
+167
-71
@@ -1,4 +1,4 @@
|
||||
#!/usr/bin/env python3
|
||||
#!/home/lucas/scripts/.venv/bin/python
|
||||
|
||||
import argparse
|
||||
import json
|
||||
@@ -96,10 +96,42 @@ def get_chapters(media: Path) -> list[dict]:
|
||||
data = json.loads(result.stdout)
|
||||
return data.get("chapters", [])
|
||||
|
||||
def get_metadata(file, fields) -> dict:
|
||||
def get_metadata_all(file: Path) -> dict[str, str]:
|
||||
metadata = {}
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"quiet",
|
||||
"-print_format",
|
||||
"json",
|
||||
"-show_format",
|
||||
str(file),
|
||||
],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
metadata = {
|
||||
str(k): str(v)
|
||||
for k, v in json.loads(result.stdout)
|
||||
.get("format", {})
|
||||
.get("tags", {})
|
||||
.items()
|
||||
}
|
||||
except (OSError, subprocess.SubprocessError, json.JSONDecodeError) as error:
|
||||
print(f"Error getting metadata for {file}: {error}")
|
||||
for field in metadata_fields:
|
||||
metadata.setdefault(field, "")
|
||||
return metadata
|
||||
|
||||
def get_metadata_fields(file, fields) -> dict:
|
||||
metadata = dict.fromkeys(fields, "")
|
||||
for field in fields:
|
||||
cmd = ["kid3-cli", "-c", f"get {field}", str(file)]
|
||||
cmd = ["kid3-cli", "-c", f'get "{field}"', str(file)]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
@@ -138,7 +170,7 @@ def write_metadata(file, fields, values,dry_run=False):
|
||||
"tag 2"
|
||||
]
|
||||
for field, value in zip(fields, values):
|
||||
cmd_base.extend( ["-c", f'set {field} "{value}"'])
|
||||
cmd_base.extend( ["-c", f'set "{field}" "{value}"'])
|
||||
cmd_base.extend(["-c", "save", str(file)])
|
||||
try:
|
||||
logger.trace(f"Writing metadata for {file}")
|
||||
@@ -152,7 +184,6 @@ def write_metadata(file, fields, values,dry_run=False):
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
logger.trace(f"KID3: {result.stdout}")
|
||||
except (OSError, subprocess.SubprocessError) as error:
|
||||
print(f"Error writing metadata for {file}: {error}")
|
||||
return
|
||||
@@ -171,10 +202,27 @@ def cleanup(text: str) -> str:
|
||||
|
||||
return text.strip()
|
||||
|
||||
def get_empty_keys(data: dict) -> str:
|
||||
empty_keys = [str(key) for key, value in data.items() if not value]
|
||||
|
||||
return f"Empty: {', '.join(empty_keys)}" if empty_keys else ""
|
||||
|
||||
def natural_key(path):
|
||||
return [
|
||||
int(part) if part.isdigit() else part.casefold()
|
||||
for part in re.split(r"(\d+)", path.stem)
|
||||
]
|
||||
|
||||
class Audiobook:
|
||||
def __init__(self, book_path: Path, library_root_path : Path) -> None:
|
||||
self.series = ""
|
||||
self.singleFile = False
|
||||
SEP = r"\s*[-_:,.]+\s*"
|
||||
self.series_patterns = [
|
||||
rf'^(\d+){SEP}(.+)$',
|
||||
rf'^(.+?){SEP}(\d+)$',
|
||||
rf'^(.+?)(?:{SEP}?Book\s+(\d+)|\s+Book\s+(\d+))?$',
|
||||
]
|
||||
self.fileConunt = 0
|
||||
self.size = 0
|
||||
self.firstFile = ""
|
||||
@@ -190,9 +238,14 @@ class Audiobook:
|
||||
self.library_root_path = library_root_path
|
||||
|
||||
def init_book(self) -> None:
|
||||
self.files = [
|
||||
file for pattern in AUDIO_SUFFIXES for file in self.bookPath.glob(pattern)
|
||||
]
|
||||
self.files = sorted(
|
||||
(
|
||||
file
|
||||
for pattern in AUDIO_SUFFIXES
|
||||
for file in self.bookPath.glob(pattern)
|
||||
),
|
||||
key=natural_key,
|
||||
)
|
||||
self.firstFile = self.get_first_file()
|
||||
self.size = sum(
|
||||
file.stat().st_size
|
||||
@@ -203,7 +256,7 @@ class Audiobook:
|
||||
self.singleFile = self.fileConunt == 1
|
||||
# Metadata
|
||||
self.load_metadata()
|
||||
logger.info(f"Found {self.bookPath.name}, {self.fileConunt} files")
|
||||
logger.info(f"Found {self.bookPath.name}, {self.fileConunt} files, {get_empty_keys(self.metadata)}")
|
||||
self.changed = False
|
||||
|
||||
def __repr__(self):
|
||||
@@ -241,11 +294,10 @@ class Audiobook:
|
||||
self.metadata["album artist"] = self.artist
|
||||
|
||||
def load_metadata(self):
|
||||
|
||||
self.metadata = get_metadata(self.get_first_file(), metadata_fields)
|
||||
self.metadata = get_metadata_all(self.get_first_file())
|
||||
self.write_fields_from_metadata()
|
||||
logger.trace(
|
||||
f"Load metadata {self.artist}, {self.name}, {self.composer}, {self.comment}, {self.genre}"
|
||||
f"Load metadata {self.metadata}"
|
||||
)
|
||||
|
||||
def fix_artist_from_comment(self):
|
||||
@@ -277,7 +329,7 @@ class Audiobook:
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if not m:
|
||||
logger.debug(f"No artist found in comment for {self.bookPath}")
|
||||
logger.trace(f"No artist found in comment for {self.bookPath}")
|
||||
return
|
||||
self.changed = True
|
||||
artist = re.sub(r"\s*\([^)]*\)", "", m.group(1).strip()).strip()
|
||||
@@ -288,6 +340,8 @@ class Audiobook:
|
||||
self.comment = cleanup(self.comment)
|
||||
if not self.artist:
|
||||
self.artist = artist
|
||||
self.changed = True
|
||||
|
||||
elif self.artist != artist:
|
||||
logger.warning(
|
||||
f"[Action Required] artist {artist} found not matching {self.artist} for {self.name}"
|
||||
@@ -296,15 +350,16 @@ class Audiobook:
|
||||
logger.debug(f"Artist {artist} matches file metadata for {self.name}")
|
||||
|
||||
def fix_artist_from_path(self):
|
||||
rel_path = self.bookPath.relative_to(self.library_root_path).parts
|
||||
|
||||
if len(rel_path) < 2:
|
||||
logger.trace(f"Artist not visible in folder structure: {rel_path}")
|
||||
return
|
||||
if not self.artist:
|
||||
rel_path = self.bookPath.relative_to(self.library_root_path).parts
|
||||
self.changed = True
|
||||
|
||||
if len(rel_path) < 2:
|
||||
logger.trace(f"Artist not visible in folder structure: {self.bookPath}")
|
||||
return
|
||||
self.artist = rel_path[0]
|
||||
|
||||
logger.info(f"[Updated] artist found in path for {self.bookPath}")
|
||||
logger.info(f"[Updated] artist found in path for {rel_path}")
|
||||
|
||||
def fix_composer_from_comment(self):
|
||||
m = re.search(
|
||||
@@ -317,68 +372,89 @@ class Audiobook:
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if not m:
|
||||
logger.debug(f"No composer found in comment for {self.name}")
|
||||
logger.trace(f"No composer found in comment for {self.name}")
|
||||
return
|
||||
self.changed = True
|
||||
self.comment = (self.comment[: m.start()] + self.comment[m.end() :]).strip(
|
||||
" .,;\n"
|
||||
)
|
||||
self.comment = cleanup(self.comment)
|
||||
if not self.composer:
|
||||
if not self.composer or len(self.composer) < 5:
|
||||
logger.info(f"[Updated] composer found in comment for {self.name}")
|
||||
self.composer = m.group(1).strip()
|
||||
|
||||
def fix_composer_from_path(self):
|
||||
if not self.composer:
|
||||
rel_path = self.bookPath.relative_to(self.library_root_path)
|
||||
m = re.search(
|
||||
r"(?:read\s+by:?|"
|
||||
r"spoken\s+by:?|"
|
||||
r"narrated\s+by:?|"
|
||||
r"gesprochen\s+von:?|"
|
||||
r"gelesen\s+von)\s+([^\n.,;]+)",
|
||||
rel_path,
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if not m:
|
||||
logger.debug(f"No composer found in path for {self.bookPath}")
|
||||
return
|
||||
rel_path = self.bookPath.relative_to(self.library_root_path)
|
||||
m = re.search(
|
||||
r"(?:read\s+by:?|"
|
||||
r"spoken\s+by:?|"
|
||||
r"narrated\s+by:?|"
|
||||
r"gesprochen\s+von:?|"
|
||||
r"gelesen\s+von)\s+([^\n.,;]+)",
|
||||
str(rel_path),
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if not m:
|
||||
logger.trace(f"No composer found in path for {self.bookPath}")
|
||||
return
|
||||
if not self.composer or len(self.composer) < 5:
|
||||
self.changed = True
|
||||
logger.info(f"[Updated] composer found in Path for {self.bookPath}")
|
||||
self.composer = m.group(1).strip()
|
||||
|
||||
def fix_name(self):
|
||||
if not self.name:
|
||||
self.name = self.bookPath.name
|
||||
name = self.bookPath.name
|
||||
for pattern in self.series_patterns:
|
||||
match = re.match(pattern, name, re.IGNORECASE)
|
||||
if match:
|
||||
if pattern.startswith(r'^(\d+)'):
|
||||
self.name = match.group(2).strip()
|
||||
else:
|
||||
self.name = match.group(1).strip()
|
||||
break
|
||||
else:
|
||||
self.name = name
|
||||
self.changed = True
|
||||
logger.info(f"[Updated] name found in path for {self.bookPath}")
|
||||
|
||||
def fix_series(self):
|
||||
if self.series:
|
||||
logger.trace(f"{self.name} already has series: {self.series}")
|
||||
rel_path = self.bookPath.relative_to(self.library_root_path).parts
|
||||
patterns = [
|
||||
r'^(\d+)\s*-\s*(.+)$', # 02 - Bookname
|
||||
r'^(.+?)\s*-\s*(\d+)$', # Bookname - 02
|
||||
r'^(.+?)\s*-\s*Book\s+(\d+)$', # Bookname - Book 2
|
||||
]
|
||||
|
||||
number=""
|
||||
if len(rel_path) <= 2:
|
||||
logger.trace(f"No series found for {self.name}")
|
||||
return
|
||||
for pattern in patterns:
|
||||
for i, pattern in enumerate(self.series_patterns):
|
||||
match = re.match(pattern, rel_path[2], re.IGNORECASE)
|
||||
if match:
|
||||
number = next(g for g in match.groups() if g.isdigit())
|
||||
break
|
||||
if not match:
|
||||
continue
|
||||
if i == 0:
|
||||
number = match.group(1)
|
||||
elif i == 1:
|
||||
number = match.group(2)
|
||||
else:
|
||||
number = match.group(2) or match.group(3)
|
||||
|
||||
break
|
||||
number_str = ""
|
||||
and_number =""
|
||||
if not number:
|
||||
logger.trace(f"No series number found for {self.name}")
|
||||
return
|
||||
self.series = rel_path[1] + "#" + number
|
||||
logger.info(f"[Updated] name found for {self.name}")
|
||||
logger.trace(f"No series number found for {self.name}")
|
||||
else:
|
||||
number_str = " #" + number
|
||||
and_number = " and number"
|
||||
self.series = rel_path[1] + number_str
|
||||
self.changed = True
|
||||
logger.info(f"[Updated] series{and_number} found for {self.name}")
|
||||
return
|
||||
|
||||
def fix_genre(self):
|
||||
self.genre = clean_genre(self.genre)
|
||||
self.changed = True
|
||||
|
||||
|
||||
def move(self, desiredPath: Path, desired_name: str, dry_run=False):
|
||||
logger.trace(f"Moving {self.bookPath} to {desiredPath}")
|
||||
@@ -455,9 +531,20 @@ class Audiobook:
|
||||
mf.write(";FFMETADATA1\n")
|
||||
|
||||
for key, value in self.metadata.items():
|
||||
mf.write(f"{ffmetadata_escape(key)}={ffmetadata_escape(value)}\n")
|
||||
if key == "title":
|
||||
mf.write(f"{ffmetadata_escape(key)}={ffmetadata_escape(self.metadata["album"])}\n")
|
||||
else:
|
||||
mf.write(f"{ffmetadata_escape(key)}={ffmetadata_escape(value)}\n")
|
||||
for info in file_info:
|
||||
media = info["media"]
|
||||
media_name = info["media"].name
|
||||
# Remove title
|
||||
media_name = media_name.replace(self.name, "")
|
||||
# Clean separators and whitespace
|
||||
media_name = re.sub(r"[\s._-]+", " ", media_name).strip()
|
||||
# Remove leading/trailing separators again if needed
|
||||
media_name = re.sub(r"^[\s._-]+|[\s._-]+$", "", media_name)
|
||||
info["media"].with_name(media_name)
|
||||
chapters = info["chapters"]
|
||||
duration_ms = info["duration_ms"]
|
||||
|
||||
@@ -486,7 +573,17 @@ class Audiobook:
|
||||
mf.write(f"title={ffmetadata_escape(media.stem)}\n")
|
||||
|
||||
current_ms += duration_ms
|
||||
|
||||
codec = subprocess.check_output(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v", "error",
|
||||
"-select_streams", "a:0",
|
||||
"-show_entries", "stream=codec_name",
|
||||
"-of", "default=noprint_wrappers=1:nokey=1",
|
||||
str(self.firstFile),
|
||||
],
|
||||
text=True,
|
||||
).strip()
|
||||
cmd_backup = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
@@ -552,14 +649,18 @@ class Audiobook:
|
||||
"attached_pic",
|
||||
str(output_file_tmp),
|
||||
]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd_default,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
cmd = cmd_default
|
||||
if codec == "mp3":
|
||||
cmd = cmd_backup
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.warning(e.stderr)
|
||||
logger.info(result.stdout)
|
||||
|
||||
if result.returncode == 234:
|
||||
@@ -614,11 +715,6 @@ class AudiobookLibrary:
|
||||
self.dry_run = dry_run
|
||||
logging.setLoggerClass(CustomLogger)
|
||||
logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment]
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s",
|
||||
handlers=[logging.FileHandler(logfilename), logging.StreamHandler()],
|
||||
)
|
||||
logger.info("Started")
|
||||
self.libraryPath: Path = Path(library_path)
|
||||
self.audiobooks: list[Audiobook] = []
|
||||
@@ -894,7 +990,6 @@ class AudiobookLibrary:
|
||||
futures = {
|
||||
executor.submit(
|
||||
audiobook.create_m4b_from_book,
|
||||
self.libraryPath,
|
||||
keep_originals,
|
||||
): audiobook
|
||||
for audiobook in jobs
|
||||
@@ -969,7 +1064,7 @@ def main():
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
setup_logging(args.log_level)
|
||||
setup_logging(args.log_level, "audiobook_library")
|
||||
library_dir = args.library.expanduser()
|
||||
library_dir = library_dir.resolve()
|
||||
if not library_dir.exists():
|
||||
@@ -993,11 +1088,12 @@ def main():
|
||||
library.fix_comments()
|
||||
library.fix_genres()
|
||||
|
||||
library.create_m4b_from_dir(
|
||||
name=args.name_filter,
|
||||
path=args.path_filter,
|
||||
keep_originals=args.keep_originals,
|
||||
)
|
||||
if args.create_m4b:
|
||||
library.create_m4b_from_dir(
|
||||
name=args.name_filter,
|
||||
path=args.path_filter,
|
||||
keep_originals=args.keep_originals,
|
||||
)
|
||||
|
||||
library.write_metadata()
|
||||
|
||||
|
||||
Reference in new issue
Block a user