lucas/Server-scripts
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(serverlibrary): move suffixes to library

Browse Source 
move audio/video suffixes to serverlibrary
remove from audiobooks_library, ffmpeg_handling
This commit is contained in:
lucas authored and Lucas Vivo committed 2026-10-29 09:02:30 +01:00
1 parent 18e1fc9325
commit ad796c174e
5 files changed
+103 -51
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

audiobook_metadata/metadata_audiobooks_library.py
+3 -4
View File
@@ -17,7 +17,6 @@ logging.setLoggerClass(CustomLogger)
logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment]
metadata_fields = ["artist", "album", "composer", "comment", "genre", "grouping", "album artist", "description"]
audio_extensions = {"*.m4b", "*.m4a", "*.mp3", "*.flac", "*.ogg", "*.opus"}
def ffmpeg_escape(path: Path) -> str:
return str(path.resolve()).replace("'", r"'\''")
@@ -192,12 +191,12 @@ class Audiobook:
def init_book(self) -> None:
self.files = [
file for pattern in audio_extensions for file in self.bookPath.glob(pattern)
file for pattern in AUDIO_SUFFIXES for file in self.bookPath.glob(pattern)
]
self.firstFile = self.get_first_file()
self.size = sum(
file.stat().st_size
for pattern in audio_extensions
for pattern in AUDIO_SUFFIXES
for file in self.bookPath.glob(pattern)
)
self.fileConunt = len(self.files)
@@ -815,7 +814,7 @@ class AudiobookLibrary:
"""
media_files = []
for ext in audio_extensions:
for ext in AUDIO_SUFFIXES:
media_files.extend(list(self.libraryPath.rglob(ext)))
dirs = {file.parent for file in media_files}
for directory in dirs:
audiobook_metadata/metadata_rank_from_list.py
-9
View File
@@ -6,15 +6,6 @@ import subprocess
from pathlib import Path
from difflib import SequenceMatcher
AUDIO_EXTENSIONS = {
".mp3",
".m4b",
".m4a",
}
RED = "\033[31m"
GREEN = "\033[32m"
RESET = "\033[0m"
USE_KID3 = True
def normalize(text: str) -> str:
return re.sub(r"[^a-z0-9]", "", text.lower())
media_editing/ffmpeg_handling.py
-3
View File
@@ -23,9 +23,6 @@ from serverlibrary import *
logging.setLoggerClass(CustomLogger)
logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment]
VIDEO_EXTS = {".mp4", ".mkv", ".avi", ".mov", ".webm"}
AUDIO_EXTS = {".mp3", ".m4a", ".m4b", ".opus"}
# ----------------------------
# OUTPUT ENFORCEMENT (clean + centralized)
# ----------------------------
serverlibrary/src/serverlibrary/__init__.py
+8 -5
View File
@@ -1,8 +1,8 @@
from .core import (
TRACE,
LOG_LEVELS,
TRACE,
CustomLogger,
setup_logging,
AUDIO_SUFFIXES,
blue,
combined_score,
get_audio_tracks,
@@ -15,18 +15,20 @@ from .core import (
score_match_fuzzy,
score_match_jaccard,
score_match_recall_jaccard,
setup_logging,
tmdb_get,
tvdb_get,
VIDEO_SUFFIXES,
word_similarity,
words,
yellow,
)
__all__ = [
"TRACE",
"LOG_LEVELS",
"TRACE",
"CustomLogger",
"setup_logging",
"AUDIO_SUFFIXES",
"blue",
"combined_score",
"get_audio_tracks",
@@ -39,10 +41,11 @@ __all__ = [
"score_match_fuzzy",
"score_match_jaccard",
"score_match_recall_jaccard",
"setup_logging",
"tmdb_get",
"tvdb_get",
"VIDEO_SUFFIXES",
"word_similarity",
"words",
"yellow",
]
serverlibrary/src/serverlibrary/core.py
+92 -30
View File
@@ -26,6 +26,7 @@ reset = "\033[0m"
class CustomLogger(logging.Logger):
def trace(self, msg: str, *args, **kwargs) -> None:
if self.isEnabledFor(TRACE):
kwargs.setdefault("stacklevel", 2)
self._log(TRACE, msg, args, **kwargs)
@@ -38,12 +39,13 @@ LOG_LEVELS = {
"CRITICAL": logging.CRITICAL,
}
def setup_logging(log_level: str = "INFO", logfile: str | None = None) -> None:
logging.addLevelName(TRACE, "TRACE")
file_formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
console_formatter = colorlog.ColoredFormatter(
"%(log_color)s%(asctime)s %(levelname)-8s%(reset)s %(name)s %(message)s",
"%(log_color)s%(asctime)s %(levelname)-8s%(reset)s %(funcName)s %(message)s",
log_colors={
"TRACE": "cyan",
"DEBUG": "blue",
@@ -78,9 +80,19 @@ def setup_logging(log_level: str = "INFO", logfile: str | None = None) -> None:
root_logger.info(f"Set logger to {log_level}, Logfile: {logfile_data}")
# -----------------------------
# Media Files
# -----------------------------
AUDIO_SUFFIXES = {"*.m4b", "*.m4a", "*.mp3", "*.flac", "*.ogg", "*.opus"}
VIDEO_SUFFIXES = {".mp4", ".mkv", ".avi", ".mov", ".webm"}
# -----------------------------
# Helpers
# -----------------------------
def safe_delete(path: Path):
if path.is_dir():
shutil.rmtree(path)
@@ -89,52 +101,84 @@ def safe_delete(path: Path):
def safe_rename(
src: Path,
dst: Path,
dry: bool = False,
src_filepath: Path,
dst_filename: Path,
dry_run: bool = False,
confirm: bool = False,
silent=False,
overwrite=False,
logger=None,
) -> tuple[str, bool]:
"""
Safely rename a file or directory without overwriting existing files.
Automatically appends _1, _2, ... if needed.
Safely rename or move a file or directory to a new destination.
If the destination already exists, a unique name is automatically
generated by appending a numeric suffix (_1, _2, ...) unless
overwrite is enabled. Supports dry-run mode, optional user
confirmation, and silent operation.
Args:
src_filepath: Source file or directory to rename.
dst_filename: Target path and filename.
dry_run: If True, only report the action without performing it.
confirm: If True, prompt for confirmation before renaming.
silent: If True, suppress informational output.
overwrite: If True, replace an existing destination.
logger: Logger handler
Returns:
A tuple containing:
- The final destination path as a string.
- True if a rename was performed successfully, otherwise False.
Raises:
FileNotFoundError: If the source path does not exist.
OSError: If the rename operation fails.
"""
src = Path(src)
dst = Path(dst)
src_filepath = Path(src_filepath).expanduser().resolve()
dst_filename = Path(dst_filename).expanduser()
if dst_filename.parent == Path("."):
dst_filename = src_filepath.parent / dst_filename
dst_filename = dst_filename.resolve()
if not src_filepath.exists():
raise FileNotFoundError(f"File {src_filepath} not found")
# nothing to do
if src.resolve() == dst.resolve():
if src_filepath == dst_filename:
return "", False
directory = dst.parent
base = dst.stem
ext = dst.suffix
# directory = dst_filename.parent
# base = dst_filename.stem
# ext = dst_filename.suffix
final_dst = dst
counter = 1
# final_dst = dst_filename
# counter = 1
# find free name
while final_dst.exists():
final_dst = directory / f"{base}_{counter}{ext}"
counter += 1
ret = f"{src} → {dst}"
# # find free name
# while final_dst.exists():
# final_dst = directory / f"{base}_{counter}{ext}"
# counter += 1
ret = f"{src_filepath} → {dst_filename}"
if not silent:
print(ret)
if dry:
if logger is not None:
logger.info(ret)
else:
print(ret)
if dry_run:
return ret, True
if confirm:
input("Press Enter to confirm...")
dst.parent.mkdir(parents=True, exist_ok=True)
dst_filename.parent.mkdir(parents=True, exist_ok=True)
if dst.exists() and not overwrite:
print("File already exists...")
if dst_filename.exists() and not overwrite:
if logger is not None:
logger.warning("File already exists...")
else:
print(ret)
return "", False
src.rename(dst)
src_filepath.rename(dst_filename)
return ret, True
@@ -177,10 +221,19 @@ def get_subtitle_tracks(source):
)
if not tracks:
tracks.append({"number": 0, "description": "", "lang": "", "language": "", "primary": False})
tracks.append(
{
"number": 0,
"description": "",
"lang": "",
"language": "",
"primary": False,
}
)
return tracks
def get_audio_tracks(
source: str, title: int | None = None, handbrake: bool = False
) -> list[dict]:
@@ -286,10 +339,19 @@ def get_audio_tracks(
)
if not tracks:
tracks.append({"number": 0, "description": "","lang":"","language":"", "primary": False})
tracks.append(
{
"number": 0,
"description": "",
"lang": "",
"language": "",
"primary": False,
}
)
return tracks
def words(text):
return set(re.findall(r"[^\W_]+", text.lower()))
@@ -360,7 +422,7 @@ def tmdb_get(endpoint, params=None):
BASE_URL = "https://api.themoviedb.org/3"
url = BASE_URL + endpoint
print(f"Set of Tmdb query: {url}")
# print_or_log(f"Set of Tmdb query: {url}")
response = requests.get(
url,
@@ -378,7 +440,7 @@ def tvdb_get(endpoint, params=None):
"""Perform a GET request against TVDB."""
BASE_URL = "https://api4.thetvdb.com/v4"
url = BASE_URL + endpoint
print(f"Set of TVDB query: {url}")
# print_or_log(f"Set of TVDB query: {url}")
response = requests.get(
url,