264 lines
11 KiB
Python
264 lines
11 KiB
Python
"""命令行录音：适合自动化、定时任务、无界面服务器。
|
||
|
||
示例::
|
||
|
||
python -m recorder.cli --list-devices
|
||
python -m recorder.cli -d 3 -t 60 -o D:\\rec --name 会议
|
||
python -m recorder.cli --probe 3
|
||
python -m recorder.cli --analyze D:\\rec\\会议.wav
|
||
python -m recorder.cli --process D:\\rec\\会议.wav --normalize lufs --target -16
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import signal
|
||
import sys
|
||
import time
|
||
|
||
from . import __version__, engine, post
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(
|
||
prog="recorder",
|
||
description="RecorderStudio · 高清晰度无损录音机（命令行模式）",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="提示：加 --gui 可启动图形界面（等价于 python -m recorder.gui）。")
|
||
p.add_argument("--version", action="version", version=f"RecorderStudio {__version__}")
|
||
|
||
g = p.add_argument_group("设备")
|
||
g.add_argument("--list-devices", action="store_true", help="列出所有输入设备并退出")
|
||
g.add_argument("--probe", type=int, metavar="INDEX",
|
||
help="检测指定设备支持哪些采样率")
|
||
g.add_argument("-d", "--device", type=int, help="输入设备索引（默认使用系统默认设备）")
|
||
g.add_argument("--no-exclusive", action="store_true",
|
||
help="关闭 WASAPI 独占模式（共享模式，兼容性更好）")
|
||
g.add_argument("--blocksize", type=int, default=0, help="缓冲区采样数，0 = 自动")
|
||
|
||
g = p.add_argument_group("格式")
|
||
g.add_argument("-r", "--rate", type=int, default=48000, help="采样率（默认 48000）")
|
||
g.add_argument("-c", "--channels", type=int, default=2, help="声道数（默认 2）")
|
||
g.add_argument("-b", "--bit-depth", default="24",
|
||
choices=["16", "24", "32", "float32"],
|
||
help="位深（默认 24）")
|
||
g.add_argument("--no-dither", action="store_true", help="关闭 TPDF 抖动")
|
||
g.add_argument("--rf64", action="store_true", help="使用 RF64 容器（>4 GB 单文件）")
|
||
g.add_argument("--gain", type=float, default=0.0, help="软件增益 dB（默认 0）")
|
||
g.add_argument("--lowcut", type=float, default=0.0,
|
||
help="线性相位低切频率 Hz（0 = 关闭）")
|
||
|
||
g = p.add_argument_group("录制")
|
||
g.add_argument("-t", "--seconds", type=float, help="录制时长（秒）；不填则按回车停止")
|
||
g.add_argument("-o", "--outdir", default="", help="输出目录（默认 ./recordings）")
|
||
g.add_argument("--name", default="{datetime}_{device}", help="文件命名模板")
|
||
g.add_argument("--split-seconds", type=float, default=0.0, help="按时长自动分段（秒）")
|
||
g.add_argument("--split-mb", type=float, default=0.0, help="按体积自动分段（MB）")
|
||
g.add_argument("--stop-after-silence", type=float, default=0.0,
|
||
help="静音多少秒后自动停止（0 = 不自动停止）")
|
||
g.add_argument("--silence-threshold", type=float, default=-50.0,
|
||
help="静音判定阈值 dBFS")
|
||
|
||
g = p.add_argument_group("后处理")
|
||
g.add_argument("--analyze", metavar="WAV", help="分析已有录音并退出")
|
||
g.add_argument("--process", metavar="WAV", help="对已有录音做后期处理")
|
||
g.add_argument("--normalize", choices=["none", "peak", "lufs"], default="none",
|
||
help="归一化方式（默认 none）")
|
||
g.add_argument("--target", type=float, default=-16.0, help="响度归一化目标 LUFS")
|
||
g.add_argument("--target-peak", type=float, default=-1.0,
|
||
help="峰值归一化目标 dBFS")
|
||
g.add_argument("--trim", action="store_true", help="裁剪首尾静音")
|
||
g.add_argument("--mono", action="store_true", help="混合为单声道")
|
||
g.add_argument("--export", default="", help="导出格式预设（flac/mp3_320/opus/m4a/wav_16…）")
|
||
g.add_argument("--report", action="store_true", help="额外生成报告与波形预览图")
|
||
|
||
g.add_argument("--gui", action="store_true", help="启动图形界面")
|
||
return p
|
||
|
||
|
||
def cmd_list_devices() -> int:
|
||
print(engine.device_summary())
|
||
return 0
|
||
|
||
|
||
def cmd_probe(index: int, channels: int) -> int:
|
||
info = engine.find_device(index) if hasattr(engine, "find_device") else None
|
||
if info is None:
|
||
devs = [d for d in engine.list_input_devices() if d.index == index]
|
||
info = devs[0] if devs else None
|
||
if info is None:
|
||
print(f"找不到设备 {index}", file=sys.stderr)
|
||
return 1
|
||
print(f"设备 [{info.index}] {info.name} 宿主 API：{info.hostapi}")
|
||
res = engine.probe_capabilities(index, channels=min(channels,
|
||
info.max_input_channels))
|
||
print(f"检测通道数：{res['channels']}\n")
|
||
for rate, entry in res["rates"].items():
|
||
if entry["supported"]:
|
||
mode = "独占" if entry.get("exclusive") else "共享"
|
||
print(f" ✓ {rate:>6} Hz {mode} {', '.join(entry['dtypes'])}")
|
||
else:
|
||
print(f" ✗ {rate:>6} Hz 不支持（{entry.get('error', '')[:60]}）")
|
||
return 0
|
||
|
||
|
||
def cmd_record(args: argparse.Namespace) -> int:
|
||
cfg = engine.RecordConfig(
|
||
device=args.device,
|
||
samplerate=args.rate,
|
||
channels=args.channels,
|
||
bit_depth=args.bit_depth,
|
||
gain_db=args.gain,
|
||
exclusive=not args.no_exclusive,
|
||
blocksize=args.blocksize,
|
||
lowcut_hz=args.lowcut,
|
||
dither=not args.no_dither,
|
||
rf64=args.rf64,
|
||
output_dir=args.outdir or os.path.join(os.getcwd(), "recordings"),
|
||
name_template=args.name,
|
||
split_seconds=args.split_seconds,
|
||
split_megabytes=args.split_mb,
|
||
silence_threshold_dbfs=args.silence_threshold,
|
||
auto_stop_silence_seconds=args.stop_after_silence,
|
||
)
|
||
rec = engine.Recorder(cfg)
|
||
try:
|
||
rec.start()
|
||
except Exception as exc:
|
||
print(f"启动失败：{exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
print(f"正在录音 → {cfg.output_dir}")
|
||
print(f"格式：{cfg.wav_format().describe()} "
|
||
f"{'独占模式' if cfg.exclusive else '共享模式'}"
|
||
f"{' 低切 %.0f Hz' % cfg.lowcut_hz if cfg.lowcut_hz else ''}")
|
||
for n in rec.notes():
|
||
print(" · " + n)
|
||
|
||
stop_by_signal = {"flag": False}
|
||
|
||
def _handler(signum, frame): # noqa: ANN001
|
||
stop_by_signal["flag"] = True
|
||
|
||
for sig in (signal.SIGINT, signal.SIGTERM):
|
||
try:
|
||
signal.signal(sig, _handler)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if args.seconds:
|
||
deadline = time.time() + args.seconds
|
||
while time.time() < deadline and not stop_by_signal["flag"]:
|
||
time.sleep(0.2)
|
||
snap = rec.live()
|
||
sys.stdout.write(
|
||
f"\r 已录 {engine.format_duration(snap.elapsed)} "
|
||
f"{engine.format_bytes(snap.bytes_written)} "
|
||
f"峰值 {snap.peak_dbfs:+.1f} dBFS "
|
||
f"响度 {snap.meter.momentary_lufs:+.1f} LUFS "
|
||
f"xrun {snap.xruns} 溢出 {snap.overflow_blocks} ")
|
||
sys.stdout.flush()
|
||
if snap.state == engine.RecorderState.IDLE.value:
|
||
break # 静音自动停止
|
||
else:
|
||
print("按回车停止录音…")
|
||
input()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
print()
|
||
result = rec.stop()
|
||
|
||
print(post.render_report(result))
|
||
if args.report and result.primary_file:
|
||
files = post.write_metadata(result)
|
||
png = post.make_preview_for(result)
|
||
if png:
|
||
files.append(png)
|
||
print("\n已生成：" + "、".join(os.path.basename(f) for f in files))
|
||
if args.export and result.primary_file:
|
||
meta = post.EXPORT_PRESETS.get(args.export)
|
||
if meta is None:
|
||
print(f"未知导出格式：{args.export}；可选：{', '.join(post.EXPORT_PRESETS)}",
|
||
file=sys.stderr)
|
||
return 1
|
||
dst = post.default_export_path(result.primary_file, args.export)
|
||
ok, msg = post.export_audio(result.primary_file, dst, args.export)
|
||
print(("导出成功：" if ok else "导出失败：") + msg)
|
||
return 0 if ok else 1
|
||
return 0
|
||
|
||
|
||
def cmd_analyze(path: str) -> int:
|
||
if not os.path.exists(path):
|
||
print(f"文件不存在：{path}", file=sys.stderr)
|
||
return 1
|
||
print(post.regenerate_report(path))
|
||
return 0
|
||
|
||
|
||
def cmd_process(args: argparse.Namespace) -> int:
|
||
if not os.path.exists(args.process):
|
||
print(f"文件不存在：{args.process}", file=sys.stderr)
|
||
return 1
|
||
opts = post.ProcessOptions(
|
||
trim_silence=args.trim,
|
||
remove_dc=True,
|
||
lowcut_hz=args.lowcut,
|
||
normalize=args.normalize,
|
||
normalize_target_lufs=args.target,
|
||
normalize_target_dbfs=args.target_peak,
|
||
mono=args.mono,
|
||
bit_depth=args.bit_depth,
|
||
dither=not args.no_dither,
|
||
)
|
||
report = post.process_file(args.process, opts,
|
||
progress=lambda m: print(" " + m))
|
||
print("\n处理完成：" + report["output"])
|
||
for s in report.get("steps", []):
|
||
print(" · " + s)
|
||
an = report.get("analysis") or {}
|
||
if an:
|
||
print(f" 输出体检：峰值 {an.get('peak_dbfs')} dBFS "
|
||
f"真峰值 {an.get('true_peak_dbtp')} dBTP "
|
||
f"响度 {an.get('integrated_lufs')} LUFS")
|
||
if args.report:
|
||
png = post.write_waveform_png(
|
||
os.path.splitext(report["output"])[0] + "_waveform.png",
|
||
*post._read_whole(report["output"]))
|
||
print(" 波形预览：" + png)
|
||
if args.export:
|
||
meta = post.EXPORT_PRESETS.get(args.export)
|
||
if meta is None:
|
||
print(f"未知导出格式：{args.export}", file=sys.stderr)
|
||
return 1
|
||
dst = post.default_export_path(report["output"], args.export)
|
||
ok, msg = post.export_audio(report["output"], dst, args.export)
|
||
print((" 导出成功：" if ok else " 导出失败：") + msg)
|
||
return 0 if ok else 1
|
||
return 0
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
args = build_parser().parse_args(argv)
|
||
if args.gui:
|
||
from .gui import run
|
||
return run([sys.argv[0]])
|
||
if args.list_devices:
|
||
return cmd_list_devices()
|
||
if args.probe is not None:
|
||
return cmd_probe(args.probe, args.channels)
|
||
if args.analyze:
|
||
return cmd_analyze(args.analyze)
|
||
if args.process:
|
||
return cmd_process(args)
|
||
if engine.sd is None:
|
||
print(f"音频后端不可用：{engine.SD_IMPORT_ERROR}", file=sys.stderr)
|
||
return 2
|
||
return cmd_record(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|