"""命令行录音：适合自动化、定时任务、无界面服务器。 示例:: 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())