diff --git a/pyproject.toml b/pyproject.toml index 15dba81..78353d9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ classifiers = [ "Topic :: System :: Distributed Computing" ] keywords = ["Personal tools and recipes"] -dependencies = ["pydantic", "tinytag"] +dependencies = ["pydantic"] [project.urls] Repository = "https://git.jmsgrogan.com/jgrogan/recipes" diff --git a/src/jgutils/converters.py b/src/jgutils/converters.py index 6faf711..efdeeb8 100644 --- a/src/jgutils/converters.py +++ b/src/jgutils/converters.py @@ -6,14 +6,12 @@ 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 @@ -26,34 +24,32 @@ 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/__init__.py b/src/jgutils/filesystem/__init__.py index dce90e8..e69de29 100644 --- a/src/jgutils/filesystem/__init__.py +++ b/src/jgutils/filesystem/__init__.py @@ -1 +0,0 @@ -from .filesystem import * # NOQA diff --git a/src/jgutils/filesystem/filesystem.py b/src/jgutils/filesystem/filesystem.py index 538c6a4..a82bd0f 100644 --- a/src/jgutils/filesystem/filesystem.py +++ b/src/jgutils/filesystem/filesystem.py @@ -1,5 +1,6 @@ from pathlib import Path - -def get_files_recursive(search_path: Path, extension: str) -> list[Path]: +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 753d34e..1afc94a 100644 --- a/src/jgutils/main_cli.py +++ b/src/jgutils/main_cli.py @@ -1,29 +1,20 @@ import argparse -import logging -from pathlib import Path +import logger 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 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() subparsers = parser.add_subparsers(required=True) @@ -38,34 +29,13 @@ 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_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_convert_parser.add_argument( + "--output_dir", + type=Path, default=Path(), + help="Directory for converted files" ) - music_refresh_parser.set_defaults(func=cli_music_refresh) + music_convert_parser.set_defaults(func=cli_replace_in_files) logging.basicConfig() logging.getLogger().setLevel(logging.INFO) @@ -73,6 +43,5 @@ def main_cli(): args = parser.parse_args() args.func(args) - if __name__ == "__main__": main_cli() diff --git a/src/jgutils/music/__init__.py b/src/jgutils/music/__init__.py deleted file mode 100644 index eec2a70..0000000 --- a/src/jgutils/music/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .music import * # NOQA diff --git a/src/jgutils/music/convert.py b/src/jgutils/music/convert.py new file mode 100644 index 0000000..74ae8a5 --- /dev/null +++ b/src/jgutils/music/convert.py @@ -0,0 +1,29 @@ +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 deleted file mode 100644 index d1077dd..0000000 --- a/src/jgutils/music/music.py +++ /dev/null @@ -1,111 +0,0 @@ -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) - - diff --git a/src/jgutils/tasks/__init__.py b/src/jgutils/tasks/__init__.py index 8e6fcd9..6804af1 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 66350cb..f2fd8b1 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: + with Pool(10) as p: p.map(_run_task, tasks) +