script_library/convert_mp3_to_flac/convert_mp3_to_flac.py
2025-08-25 22:40:47 +08:00

71 lines
2.2 KiB
Python

import os
import argparse
from pydub import AudioSegment # 依赖 audioop-lts
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
def convert_mp3_to_flac(mp3_path: str, delete_original: bool = False):
filename = os.path.basename(mp3_path)
flac_path = mp3_path.replace(".mp3", ".flac")
# 如果目标文件已存在,跳过
if os.path.exists(flac_path):
return "skip", f"已存在,跳过: {flac_path}"
try:
audio = AudioSegment.from_mp3(mp3_path)
audio.export(flac_path, format="flac")
if delete_original:
os.remove(mp3_path)
return "success", f"转换完成并删除原文件: {flac_path}"
else:
return "success", f"转换完成: {flac_path}"
except Exception as e:
return "fail", f"转换失败: {mp3_path}, 错误: {e}"
def main():
parser = argparse.ArgumentParser(description="批量将 MP3 转换为 FLAC")
parser.add_argument("input_dir", help="输入的 mp3 文件夹路径")
parser.add_argument("-t", "--threads", type=int, default=4, help="线程数 (默认: 4)")
parser.add_argument("--delete", action="store_true", help="转换后删除原 mp3 文件")
args = parser.parse_args()
input_dir = args.input_dir
threads = args.threads
delete_original = args.delete
mp3_files = [
os.path.join(input_dir, f) for f in os.listdir(input_dir) if f.lower().endswith(".mp3")
]
if not mp3_files:
print("未找到任何 mp3 文件。")
return
stats = {"success": 0, "skip": 0, "fail": 0}
with ThreadPoolExecutor(max_workers=threads) as executor:
futures = {
executor.submit(convert_mp3_to_flac, f, delete_original): f for f in mp3_files
}
for future in tqdm(as_completed(futures), total=len(futures), desc="转换进度", ncols=80):
status, msg = future.result()
print(msg)
if status in stats:
stats[status] += 1
# 统计总结
print("\n====== 转换总结 ======")
print(f"总文件数: {len(mp3_files)}")
print(f"成功: {stats['success']}")
print(f"跳过: {stats['skip']}")
print(f"失败: {stats['fail']}")
if __name__ == "__main__":
main()