.gitignore

@@ -1,4 +1,3 @@
 .venv
 *.pyc
 __pycache__/
-*.egg-info/

pyproject.toml

@@ -1,100 +0,0 @@
-[project]
-name = "jgutils"
-version = "0.0.0"
-authors = [
-  { name="James Grogan", email="james@jmsgrogan.com" },
-]
-description = "A collection of tools for running my PC."
-readme = "README.md"
-requires-python = ">=3.8"
-classifiers = [
-    "Development Status :: 3 - Alpha",
-    "Programming Language :: Python :: 3",
-    "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)",
-    "Operating System :: OS Independent",
-    "Topic :: System :: Distributed Computing"
-]
-keywords = ["Personal tools and recipes"]
-dependencies = ["pydantic"]
-
-[project.urls]
-Repository = "https://git.jmsgrogan.com/jgrogan/recipes"
-Homepage = "https://jmsgrogan.com"
-
-[build-system]
-requires = ["setuptools>=61.0", "wheel"]
-build-backend = "setuptools.build_meta"
-
-[project.optional-dependencies]
-test = [
-    "pytest",
-    "pytest-cov",
-    "pytest-sugar",
-    "black",
-    "mypy",
-    "flake8",
-    "pylint"
-]
-
-[project.scripts]
-jgutils = "jgutils.main_cli:main_cli"
-
-[tool.setuptools.package-data]
-"jgutils" = ["py.typed"]
-
-[tool.setuptools.packages.find]
-where = ["src"]
-
-[tool.mypy]
-ignore_missing_imports = true
-
-[tool.pytest.ini_options]
-testpaths = ["test",]
-log_cli = 1
-log_cli_level = "debug"
-addopts = "--cov=jgutils --cov-report term --cov-report xml:coverage.xml --cov-report html"
-
-[tool.tox]
-legacy_tox_ini = """
-[tox]
-requires =
-    tox>=4
-env_list = lint, type, style, py{311}, docs
-skip_missing_interpreters = true
-
-[testenv]
-description = run unit tests
-deps =
-    pytest>=7
-    pytest-cov
-    pytest-sugar
-commands =
-    pytest {posargs:test}
-
-[testenv:lint]
-description = run linters
-skip_install = true
-deps =
-    black
-commands = black {posargs:src}
-
-[testenv:style]
-description = run style check
-skip_install = true
-deps =
-    flake8
-commands = flake8 {posargs:src}
-
-[testenv:type]
-description = run type checks
-deps =
-    mypy>=0.991
-    .[types]
-commands =
-    mypy {posargs:src test}
-
-[testenv:docs]
-changedir = docs
-deps = sphinx
-commands = sphinx-build -W -b html -d {envtmpdir}/doctrees . {envtmpdir}/html
-"""

src/jgutils/converters.py

@@ -1,55 +0,0 @@
-from pathlib import Path
-from typing import NamedTuple, Callable
-import logging
-
-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
-    path to the output directory in the config and changed its extension.
-    """
-
-    relative_path = input_path.relative_to(config.input_dir)
-    output_filename = str(input_path.stem) + f".{config.output_ext}"
-    output_filename.replace("'", "")
-    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]:
-    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]:
-    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)
-    output_files = get_uncoverted_files(input_files, config)
-
-    tasks = _build_conversion_tasks(input_files,
-                                    output_files,
-                                    config,
-                                    conversion_func)
-    run_tasks(tasks)

src/jgutils/filesystem/filesystem.py

@@ -1,6 +0,0 @@
-from pathlib import Path
-
-def get_files_recursive(search_path: Path,
-                        extension: str) -> list[Path]:
-    return list(search_path.rglob(f"*.{extension}"))
-

src/jgutils/main_cli.py

@@ -1,47 +0,0 @@
-import argparse
-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"
-    )
-    music.convert(config)
-
-def main_cli():
-    parser = argparse.ArgumentParser()
-    subparsers = parser.add_subparsers(required=True)
-
-    music_parser = subparsers.add_parser("music")
-    music_subparsers = music_parser.add_subparsers(required=True)
-
-    music_convert_parser = music_subparsers.add_parser("convert")
-    music_convert_parser.add_argument(
-        "--input_dir",
-        type=Path,
-        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)
-
-    logging.basicConfig()
-    logging.getLogger().setLevel(logging.INFO)
-
-    args = parser.parse_args()
-    args.func(args)
-
-if __name__ == "__main__":
-    main_cli()

src/jgutils/music/convert.py

@@ -1,29 +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)
-
-

src/jgutils/tasks/__init__.py

@@ -1 +0,0 @@
-from .tasks import * # NOQA

src/jgutils/tasks/tasks.py

@@ -1,28 +0,0 @@
-import os
-from pathlib import Path
-import logging
-import shutil
-import subprocess
-from typing import NamedTuple
-from multiprocessing import Pool
-
-logger = logging.getLogger(__name__)
-
-
-class Task(NamedTuple):
-    cmd: str
-    output_tmp: Path
-    output_path: Path
-
-
-def _run_task(args):
-    task = args[0]
-
-    subprocess.run(cmd, shell=True)
-    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)
-

src/music/convert.py

@@ -0,0 +1,112 @@
+import argparse
+import os
+import logging
+import shutil
+from pathlib import Path
+from typing import NamedTuple
+import subprocess
+import uuid
+from multiprocessing import Pool
+
+logger = logging.getLogger(__name__)
+
+
+def convert_audio_file(args):
+    cmd, output_tmp, output_path = args
+
+    subprocess.run(cmd, shell=True)
+    os.makedirs(output_path.parent, exist_ok=True)
+    shutil.move(output_tmp, output_path)
+
+
+class ConversionConfig(NamedTuple):
+    input_dir: Path
+    output_dir: Path
+    output_fmt: str
+    input_fmt: str
+
+
+def _get_converted_path(input_path: Path, config: ConversionConfig):
+    """
+    Return the path you would obtain if you moved the file in the input
+    path to the output directory in the config and changed its extension.
+    """
+
+    relative_path = input_path.relative_to(config.input_dir)
+    output_filename = str(input_path.stem) + f".{config.output_fmt}"
+    output_filename.replace("'", "")
+    output_path = config.output_dir / relative_path.parent / output_filename
+    return output_path
+
+
+def get_files_recursive(search_path: Path, extension: str):
+    return list(search_path.rglob(f"*.{extension}"))
+
+
+def convert(config: ConversionConfig):
+
+    logger.info("Converting files in %s", config.input_dir)
+    logger.info("Writing output to: %s", config.output_dir)
+
+    os.makedirs(config.output_dir, exist_ok=True)
+
+    input_files = get_files_recursive(
+        config.input_dir.resolve(), config.input_fmt
+    )
+    output_files = []
+    for input_file in input_files:
+        candidate_output = _get_converted_path(input_file)
+        if not candidate_output.exists():
+            output_files.append(candidate_output)
+    print(output_files)
+
+    tasks = []
+    for idx, (input_path, output_path) in enumerate(zip(input_files, output_files)):
+
+        logger.info("Converting file %d of %d", idx, len(input_files))
+
+        output_tmp = self.output_dir / (str(uuid.uuid4()) + f".{self.output_fmt}")
+        cmd = f"ffmpeg -i '{input_path}' -ab 320k -map_metadata 0 -id3v2_version 3 '{output_tmp}'"
+        tasks.append((cmd, output_tmp, output_path))
+
+    with Pool(10) as p:
+        p.map(convert_audio_file, tasks)
+
+
+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)
+
+
+if __name__ == "__main__":
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "--input_dir",
+        type=Path,
+        default=Path(),
+        help="Directory with input files for conversion.",
+    )
+
+    parser.add_argument(
+        "--output_dir", type=Path, default=Path(), help="Directory for converted files"
+    )
+
+    args = parser.parse_args()
+
+    logging.basicConfig()
+    logging.getLogger().setLevel(logging.INFO)
+
+    converter = AudioFileConverter(args.input_dir.resolve(), args.output_dir.resolve())
+    converter.convert()
+    # move(args.input_dir.resolve(), args.output_dir.resolve())