stuff
This commit is contained in:
parent
b7e93fc970
commit
ebf2035c54
30 changed files with 305 additions and 462 deletions
|
|
@ -1,3 +0,0 @@
|
|||
# musicomp
|
||||
|
||||
A music compression tool.
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
{
|
||||
lib,
|
||||
python3Packages,
|
||||
opusTools,
|
||||
}:
|
||||
python3Packages.buildPythonApplication {
|
||||
pname = "musicomp";
|
||||
version = "0.1.0";
|
||||
|
||||
src = ./.;
|
||||
|
||||
pyproject = true;
|
||||
|
||||
doCheck = false;
|
||||
|
||||
build-system = [python3Packages.hatchling];
|
||||
|
||||
makeWrapperArgs = [
|
||||
"--prefix"
|
||||
"PATH"
|
||||
":"
|
||||
(lib.makeBinPath [opusTools])
|
||||
];
|
||||
|
||||
meta.mainProgram = "musicomp";
|
||||
}
|
||||
|
|
@ -1,17 +0,0 @@
|
|||
[project]
|
||||
name = "musicomp"
|
||||
version = "0.1.0"
|
||||
description = "Music compression tool"
|
||||
authors = [{ name = "Lukas Wurzinger", email = "lukas@wrz.one" }]
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project.scripts]
|
||||
musicomp = "musicomp.cli:main"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/musicomp"]
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
from .cli import main
|
||||
|
||||
main()
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
from os import PathLike
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def clean(dst: str | PathLike[str]) -> None:
|
||||
Path(dst).unlink(missing_ok=True)
|
||||
|
|
@ -1,117 +0,0 @@
|
|||
import sys
|
||||
from multiprocessing import Pool, cpu_count
|
||||
from argparse import ArgumentParser, ArgumentTypeError, Namespace
|
||||
from pathlib import Path
|
||||
from .todo import Todo
|
||||
|
||||
args = Namespace()
|
||||
|
||||
|
||||
def task(todo: Todo) -> None:
|
||||
todo.run()
|
||||
if args.verbose:
|
||||
print("finished", todo, file=sys.stderr)
|
||||
|
||||
|
||||
def main():
|
||||
def workers_type_func(value: object) -> int:
|
||||
try:
|
||||
value = int(value) # pyright: ignore[reportArgumentType]
|
||||
if value <= 0:
|
||||
raise ArgumentTypeError(f"{value} is not a positive integer")
|
||||
except ValueError:
|
||||
raise ArgumentTypeError(f"{value} is not an integer")
|
||||
|
||||
return value
|
||||
|
||||
parser = ArgumentParser()
|
||||
|
||||
_ = parser.add_argument(
|
||||
"-w",
|
||||
"--workers",
|
||||
default=cpu_count(),
|
||||
type=workers_type_func,
|
||||
help="amount of worker processes",
|
||||
)
|
||||
_ = parser.add_argument(
|
||||
"-i",
|
||||
"--interactive",
|
||||
action="store_true",
|
||||
help="prompt before running",
|
||||
)
|
||||
_ = parser.add_argument(
|
||||
"-k",
|
||||
"--keep",
|
||||
action="store_true",
|
||||
help="whether source files should be kept if both directories are the same",
|
||||
)
|
||||
_ = parser.add_argument(
|
||||
"-r",
|
||||
"--redo",
|
||||
action="store_true",
|
||||
help="whether everything should be re-encoded regardless of whether they have already been transcoded",
|
||||
)
|
||||
_ = parser.add_argument(
|
||||
"-v",
|
||||
"--verbose",
|
||||
action="count",
|
||||
default=0,
|
||||
help="verbose output",
|
||||
)
|
||||
|
||||
_ = parser.add_argument(
|
||||
"music",
|
||||
nargs=1,
|
||||
type=Path,
|
||||
help="the source directory",
|
||||
)
|
||||
_ = parser.add_argument(
|
||||
"comp",
|
||||
nargs=1,
|
||||
type=Path,
|
||||
help="the destination directory for compressed files",
|
||||
)
|
||||
|
||||
global args
|
||||
args = parser.parse_args(sys.argv[1:])
|
||||
|
||||
assert isinstance(args.workers, int) # pyright: ignore[reportAny]
|
||||
assert isinstance(args.interactive, bool) # pyright: ignore[reportAny]
|
||||
assert isinstance(args.keep, bool) # pyright: ignore[reportAny]
|
||||
assert isinstance(args.redo, bool) # pyright: ignore[reportAny]
|
||||
assert isinstance(args.verbose, int) # pyright: ignore[reportAny]
|
||||
assert isinstance(args.music, list) # pyright: ignore[reportAny]
|
||||
assert len(args.music) == 1 # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
|
||||
assert isinstance(args.music[0], Path) # pyright: ignore[reportUnknownMemberType]
|
||||
assert isinstance(args.comp, list) # pyright: ignore[reportAny]
|
||||
assert len(args.comp) == 1 # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
|
||||
assert isinstance(args.comp[0], Path) # pyright: ignore[reportUnknownMemberType]
|
||||
|
||||
src_dir = args.music[0] # pyright: ignore[reportUnknownMemberType]
|
||||
dst_dir = args.comp[0] # pyright: ignore[reportUnknownMemberType]
|
||||
|
||||
plan = list(
|
||||
Todo.plan(
|
||||
src_dir,
|
||||
dst_dir,
|
||||
replace=src_dir.samefile(dst_dir) and not args.keep,
|
||||
redo=args.redo,
|
||||
)
|
||||
)
|
||||
|
||||
if len(plan) == 0:
|
||||
print("Nothing to do", file=sys.stderr)
|
||||
sys.exit(0)
|
||||
|
||||
if args.verbose >= 1 or args.interactive:
|
||||
print("Plan:", file=sys.stderr)
|
||||
for todo in plan:
|
||||
print(todo, file=sys.stderr)
|
||||
|
||||
if args.interactive:
|
||||
result = input("Do you want to continue? [Y/n] ")
|
||||
if result.lower() not in ("", "y", "yes"):
|
||||
sys.exit(1)
|
||||
|
||||
with Pool(args.workers) as pool:
|
||||
_ = pool.map(task, plan)
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
from os import PathLike
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def replace(src: str | PathLike[str]) -> None:
|
||||
Path(src).unlink(missing_ok=True)
|
||||
|
|
@ -1,68 +0,0 @@
|
|||
from pathlib import Path
|
||||
from os import PathLike
|
||||
from enum import StrEnum
|
||||
from dataclasses import dataclass
|
||||
from typing import Self, override
|
||||
from collections.abc import Generator
|
||||
from .transcode import transcode as todo_transcode
|
||||
from .clean import clean as todo_clean
|
||||
from .replace import replace as todo_replace
|
||||
|
||||
SRC_SUFFIX = ".flac"
|
||||
DST_SUFFIX = ".opus"
|
||||
|
||||
|
||||
class TodoAct(StrEnum):
|
||||
TRANSCODE = "transcode"
|
||||
CLEAN = "clean"
|
||||
REPLACE = "replace"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Todo:
|
||||
act: TodoAct
|
||||
src: str | PathLike[str]
|
||||
dst: str | PathLike[str]
|
||||
|
||||
@classmethod
|
||||
def plan(
|
||||
cls: type[Self],
|
||||
src_dir: str | PathLike[str],
|
||||
dst_dir: str | PathLike[str],
|
||||
replace: bool = False,
|
||||
redo: bool = False,
|
||||
) -> Generator[Self]:
|
||||
def list_files(dir: str | PathLike[str], suffix: str) -> list[Path]:
|
||||
files: list[Path] = []
|
||||
for f in Path(dir).rglob("*"):
|
||||
if f.is_file() and f.suffix == suffix:
|
||||
files.append(f)
|
||||
return files
|
||||
|
||||
src_files = list_files(src_dir, SRC_SUFFIX)
|
||||
dst_files = list_files(dst_dir, DST_SUFFIX)
|
||||
|
||||
for f in src_files:
|
||||
e = dst_dir / (f.relative_to(src_dir).with_suffix(DST_SUFFIX))
|
||||
if redo or e not in dst_files:
|
||||
yield cls(TodoAct.TRANSCODE, f, e)
|
||||
if replace:
|
||||
yield cls(TodoAct.REPLACE, f, e)
|
||||
|
||||
for f in dst_files:
|
||||
e = src_dir / (f.relative_to(dst_dir).with_suffix(SRC_SUFFIX))
|
||||
if e not in src_files:
|
||||
yield cls(TodoAct.CLEAN, f, e)
|
||||
|
||||
def run(self) -> None:
|
||||
match self.act:
|
||||
case TodoAct.TRANSCODE:
|
||||
todo_transcode(self.src, self.dst)
|
||||
case TodoAct.CLEAN:
|
||||
todo_clean(self.dst)
|
||||
case TodoAct.REPLACE:
|
||||
todo_replace(self.src)
|
||||
|
||||
@override
|
||||
def __str__(self) -> str:
|
||||
return f"{self.act} {self.src} -> {self.dst}"
|
||||
|
|
@ -1,34 +0,0 @@
|
|||
from pathlib import Path
|
||||
from subprocess import run
|
||||
from os import PathLike
|
||||
|
||||
|
||||
class TranscodingError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def transcode(src: str | PathLike[str], dst: str | PathLike[str]) -> None:
|
||||
dst = Path(dst)
|
||||
|
||||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if dst.is_file():
|
||||
dst.unlink()
|
||||
|
||||
opusenc: tuple[str, ...] = (
|
||||
"opusenc",
|
||||
"--quiet",
|
||||
"--bitrate",
|
||||
"96.000",
|
||||
"--music",
|
||||
"--vbr",
|
||||
"--comp",
|
||||
"10",
|
||||
"--",
|
||||
str(src),
|
||||
str(dst),
|
||||
)
|
||||
|
||||
cp = run(opusenc)
|
||||
if cp.returncode != 0:
|
||||
raise TranscodingError(f"opusenc exited with code {cp.returncode}")
|
||||
Loading…
Add table
Add a link
Reference in a new issue