From 29719cba6ea58067edf77392c5443459d3a0f756 Mon Sep 17 00:00:00 2001 From: jgrogan Date: Sun, 29 Sep 2024 18:26:57 +0100 Subject: [PATCH 1/2] Apply come formatting --- src/jgutils/converters.py | 28 ++++++++++++++++------------ src/jgutils/filesystem/filesystem.py | 5 ++--- src/jgutils/main_cli.py | 12 +++++------- src/jgutils/music/convert.py | 7 ++----- src/jgutils/tasks/__init__.py | 2 +- src/jgutils/tasks/tasks.py | 6 +++--- 6 files changed, 29 insertions(+), 31 deletions(-) diff --git a/src/jgutils/converters.py b/src/jgutils/converters.py index efdeeb8..6faf711 100644 --- a/src/jgutils/converters.py +++ b/src/jgutils/converters.py @@ -6,12 +6,14 @@ from .tasks import Task logger = logging.getLogger(__name__) + class ConversionConfig(NamedTuple): input_dir: Path output_dir: Path output_ext: str input_ext: str + def _get_converted_path(input_path: Path, config: ConversionConfig): """ Return the path you would obtain if you moved the file in the input @@ -24,32 +26,34 @@ def _get_converted_path(input_path: Path, config: ConversionConfig): output_path = config.output_dir / relative_path.parent / output_filename return output_path -def _build_conversion_tasks(input_files: list[Path], - output_files: list[Path], - config: ConversionConfig, - conversion_func: Callable) -> list[Tasks]: + +def _build_conversion_tasks( + input_files: list[Path], + output_files: list[Path], + config: ConversionConfig, + conversion_func: Callable, +) -> list[Tasks]: tasks = [] for input_path, output_path in zip(input_files, output_files): cmd = conversion_func(input_path, config.output_dir, config.output_ext) tasks.append(Task(cmd, output_tmp, output_path)) return tasks -def get_unconverted_files(input_files: list[Path], - config: ConversionConfig) -> list[Path]: + +def get_unconverted_files( + input_files: list[Path], config: ConversionConfig +) -> list[Path]: output_files = [_get_converted_path(f, config) for f in input_files] return [f for f in output_files if not f.exists()] + def convert(config: ConversionConfig, conversion_func: Callable): logger.info("Converting files in %s", config.input_dir) logger.info("Writing output to: %s", config.output_dir) - input_files = get_files_recursive( - config.input_dir.resolve(), config.input_ext) + input_files = get_files_recursive(config.input_dir.resolve(), config.input_ext) output_files = get_uncoverted_files(input_files, config) - tasks = _build_conversion_tasks(input_files, - output_files, - config, - conversion_func) + tasks = _build_conversion_tasks(input_files, output_files, config, conversion_func) run_tasks(tasks) diff --git a/src/jgutils/filesystem/filesystem.py b/src/jgutils/filesystem/filesystem.py index a82bd0f..538c6a4 100644 --- a/src/jgutils/filesystem/filesystem.py +++ b/src/jgutils/filesystem/filesystem.py @@ -1,6 +1,5 @@ from pathlib import Path -def get_files_recursive(search_path: Path, - extension: str) -> list[Path]: - return list(search_path.rglob(f"*.{extension}")) +def get_files_recursive(search_path: Path, extension: str) -> list[Path]: + return list(search_path.rglob(f"*.{extension}")) diff --git a/src/jgutils/main_cli.py b/src/jgutils/main_cli.py index 1afc94a..45d3947 100644 --- a/src/jgutils/main_cli.py +++ b/src/jgutils/main_cli.py @@ -5,16 +5,15 @@ from jgutils import music logger = logging.getLogger(__name__) + def cli_music_convert(args): config = music.CovertionConfig( - args.input_dir.resolve(), - args.output_dir.resolve(), - "mp3", - "flac" + args.input_dir.resolve(), args.output_dir.resolve(), "mp3", "flac" ) music.convert(config) + def main_cli(): parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(required=True) @@ -31,9 +30,7 @@ def main_cli(): ) music_convert_parser.add_argument( - "--output_dir", - type=Path, default=Path(), - help="Directory for converted files" + "--output_dir", type=Path, default=Path(), help="Directory for converted files" ) music_convert_parser.set_defaults(func=cli_replace_in_files) @@ -43,5 +40,6 @@ def main_cli(): args = parser.parse_args() args.func(args) + if __name__ == "__main__": main_cli() diff --git a/src/jgutils/music/convert.py b/src/jgutils/music/convert.py index 74ae8a5..906ad93 100644 --- a/src/jgutils/music/convert.py +++ b/src/jgutils/music/convert.py @@ -6,9 +6,8 @@ import uuid logger = logging.getLogger(__name__) -def ffmpeg_convert(input_path: Path, - output_dir: Path, - output_ext: str) -> str: + +def ffmpeg_convert(input_path: Path, output_dir: Path, output_ext: str) -> str: output_tmp = output_dir / (str(uuid.uuid4()) + f".{output_ext}") cmd = f"ffmpeg -i '{input_path}' -ab 320k -map_metadata 0 -id3v2_version 3 '{output_tmp}'" @@ -25,5 +24,3 @@ def move(input_dir: Path, output_dir: Path): output_path = output_dir / relative_path os.makedirs(output_path.parent, exist_ok=True) shutil.move(path, output_path) - - diff --git a/src/jgutils/tasks/__init__.py b/src/jgutils/tasks/__init__.py index 6804af1..8e6fcd9 100644 --- a/src/jgutils/tasks/__init__.py +++ b/src/jgutils/tasks/__init__.py @@ -1 +1 @@ -from .tasks import * # NOQA +from .tasks import * # NOQA diff --git a/src/jgutils/tasks/tasks.py b/src/jgutils/tasks/tasks.py index f2fd8b1..66350cb 100644 --- a/src/jgutils/tasks/tasks.py +++ b/src/jgutils/tasks/tasks.py @@ -22,7 +22,7 @@ def _run_task(args): os.makedirs(task.output_path.parent, exist_ok=True) shutil.move(task.output_tmp, task.output_path) -def run_tasks(tasks, pool_size: 10): - with Pool(10) as p: - p.map(_run_task, tasks) +def run_tasks(tasks, pool_size: 10): + with Pool(10) as p: + p.map(_run_task, tasks) From 8974849b8b3513f88005dd1058352f992154bd58 Mon Sep 17 00:00:00 2001 From: jgrogan Date: Sun, 29 Sep 2024 19:42:35 +0100 Subject: [PATCH 2/2] Add music tagging --- pyproject.toml | 2 +- src/jgutils/filesystem/__init__.py | 1 + src/jgutils/main_cli.py | 39 +++++++++- src/jgutils/music/__init__.py | 1 + src/jgutils/music/convert.py | 26 ------- src/jgutils/music/music.py | 111 +++++++++++++++++++++++++++++ 6 files changed, 150 insertions(+), 30 deletions(-) create mode 100644 src/jgutils/music/__init__.py delete mode 100644 src/jgutils/music/convert.py create mode 100644 src/jgutils/music/music.py diff --git a/pyproject.toml b/pyproject.toml index 78353d9..15dba81 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ classifiers = [ "Topic :: System :: Distributed Computing" ] keywords = ["Personal tools and recipes"] -dependencies = ["pydantic"] +dependencies = ["pydantic", "tinytag"] [project.urls] Repository = "https://git.jmsgrogan.com/jgrogan/recipes" diff --git a/src/jgutils/filesystem/__init__.py b/src/jgutils/filesystem/__init__.py index e69de29..dce90e8 100644 --- a/src/jgutils/filesystem/__init__.py +++ b/src/jgutils/filesystem/__init__.py @@ -0,0 +1 @@ +from .filesystem import * # NOQA diff --git a/src/jgutils/main_cli.py b/src/jgutils/main_cli.py index 45d3947..753d34e 100644 --- a/src/jgutils/main_cli.py +++ b/src/jgutils/main_cli.py @@ -1,5 +1,6 @@ import argparse -import logger +import logging +from pathlib import Path from jgutils import music @@ -13,6 +14,15 @@ def cli_music_convert(args): ) music.convert(config) +def cli_music_metadata(args): + + collection = music.get_metadata(args.input_dir.resolve(), "flac") + with open(args.output_path.resolve(), 'w') as f: + f.write(collection.model_dump_json(indent=4)) + +def cli_music_refresh(args): + + collection = music.refresh(args.input_dir.resolve()) def main_cli(): parser = argparse.ArgumentParser() @@ -28,11 +38,34 @@ def main_cli(): default=Path(), help="Directory with input files for conversion.", ) - music_convert_parser.add_argument( "--output_dir", type=Path, default=Path(), help="Directory for converted files" ) - music_convert_parser.set_defaults(func=cli_replace_in_files) + music_convert_parser.set_defaults(func=cli_music_convert) + + music_md_parser = music_subparsers.add_parser("metadata") + music_md_parser.add_argument( + "--input_dir", + type=Path, + default=Path(), + help="Directory with input music files.", + ) + music_md_parser.add_argument( + "--output_path", + type=Path, + default=Path() / "music_collection.json", + help="Path to save collection to.", + ) + music_md_parser.set_defaults(func=cli_music_metadata) + + music_refresh_parser = music_subparsers.add_parser("refresh") + music_refresh_parser.add_argument( + "--input_dir", + type=Path, + default=Path(), + help="Directory with input music files.", + ) + music_refresh_parser.set_defaults(func=cli_music_refresh) logging.basicConfig() logging.getLogger().setLevel(logging.INFO) diff --git a/src/jgutils/music/__init__.py b/src/jgutils/music/__init__.py new file mode 100644 index 0000000..eec2a70 --- /dev/null +++ b/src/jgutils/music/__init__.py @@ -0,0 +1 @@ +from .music import * # NOQA diff --git a/src/jgutils/music/convert.py b/src/jgutils/music/convert.py deleted file mode 100644 index 906ad93..0000000 --- a/src/jgutils/music/convert.py +++ /dev/null @@ -1,26 +0,0 @@ -import os -import logging -import shutil -from pathlib import Path -import uuid - -logger = logging.getLogger(__name__) - - -def ffmpeg_convert(input_path: Path, output_dir: Path, output_ext: str) -> str: - output_tmp = output_dir / (str(uuid.uuid4()) + f".{output_ext}") - cmd = f"ffmpeg -i '{input_path}' -ab 320k -map_metadata 0 -id3v2_version 3 '{output_tmp}'" - - -def move(input_dir: Path, output_dir: Path): - - logger.info("Moving files in %s to %s", input_dir, output_dir) - os.makedirs(output_dir, exist_ok=True) - - mp3_files = list(input_dir.resolve().rglob("*.mp3")) - for idx, path in enumerate(mp3_files): - logger.info("Moving file %d of %d", idx, len(mp3_files)) - relative_path = path.relative_to(input_dir) - output_path = output_dir / relative_path - os.makedirs(output_path.parent, exist_ok=True) - shutil.move(path, output_path) diff --git a/src/jgutils/music/music.py b/src/jgutils/music/music.py new file mode 100644 index 0000000..d1077dd --- /dev/null +++ b/src/jgutils/music/music.py @@ -0,0 +1,111 @@ +import os +import logging +import shutil +from pathlib import Path +import uuid + +from tinytag import TinyTag +from pydantic import BaseModel + +from jgutils.filesystem import get_files_recursive + +logger = logging.getLogger(__name__) + + +class Song(BaseModel): + + title: str + identifier: str + formats: list[Path] = [] + +class Album(BaseModel): + + title: str + songs: list[Song] = [] + +class Artist(BaseModel): + + name: str + albums: list[Album] = [] + songs: list[Song] = [] + + def get_album(self, title: str) -> Album | None: + for album in self.albums: + if album.title == title: + return album + return None + +class MusicCollection(BaseModel): + + artists: list[Artist] = [] + + def get_artist(self, name:str) -> Artist | None: + for artist in self.artists: + if artist.name == name: + return artist + return None + + +def ffmpeg_convert(input_path: Path, output_dir: Path, output_ext: str) -> str: + output_tmp = output_dir / (str(uuid.uuid4()) + f".{output_ext}") + cmd = f"ffmpeg -i '{input_path}' -ab 320k -map_metadata 0 -id3v2_version 3 '{output_tmp}'" + + +def move(input_dir: Path, output_dir: Path): + + logger.info("Moving files in %s to %s", input_dir, output_dir) + os.makedirs(output_dir, exist_ok=True) + + mp3_files = list(input_dir.resolve().rglob("*.mp3")) + for idx, path in enumerate(mp3_files): + logger.info("Moving file %d of %d", idx, len(mp3_files)) + relative_path = path.relative_to(input_dir) + output_path = output_dir / relative_path + os.makedirs(output_path.parent, exist_ok=True) + shutil.move(path, output_path) + +def get_metadata(input_dir: Path, extension: str) -> MusicCollection: + + files = get_files_recursive(input_dir, extension) + + collection = MusicCollection() + + for eachFile in files: + tag = TinyTag.get(eachFile) + if not tag.title: + logger.warn("Found tag with no title, skipping: %s", tag) + continue + + artist = collection.get_artist(tag.artist) + if not artist: + artist = Artist(name=tag.artist) + collection.artists.append(artist) + + if tag.album: + album = artist.get_album(tag.album) + if not album: + album = Album(title=tag.album) + artist.albums.append(album) + + song = Song(title=tag.title, identifier=str(uuid.uuid4())) + song.formats.append(eachFile.relative_to(input_dir)) + if tag.album: + album.songs.append(song) + else: + artist.songs.append(song) + + return collection + +def refresh(input_dir: Path): + + files = get_files_recursive(input_dir, "flac") + + for eachFile in files: + tag = TinyTag.get(eachFile) + if not tag.title: + logger.warn("Found tag with no title, skipping: %s", tag) + continue + + os.make_dirs(input_dir / tag.artist, exist_ok=True) + +