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) + +