Change to python project format

This commit is contained in:
James Grogan 2024-09-29 18:23:09 +01:00
parent b20465a3d7
commit 61599cb76c
27 changed files with 268 additions and 123 deletions

3
.gitignore vendored
View file

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

100
pyproject.toml Normal file
View file

@ -0,0 +1,100 @@
[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
"""

55
src/jgutils/converters.py Normal file
View file

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

View file

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

47
src/jgutils/main_cli.py Normal file
View file

@ -0,0 +1,47 @@
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()

View file

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

0
src/jgutils/py.typed Normal file
View file

View file

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

View file

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

View file

@ -1,122 +0,0 @@
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 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)
class ConversionConfig(NamedTuple):
input_dir: Path
output_dir: Path
output_ext: str
input_ext: str
class Task(NamedTuple):
cmd: str
output_tmp: Path
output_path: Path
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 get_files_recursive(search_path: Path,
extension: str) -> list[Path]:
return list(search_path.rglob(f"*.{extension}"))
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 build_conversion_tasks(input_files: list[Path],
output_files: list[Path],
config: ConversionConfig) -> list[Tasks]:
tasks = []
for input_path, output_path in zip(input_files, output_files):
cmd = get_conversion_cmd(input_path, config.output_dir, config.output_ext)
tasks.append((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):
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_ext)
output_files = get_uncoverted_files(input_files, config)
tasks = build_conversion_tasks(input_files, output_files, config)
with Pool(10) as p:
p.map(run_task, 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())