A
This commit is contained in:
commit
b4fc3a1c23
3 files changed
+477
No files matched your search
@@ -0,0 +1,42 @@
|
||||
# Repository Guidelines
|
||||
|
||||
## Overview
|
||||
TypeChange is a small Python script (`convert.py`) that scans a music library and converts supported audio files to AAC (`.m4a`) using FFmpeg. It is currently Windows-oriented (default paths like `E:\Music`) and overwrites outputs in a flat `out/` directory.
|
||||
|
||||
## Project Structure & Module Organization
|
||||
- `convert.py`: main entry point (file discovery, name de-duplication, parallel conversion).
|
||||
- `ffmpeg-8.0.1-essentials_build/`: vendored FFmpeg Windows binaries and docs.
|
||||
- `out/`: generated output files (safe to delete/recreate; avoid treating as source).
|
||||
- `.venv/`, `.idea/`: local environment/IDE metadata (do not rely on them in code).
|
||||
|
||||
## Build, Test, and Development Commands
|
||||
This repo has no build step. Typical local workflow (Windows):
|
||||
- Create venv: `py -3.11 -m venv .venv`
|
||||
- Install deps: `.venv\Scripts\python.exe -m pip install -r requirements.txt`
|
||||
- Run converter: `.venv\Scripts\python.exe convert.py`
|
||||
- Use system FFmpeg instead of bundled: ensure `ffmpeg` is on `PATH` and remove/rename the local `ffmpeg-*/bin/ffmpeg.exe`.
|
||||
|
||||
## Configuration Notes
|
||||
- Defaults live at the top of `convert.py`: `SOURCE_ROOT` (`E:\Music`), `SKIP_ROOT` (`E:\Music\工具`), and `OUT_ROOT` (`E:\Music\工具\TypeChange\out`).
|
||||
- Outputs are flattened by filename; if multiple sources map to the same `*.m4a`, the last one found wins and the script prints an overwrite summary.
|
||||
- Bitrate/codec are controlled in the FFmpeg args (currently `-c:a aac` and `-b:a 256k`).
|
||||
|
||||
## Coding Style & Naming Conventions
|
||||
- Python: 4-space indentation, PEP 8, `snake_case` for functions/vars, `UPPER_CASE` for module constants.
|
||||
- Prefer `pathlib.Path` and pure helper functions (e.g., keep path selection logic separate from `subprocess` execution).
|
||||
- Avoid new dependencies unless they clearly simplify behavior; if added, document them and how to install.
|
||||
|
||||
## Testing Guidelines
|
||||
No automated tests are currently included. When changing conversion logic, do a small manual smoke test by pointing `SOURCE_ROOT` to a temporary folder and verifying:
|
||||
- files are skipped under `SKIP_ROOT`
|
||||
- output naming collision behavior matches expectations
|
||||
- FFmpeg errors are surfaced (non-zero exit shows `[FAIL]`)
|
||||
|
||||
## Commit & Pull Request Guidelines
|
||||
This directory does not currently include Git history. If you track it with Git:
|
||||
- Use Conventional Commits (e.g., `feat:`, `fix:`, `chore:`) and keep changes focused.
|
||||
- In PRs, include a short summary, what OS you tested on, and note any behavior that can overwrite files (paths, naming, bitrate).
|
||||
|
||||
## Agent / Automation Notes
|
||||
- Avoid expensive scans over `out/` and `ffmpeg-8.0.1-essentials_build/` (large generated/vendored assets).
|
||||
- Keep automated changes minimal and prefer editing `convert.py` unless the task explicitly requires restructuring.
|
||||
+434
@@ -0,0 +1,434 @@
|
||||
"""
|
||||
Convert all music under E:\\Music (except E:\\Music\\工具) to AAC (.m4a) using ffmpeg.
|
||||
Outputs are written flat into E:\\Music\\工具\\TypeChange\\out; if two source files
|
||||
or existing outputs share the same name, the last one wins and a summary of
|
||||
overwrites is printed at the end.
|
||||
|
||||
If `mutagen` is installed, FLAC tags (Vorbis comments) and cover art are written
|
||||
to the output M4A using standard MP4/iTunes atoms (including `covr`).
|
||||
"""
|
||||
|
||||
import os
|
||||
import argparse
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
# Source and destination roots
|
||||
SOURCE_ROOT = Path(r"E:\Music\完成\单曲")
|
||||
SKIP_ROOT = SOURCE_ROOT / "工具"
|
||||
OUT_ROOT = Path(r"E:\Music\工具\TypeChange\out")
|
||||
|
||||
# ffmpeg binary bundled next to this script; falls back to PATH if missing
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
LOCAL_FFMPEG = SCRIPT_DIR / "ffmpeg-8.0.1-essentials_build" / "bin" / "ffmpeg.exe"
|
||||
FFMPEG_BIN = str(LOCAL_FFMPEG if LOCAL_FFMPEG.exists() else "ffmpeg")
|
||||
|
||||
try:
|
||||
from mutagen.flac import FLAC
|
||||
from mutagen.mp4 import MP4, MP4Cover, MP4FreeForm
|
||||
|
||||
try:
|
||||
from mutagen.mp4 import AtomDataType
|
||||
except ImportError:
|
||||
AtomDataType = None
|
||||
|
||||
HAS_MUTAGEN = True
|
||||
except ImportError:
|
||||
FLAC = None
|
||||
MP4 = None
|
||||
MP4Cover = None
|
||||
MP4FreeForm = None
|
||||
AtomDataType = None
|
||||
HAS_MUTAGEN = False
|
||||
|
||||
# Extensions we will attempt to transcode
|
||||
SOURCE_EXTS = {
|
||||
".flac",
|
||||
".wav",
|
||||
".mp3",
|
||||
".ogg",
|
||||
".m4a",
|
||||
".aac",
|
||||
".wma",
|
||||
".ape",
|
||||
".alac",
|
||||
".wv",
|
||||
".aif",
|
||||
".aiff",
|
||||
".mka",
|
||||
}
|
||||
|
||||
COVER_FILENAMES = [
|
||||
"cover.jpg",
|
||||
"folder.jpg",
|
||||
"front.jpg",
|
||||
"album.jpg",
|
||||
"artwork.jpg",
|
||||
"cover.jpeg",
|
||||
"folder.jpeg",
|
||||
"front.jpeg",
|
||||
"album.jpeg",
|
||||
"artwork.jpeg",
|
||||
"cover.png",
|
||||
"folder.png",
|
||||
"front.png",
|
||||
"album.png",
|
||||
"artwork.png",
|
||||
]
|
||||
|
||||
|
||||
def is_in_skip_root(path: Path) -> bool:
|
||||
"""Check whether path lives under the skipped root folder."""
|
||||
try:
|
||||
path.resolve().relative_to(SKIP_ROOT.resolve())
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def iter_source_files():
|
||||
"""Yield candidate source files to convert."""
|
||||
for root, dirs, files in os.walk(SOURCE_ROOT):
|
||||
# Skip the entire SKIP_ROOT subtree early
|
||||
current_root = Path(root)
|
||||
if is_in_skip_root(current_root):
|
||||
dirs[:] = []
|
||||
continue
|
||||
|
||||
for name in files:
|
||||
path = current_root / name
|
||||
if path.suffix.lower() in SOURCE_EXTS:
|
||||
yield path
|
||||
|
||||
|
||||
def build_output_path(src: Path) -> Path:
|
||||
"""Place all outputs in OUT_ROOT, using only the filename with .m4a extension."""
|
||||
return OUT_ROOT / src.with_suffix(".m4a").name
|
||||
|
||||
|
||||
def find_external_cover_image(src: Path) -> Path | None:
|
||||
"""Find a cover image file in the same folder as src (e.g. cover.jpg / folder.jpg)."""
|
||||
try:
|
||||
files = {p.name.lower(): p for p in src.parent.iterdir() if p.is_file()}
|
||||
except OSError:
|
||||
return None
|
||||
for name in COVER_FILENAMES:
|
||||
path = files.get(name)
|
||||
if path:
|
||||
return path
|
||||
return None
|
||||
|
||||
|
||||
def _parse_int(value: str | None) -> int | None:
|
||||
if not value:
|
||||
return None
|
||||
value = value.strip()
|
||||
if not value:
|
||||
return None
|
||||
# Handle "1/10" style values.
|
||||
value = value.split("/", 1)[0].strip()
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_number_total(number_value: str | None, total_value: str | None) -> tuple[int, int] | None:
|
||||
if not number_value:
|
||||
return None
|
||||
|
||||
number_value = number_value.strip()
|
||||
total = _parse_int(total_value)
|
||||
|
||||
# Prefer parsing "1/10" from the number field if present.
|
||||
if "/" in number_value:
|
||||
left, right = number_value.split("/", 1)
|
||||
number = _parse_int(left)
|
||||
total = total if total is not None else _parse_int(right)
|
||||
else:
|
||||
number = _parse_int(number_value)
|
||||
|
||||
if number is None:
|
||||
return None
|
||||
return number, (total or 0)
|
||||
|
||||
|
||||
def _mp4_freeform_values(values: list[str]) -> list:
|
||||
if not values:
|
||||
return []
|
||||
if not MP4FreeForm:
|
||||
return []
|
||||
|
||||
# MP4 freeform atoms store bytes with a declared data type (typically UTF-8).
|
||||
if AtomDataType:
|
||||
return [MP4FreeForm(v.encode("utf-8"), dataformat=AtomDataType.UTF8) for v in values if v]
|
||||
return [MP4FreeForm(v.encode("utf-8")) for v in values if v]
|
||||
|
||||
|
||||
def sync_tags_from_flac(src: Path, dst: Path) -> bool:
|
||||
"""Copy FLAC (Vorbis) tags + cover art to M4A using MP4/iTunes atoms."""
|
||||
if not HAS_MUTAGEN or not FLAC or not MP4:
|
||||
return False
|
||||
|
||||
flac = FLAC(str(src))
|
||||
mp4 = MP4(str(dst))
|
||||
if mp4.tags is None:
|
||||
mp4.add_tags()
|
||||
else:
|
||||
mp4.tags.clear()
|
||||
|
||||
src_tags: dict[str, list[str]] = {}
|
||||
if flac.tags:
|
||||
for k, v in flac.tags.items():
|
||||
key = str(k).strip().upper()
|
||||
values = [str(x) for x in (v or [])]
|
||||
if not key or not values:
|
||||
continue
|
||||
src_tags.setdefault(key, []).extend(values)
|
||||
|
||||
def pick(*keys: str) -> list[str]:
|
||||
for key in keys:
|
||||
values = src_tags.get(key.upper())
|
||||
if values:
|
||||
return [v for v in values if v]
|
||||
return []
|
||||
|
||||
def pick_first(*keys: str) -> str | None:
|
||||
values = pick(*keys)
|
||||
return values[0] if values else None
|
||||
|
||||
used_keys: set[str] = set()
|
||||
|
||||
def set_str_list(mp4_key: str, values: list[str], *src_keys: str):
|
||||
cleaned = [v for v in values if v and v.strip()]
|
||||
if cleaned:
|
||||
mp4[mp4_key] = cleaned
|
||||
for k in src_keys:
|
||||
used_keys.add(k.upper())
|
||||
|
||||
set_str_list("\xa9nam", pick("TITLE"), "TITLE")
|
||||
set_str_list("\xa9ART", pick("ARTIST"), "ARTIST")
|
||||
set_str_list("aART", pick("ALBUMARTIST", "ALBUM ARTIST"), "ALBUMARTIST", "ALBUM ARTIST")
|
||||
set_str_list("\xa9alb", pick("ALBUM"), "ALBUM")
|
||||
set_str_list("\xa9gen", pick("GENRE"), "GENRE")
|
||||
set_str_list("\xa9wrt", pick("COMPOSER"), "COMPOSER")
|
||||
set_str_list("\xa9cmt", pick("COMMENT", "DESCRIPTION"), "COMMENT", "DESCRIPTION")
|
||||
|
||||
date_value = pick("DATE", "YEAR", "ORIGINALDATE", "ORIGINALYEAR")
|
||||
if date_value:
|
||||
mp4["\xa9day"] = date_value
|
||||
used_keys.update({"DATE", "YEAR", "ORIGINALDATE", "ORIGINALYEAR"})
|
||||
|
||||
lyrics_values = pick("LYRICS", "UNSYNCEDLYRICS", "LYRIC", "UNSYNCED LYRICS")
|
||||
if lyrics_values:
|
||||
mp4["\xa9lyr"] = ["\n".join([v for v in lyrics_values if v])]
|
||||
used_keys.update({"LYRICS", "UNSYNCEDLYRICS", "LYRIC", "UNSYNCED LYRICS"})
|
||||
|
||||
track_pair = _parse_number_total(
|
||||
pick_first("TRACKNUMBER", "TRACK"),
|
||||
pick_first("TRACKTOTAL", "TOTALTRACKS", "TRACKTOTALS"),
|
||||
)
|
||||
if track_pair:
|
||||
mp4["trkn"] = [track_pair]
|
||||
used_keys.update({"TRACKNUMBER", "TRACK", "TRACKTOTAL", "TOTALTRACKS", "TRACKTOTALS"})
|
||||
|
||||
disc_pair = _parse_number_total(
|
||||
pick_first("DISCNUMBER", "DISC"),
|
||||
pick_first("DISCTOTAL", "TOTALDISCS", "DISCTOTALS"),
|
||||
)
|
||||
if disc_pair:
|
||||
mp4["disk"] = [disc_pair]
|
||||
used_keys.update({"DISCNUMBER", "DISC", "DISCTOTAL", "TOTALDISCS", "DISCTOTALS"})
|
||||
|
||||
bpm = _parse_int(pick_first("BPM", "TBPM"))
|
||||
if bpm is not None:
|
||||
mp4["tmpo"] = [bpm]
|
||||
used_keys.update({"BPM", "TBPM"})
|
||||
|
||||
compilation_value = pick_first("COMPILATION")
|
||||
if compilation_value:
|
||||
mp4["cpil"] = [compilation_value.strip().lower() in {"1", "true", "yes"}]
|
||||
used_keys.add("COMPILATION")
|
||||
|
||||
# Cover art: prefer embedded FLAC picture type 3 (front cover), else external cover file.
|
||||
cover_bytes: bytes | None = None
|
||||
cover_is_png = False
|
||||
pictures = list(getattr(flac, "pictures", []) or [])
|
||||
if pictures:
|
||||
front = next((p for p in pictures if getattr(p, "type", None) == 3), pictures[0])
|
||||
cover_bytes = getattr(front, "data", None)
|
||||
mime = (getattr(front, "mime", "") or "").lower()
|
||||
cover_is_png = mime.endswith("png") or (isinstance(cover_bytes, (bytes, bytearray)) and cover_bytes[:8] == b"\x89PNG\r\n\x1a\n")
|
||||
|
||||
if not cover_bytes:
|
||||
external_cover = find_external_cover_image(src)
|
||||
if external_cover:
|
||||
try:
|
||||
cover_bytes = external_cover.read_bytes()
|
||||
cover_is_png = external_cover.suffix.lower() == ".png" or cover_bytes[:8] == b"\x89PNG\r\n\x1a\n"
|
||||
except OSError:
|
||||
cover_bytes = None
|
||||
|
||||
if cover_bytes and MP4Cover:
|
||||
mp4["covr"] = [
|
||||
MP4Cover(
|
||||
bytes(cover_bytes),
|
||||
imageformat=MP4Cover.FORMAT_PNG if cover_is_png else MP4Cover.FORMAT_JPEG,
|
||||
)
|
||||
]
|
||||
|
||||
# Preserve remaining Vorbis tags as MP4 freeform atoms.
|
||||
for key, values in sorted(src_tags.items()):
|
||||
if key in used_keys:
|
||||
continue
|
||||
if key in {"METADATA_BLOCK_PICTURE"}:
|
||||
continue
|
||||
freeform_key = f"----:com.apple.iTunes:{key}"
|
||||
freeform_values = _mp4_freeform_values(values)
|
||||
if freeform_values:
|
||||
mp4[freeform_key] = freeform_values
|
||||
|
||||
mp4.save()
|
||||
return True
|
||||
|
||||
|
||||
def convert_file(src: Path, dst: Path, *, tags_only: bool) -> bool:
|
||||
"""Run ffmpeg to transcode a single file to AAC. Returns True on success."""
|
||||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if tags_only:
|
||||
if src.suffix.lower() != ".flac":
|
||||
return True
|
||||
if not HAS_MUTAGEN:
|
||||
sys.stderr.write(f"[SKIP] mutagen missing; cannot sync tags: {src}\n")
|
||||
return False
|
||||
if not dst.exists():
|
||||
sys.stderr.write(f"[SKIP] output missing; cannot sync tags: {dst}\n")
|
||||
return False
|
||||
try:
|
||||
return sync_tags_from_flac(src, dst)
|
||||
except Exception as exc:
|
||||
sys.stderr.write(f"[WARN] Tag sync failed for {dst}: {exc}\n")
|
||||
return False
|
||||
|
||||
cmd = [FFMPEG_BIN, "-y"] # overwrite if needed
|
||||
cmd += ["-i", str(src)]
|
||||
cmd += [
|
||||
"-map",
|
||||
"0:a", # audio only; cover art + full tags are applied in a separate step
|
||||
"-map_metadata",
|
||||
"0", # copy tags
|
||||
"-map_chapters",
|
||||
"0",
|
||||
"-c:a",
|
||||
"aac", # re-encode audio
|
||||
"-b:a",
|
||||
"256k",
|
||||
"-movflags",
|
||||
"+faststart",
|
||||
str(dst),
|
||||
]
|
||||
|
||||
try:
|
||||
subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
if src.suffix.lower() == ".flac" and HAS_MUTAGEN:
|
||||
try:
|
||||
sync_tags_from_flac(src, dst)
|
||||
except Exception as exc:
|
||||
sys.stderr.write(f"[WARN] Tag sync failed for {dst}: {exc}\n")
|
||||
return True
|
||||
except subprocess.CalledProcessError as exc:
|
||||
sys.stderr.write(f"[FAIL] {src} -> {dst} ({exc})\n")
|
||||
return False
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(description="Transcode music to AAC (.m4a) and preserve tags.")
|
||||
parser.add_argument(
|
||||
"--tags-only",
|
||||
action="store_true",
|
||||
help="Skip transcoding; only sync FLAC tags/cover to existing outputs in OUT_ROOT.",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
if not SOURCE_ROOT.exists():
|
||||
sys.stderr.write(f"Source root missing: {SOURCE_ROOT}\n")
|
||||
sys.exit(1)
|
||||
|
||||
files = list(iter_source_files())
|
||||
if not files:
|
||||
print("No source files found; nothing to do.")
|
||||
return
|
||||
|
||||
# 目标名 -> 源文件列表(保留最后一个)
|
||||
dest_to_sources = {}
|
||||
duplicate_warnings = []
|
||||
preexisting_outputs = []
|
||||
|
||||
for src in files:
|
||||
dst = build_output_path(src)
|
||||
if dst in dest_to_sources:
|
||||
dest_to_sources[dst].append(src)
|
||||
else:
|
||||
dest_to_sources[dst] = [src]
|
||||
|
||||
tasks = []
|
||||
for dst, src_list in dest_to_sources.items():
|
||||
if len(src_list) > 1:
|
||||
duplicate_warnings.append((dst, list(src_list)))
|
||||
if dst.exists():
|
||||
preexisting_outputs.append(dst)
|
||||
# 只保留列表中的最后一个源文件作为最终输出
|
||||
tasks.append((src_list[-1], dst))
|
||||
|
||||
if duplicate_warnings:
|
||||
print("检测到同名输出,将覆盖(按遍历顺序最后出现的文件生效):")
|
||||
for dst, srcs in duplicate_warnings:
|
||||
print(f" {dst.name}:")
|
||||
for s in srcs:
|
||||
print(f" - {s}")
|
||||
|
||||
if preexisting_outputs:
|
||||
print("输出目录中已存在同名文件,将被覆盖:")
|
||||
for p in preexisting_outputs:
|
||||
print(f" {p}")
|
||||
|
||||
max_workers = max(1, os.cpu_count() or 4)
|
||||
successes = 0
|
||||
|
||||
print(f"Using ffmpeg at: {FFMPEG_BIN}")
|
||||
if not HAS_MUTAGEN:
|
||||
print("Note: Install 'mutagen' to fully preserve FLAC tags/cover art (pip install mutagen).")
|
||||
if args.tags_only:
|
||||
print("Mode: tags-only (no transcoding).")
|
||||
if args.tags_only:
|
||||
print(f"Found {len(files)} source file(s), {len(tasks)} unique target name(s) to sync.")
|
||||
else:
|
||||
print(f"Found {len(files)} file(s) to convert, {len(tasks)} unique target name(s).")
|
||||
print(f"Using up to {max_workers} parallel worker(s).\n")
|
||||
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
future_map = {}
|
||||
for src, dst in tasks:
|
||||
future = executor.submit(convert_file, src, dst, tags_only=args.tags_only)
|
||||
future_map[future] = (src, dst)
|
||||
|
||||
for future in as_completed(future_map):
|
||||
src, dst = future_map[future]
|
||||
ok = future.result()
|
||||
if ok:
|
||||
successes += 1
|
||||
print(f"[OK ] {src} -> {dst}")
|
||||
else:
|
||||
print(f"[FAIL] {src} -> {dst}")
|
||||
|
||||
print(f"\nDone. {successes}/{len(tasks)} target(s) written from {len(files)} source file(s).")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1 @@
|
||||
mutagen>=1.46
|
||||
Reference in new issue
Block a user