diff --git a/config/youtube.ini b/config/youtube.ini index 7efb7ff..8d97b54 100644 --- a/config/youtube.ini +++ b/config/youtube.ini @@ -2,10 +2,22 @@ # 只写 stream key（推荐方式） key = x04z-564w-aks7-embw-30y4 +# ===== 优先：推流画质 ===== +# 视频码率 Mbps（720p 推荐 4-4.5，YouTube 建议 3-8） +video_bitrate_mbps = 4.5 + +# 推流状态日志间隔（秒），每 N 秒打印帧率/码率/网络等 +stats_interval_sec = 10 + +# ===== 其次：音频版权处理 ===== # 是否启用 Demucs 人声分离（防 YouTube Content ID 版权识别） # 是 = 启用，只保留人声；否 = 使用原始音频 enable_demucs = 是 # Demucs 是否优先使用 NVIDIA GPU 加速 # 是 = 有显卡则用 GPU；否 = 强制 CPU；无显卡时自动回退到 CPU -demucs_use_gpu = 是 \ No newline at end of file +demucs_use_gpu = 是 + +# 人声混合比例（0.5-1.0），改善人声自然度，减少尖锐感 +# 0.85 = 85%人声+15%其他（推荐，更自然）；1.0 = 纯人声（可能尖锐） +demucs_vocal_mix = 0.85 \ No newline at end of file diff --git a/demucs_pipeline.py b/demucs_pipeline.py index f05c3e5..7f28470 100644 --- a/demucs_pipeline.py +++ b/demucs_pipeline.py @@ -14,6 +14,12 @@ import queue import time import platform +try: + from stream_stats import StreamStats, parse_and_maybe_print + _STREAM_STATS_AVAILABLE = True +except ImportError: + _STREAM_STATS_AVAILABLE = False + # 可选依赖：Demucs 和 PyTorch DEMUCS_AVAILABLE = False DEMUCS_MODEL = None @@ -69,8 +75,11 @@ DEMUCS_CHUNK_SEC = 2.0 BYTES_PER_SAMPLE = 2 -def _demucs_separate_vocals(model, audio_chunk, device): - """对音频块进行人声分离，返回人声轨道""" +def _demucs_separate_vocals(model, audio_chunk, device, vocal_mix_ratio=1.0): + """ + 对音频块进行人声分离，返回人声轨道。 + vocal_mix_ratio: 1.0=纯人声；0.85=85%人声+15%other（更自然，减少尖锐感） + """ import torch # audio_chunk: numpy or tensor, shape (channels, samples) if hasattr(audio_chunk, 'numpy'): @@ -85,11 +94,16 @@ def _demucs_separate_vocals(model, audio_chunk, device): out = apply_model(model, wav, device=device, progress=False, overlap=0.25) # out: (batch, sources, channels, samples) # sources: [drums, bass, other, vocals] - vocals = out[0, 3] # 取 vocals - return vocals.cpu().numpy() + vocals = out[0, 3] + if vocal_mix_ratio >= 1.0: + return vocals.cpu().numpy() + # 混合 other 以保留更自然的人声，减少纯人声的尖锐感 + other = out[0, 2] + mixed = vocals * vocal_mix_ratio + other * (1.0 - vocal_mix_ratio) + return mixed.cpu().numpy() -def _run_demucs_worker(audio_read_pipe, audio_write_socket, model, device, stop_event): +def _run_demucs_worker(audio_read_pipe, audio_write_socket, model, device, stop_event, vocal_mix_ratio=1.0): """Demucs 工作线程：从 FFmpeg 读音频，分离人声，写入 socket""" import numpy as np chunk_samples = int(DEMUCS_SAMPLE_RATE * DEMUCS_CHUNK_SEC * DEMUCS_CHANNELS) @@ -117,7 +131,7 @@ def _run_demucs_worker(audio_read_pipe, audio_write_socket, model, device, stop_ arr = np.frombuffer(chunk, dtype=np.int16) arr = arr.reshape(-1, DEMUCS_CHANNELS).T.astype(np.float32) / 32768.0 try: - vocals = _demucs_separate_vocals(model, arr, device) + vocals = _demucs_separate_vocals(model, arr, device, vocal_mix_ratio) # float -> s16le out = (vocals * 32768).astype(np.int16).tobytes() try: @@ -136,7 +150,7 @@ def _run_demucs_worker(audio_read_pipe, audio_write_socket, model, device, stop_ try: arr = np.frombuffer(buf, dtype=np.int16) arr = arr.reshape(-1, DEMUCS_CHANNELS).T.astype(np.float32) / 32768.0 - vocals = _demucs_separate_vocals(model, arr, device) + vocals = _demucs_separate_vocals(model, arr, device, vocal_mix_ratio) out = (vocals * 32768).astype(np.int16).tobytes() audio_write_socket.sendall(out) except Exception: @@ -164,6 +178,8 @@ def run_demucs_pipeline( stream_ended_exception, cleanup_proc_fn, stderr_reader_fn, + vocal_mix_ratio: float = 0.85, + stats_interval_sec: int = 10, ): """ 运行 Demucs 管道：拉流 → 人声分离 → 推流 @@ -178,7 +194,8 @@ def run_demucs_pipeline( device = next(model.parameters()).device device_str = "cuda" if "cuda" in str(device) else "cpu" - print(f"[INFO] Demucs 人声分离已启用 ({device_str})") + mix_info = "纯人声" if vocal_mix_ratio >= 1.0 else f"{int(vocal_mix_ratio*100)}%人声+{int((1-vocal_mix_ratio)*100)}%其他" + print(f"[INFO] Demucs 人声分离已启用 ({device_str})，模式: {mix_info}") # TCP 服务器用于音频（跨平台） server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -246,7 +263,9 @@ def run_demucs_pipeline( "-c:v", nvenc_params.get("codec_v", "libx264"), "-preset", nvenc_params.get("preset_v", "fast"), "-profile:v", "high", - "-b:v", "2600k", "-maxrate", "3000k", "-bufsize", "5200k", + "-b:v", f"{nvenc_params.get('video_bitrate_k', 4500)}k", + "-maxrate", f"{nvenc_params.get('video_bitrate_k', 4500)+500}k", + "-bufsize", f"{nvenc_params.get('video_bitrate_k', 4500)*2}k", "-vsync", "cfr", "-g", "60", "-keyint_min", "60", "-r", "30", "-bf", "0", "-sc_threshold", "0", "-x264-params", "force-cfr=1:scenecut=0:open_gop=0:sync-lookahead=0", @@ -320,7 +339,7 @@ def run_demucs_pipeline( # 注意：proc_a 输出到 stdout，我们读。conn 是 FFmpeg 作为 client 连上来的，所以 FFmpeg 在读，我们写 conn。 demucs_thread = threading.Thread( target=_run_demucs_worker, - args=(proc_a.stdout, conn[0], model, device, stop_event), + args=(proc_a.stdout, conn[0], model, device, stop_event, vocal_mix_ratio), daemon=True, ) demucs_thread.start() @@ -331,14 +350,23 @@ def run_demucs_pipeline( reader_t.daemon = True reader_t.start() - # 监控 proc_out 的 stderr（推流到 YouTube，错误必须可见） + # 监控 proc_out 的 stderr（推流到 YouTube），解析状态用于详细日志 + stream_stats = None + if _STREAM_STATS_AVAILABLE: + stream_stats = StreamStats(interval_sec=stats_interval_sec) + stream_stats.start_periodic_printer() + print(f"[INFO] 推流状态监控已启用，每 {stats_interval_sec}s 输出帧率/码率/网络等") + def _proc_out_stderr_reader(): try: for line in iter(proc_out.stderr.readline, b""): if line: decoded = line.decode("utf-8", errors="replace").strip() if decoded: - print(f"[YouTube推流] {decoded}") + if stream_stats and parse_and_maybe_print(decoded, stream_stats, "[YouTube推流]", verbose=False): + pass # 已由 parse 处理，进度行不重复打印 + else: + print(f"[YouTube推流] {decoded}") except Exception: pass @@ -376,6 +404,8 @@ def run_demucs_pipeline( if line: proc_v.last_out_ts = time.time() + if stream_stats: + stream_stats.parse_lag_event(line) # 源端抖动统计 print(f"[FFmpeg] {line}") line_lower = line.lower() if any(kw in line_lower for kw in end_keywords): @@ -396,6 +426,8 @@ def run_demucs_pipeline( traceback.print_exc() finally: stop_event.set() + if stream_stats is not None: + stream_stats.stop() if proc_out is not None and proc_out.poll() is not None and proc_out.returncode != 0: print(f"[INFO] YouTube 推流进程退出码: {proc_out.returncode}") cleanup_proc_fn(proc_v) diff --git a/douyin_youtube_ffplay.py b/douyin_youtube_ffplay.py index cc44373..1e1f573 100644 --- a/douyin_youtube_ffplay.py +++ b/douyin_youtube_ffplay.py @@ -22,6 +22,12 @@ except ImportError: _DEMUCS_MODULE_AVAILABLE = False _check_demucs_available = lambda: False +try: + from stream_stats import StreamStats, parse_and_maybe_print + _STREAM_STATS_AVAILABLE = True +except ImportError: + _STREAM_STATS_AVAILABLE = False + class StreamEnded(Exception): """主播真实下播的专用异常""" @@ -98,6 +104,9 @@ def start_douyin_youtube_ffplay( enable_demucs = False demucs_use_gpu = True + demucs_vocal_mix = 0.85 # 0.85=人声更自然，1.0=纯人声（可能尖锐） + video_bitrate_mbps = 4.5 # 优先画质，YouTube 720p 推荐 4-4.5 + stats_interval_sec = 10 # 推流状态日志间隔 if os.path.exists(config_file): config = configparser.ConfigParser() config.read(config_file, encoding="utf-8") @@ -109,13 +118,34 @@ def start_douyin_youtube_ffplay( enable_demucs = val.lower() in ("1", "true", "yes", "是") elif key_lower == "demucs_use_gpu": demucs_use_gpu = val.lower() not in ("0", "false", "no", "否") + elif key_lower == "demucs_vocal_mix": + try: + v = float(val) + if 0.5 <= v <= 1.0: + demucs_vocal_mix = v + except ValueError: + pass + elif key_lower == "video_bitrate_mbps": + try: + v = float(val) + if 2.0 <= v <= 8.0: + video_bitrate_mbps = v + except ValueError: + pass + elif key_lower == "stats_interval_sec": + try: + v = int(val) + if 5 <= v <= 120: + stats_interval_sec = v + except ValueError: + pass elif val and key_lower in ("key", "stream_key", "streamkey"): youtube_rtmp = val if val.startswith("rtmp://") else rtmp_base + val print(f"[INFO] 从 {config_file} 读取 YouTube stream key: {youtube_rtmp}") # 兼容：若上面未匹配到 key，取第一个非配置项的值作为 stream key - if youtube_rtmp == rtmp_base + default_key: + if youtube_rtmp == rtmp_base + default_key: for k, v in config.items("youtube"): - if k.strip().lower() not in ("enable_demucs", "demucs_use_gpu") and v.strip(): + if k.strip().lower() not in ("enable_demucs", "demucs_use_gpu", "demucs_vocal_mix", "video_bitrate_mbps", "stats_interval_sec") and v.strip(): v = v.strip() youtube_rtmp = v if v.startswith("rtmp://") else rtmp_base + v print(f"[INFO] 从 {config_file} 读取 YouTube stream key: {youtube_rtmp}") @@ -157,6 +187,7 @@ def start_douyin_youtube_ffplay( print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}") # ====================== NVENC 检测 ====================== + # FFmpeg NVENC 需驱动 570+，否则会报 "Driver does not support the required nvenc API version" codec_v = "libx264" preset_v = "veryfast" tune_param = ["-tune", "zerolatency"] @@ -166,18 +197,37 @@ def start_douyin_youtube_ffplay( if check.returncode == 0: enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5) if "h264_nvenc" in (enc.stdout or ""): - codec_v = "h264_nvenc" - preset_v = "p5" - tune_param = ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"] - print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") + # 检查驱动版本，NVENC API 13.0 需 570+ + driver_ok = False + try: + ver_out = subprocess.run( + ["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader"], + capture_output=True, text=True, timeout=5 + ) + if ver_out.returncode == 0 and ver_out.stdout: + ver_str = ver_out.stdout.strip().split("\n")[0].strip() + major = int(ver_str.split(".")[0]) if ver_str else 0 + driver_ok = major >= 570 + if not driver_ok: + print(f"[INFO] NVIDIA 驱动 {ver_str} 低于 570，使用软件编码 (libx264)") + except Exception: + pass + if driver_ok: + codec_v = "h264_nvenc" + preset_v = "p5" + br_k = int(video_bitrate_mbps * 1000) + tune_param = ["-tune", "ll", "-rc", "vbr", "-b:v", f"{br_k}k", "-maxrate", f"{br_k+500}k", "-bufsize", f"{br_k*2}k"] + print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") except Exception as e: print(f"[WARN] NVENC 检测失败，回退软件编码: {e}") + br_k = int(video_bitrate_mbps * 1000) nvenc_params = { "codec_v": codec_v, "preset_v": preset_v, "tune_param": tune_param, "demucs_use_gpu": demucs_use_gpu, + "video_bitrate_k": br_k, } use_demucs = enable_demucs and _DEMUCS_MODULE_AVAILABLE and _check_demucs_available() @@ -202,9 +252,8 @@ def start_douyin_youtube_ffplay( "-preset", "fast", # "-tune", "zerolatency", "-profile:v", "high", - # 720p - # "-minrate", "3200k","-b:v", "3200k", "-maxrate", "3200k", "-bufsize", "6400k", # bufsize = 2× bitrate，保持真 CBR - "-b:v", "2600k","-maxrate", "3000k","-bufsize", "5200k", + # 720p，画质优先，YouTube 推荐 3-8 Mbps + "-b:v", f"{br_k}k", "-maxrate", f"{br_k+500}k", "-bufsize", f"{br_k*2}k", "-vsync", "cfr", "-g", "60", "-keyint_min", "60", "-r", "30", @@ -482,6 +531,8 @@ def start_douyin_youtube_ffplay( stream_ended_exception=StreamEnded, cleanup_proc_fn=cleanup_proc, stderr_reader_fn=_stderr_reader_thread, + vocal_mix_ratio=demucs_vocal_mix, + stats_interval_sec=stats_interval_sec, ) except StreamEnded: raise @@ -506,6 +557,11 @@ def start_douyin_youtube_ffplay( reader_t.start() start_time = last_frame_time = time.time() + stream_stats = None + if _STREAM_STATS_AVAILABLE: + stream_stats = StreamStats(interval_sec=stats_interval_sec) + stream_stats.start_periodic_printer() + print(f"[INFO] 推流状态监控已启用，每 {stats_interval_sec}s 输出帧率/码率/网络等") while True: try: @@ -523,15 +579,22 @@ def start_douyin_youtube_ffplay( break continue - line = line.strip() - if line.startswith("__ERR__READER__"): - print(f"[ERROR] stderr reader 异常: {line}") + line_str = line.strip() if isinstance(line, str) else line.decode("utf-8", errors="replace").strip() + if line_str.startswith("__ERR__READER__"): + print(f"[ERROR] stderr reader 异常: {line_str}") continue - if line: + if line_str: proc.last_out_ts = time.time() - print(f"[FFmpeg] {line}") - line_lower = line.lower() + if stream_stats: + stream_stats.parse_lag_event(line_str) + if stream_stats.parse_ffmpeg_progress(line_str): + pass # 进度行由周期摘要代替，不重复打印 + else: + print(f"[FFmpeg] {line_str}") + else: + print(f"[FFmpeg] {line_str}") + line_lower = line_str.lower() if any(kw in line_lower for kw in END_KEYWORDS): print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流") @@ -555,6 +618,9 @@ def start_douyin_youtube_ffplay( cleanup_proc(proc) retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) break + finally: + if stream_stats is not None: + stream_stats.stop() continue diff --git a/stream_stats.py b/stream_stats.py new file mode 100644 index 0000000..05429be --- /dev/null +++ b/stream_stats.py @@ -0,0 +1,174 @@ +# stream_stats.py +# 推流状态监控：解析 FFmpeg 输出，统计帧率、码率、网络等，便于调试 +""" +推流状态监控模块 +解析 FFmpeg stderr 输出，提取：帧率、码率、编码速度、源延迟事件 +可选：Linux 网络上下行速度 +""" +import re +import time +import threading +import platform + + +class StreamStats: + """推流状态统计，线程安全""" + + def __init__(self, interval_sec=10): + self.interval_sec = interval_sec + self._lock = threading.Lock() + self.frame = 0 + self.fps = 0.0 + self.bitrate_kbps = 0.0 + self.speed = 0.0 + self.encode_time = "" + self.size_kb = 0 + self.quality_q = 0.0 + self.lag_events = 0 # "Resumed reading" 次数，表示源端抖动 + self.last_update_ts = 0.0 + self.start_ts = time.time() + self._stop = threading.Event() + self._printer_thread = None + # 网络统计 (Linux) + self._net_last = None + self._net_last_ts = 0 + self.rx_mbps = 0.0 + self.tx_mbps = 0.0 + + def parse_ffmpeg_progress(self, line: str) -> bool: + """解析 FFmpeg 进度行，返回是否成功解析""" + # frame= 12345 fps= 30 q=28.0 size= 12345kB time=00:06:51.50 bitrate=2460.0kbits/s speed=1.00x + m = re.search(r'frame=\s*(\d+)', line, re.I) + if not m: + return False + with self._lock: + try: + self.frame = int(m.group(1)) + except (ValueError, IndexError): + pass + m2 = re.search(r'fps=\s*([\d.]+)', line, re.I) + if m2: + self.fps = float(m2.group(1)) + m3 = re.search(r'bitrate=\s*([\d.]+)\s*kbits/s', line, re.I) + if m3: + self.bitrate_kbps = float(m3.group(1)) + m4 = re.search(r'speed=\s*([\d.]+)x', line, re.I) + if m4: + self.speed = float(m4.group(1)) + m5 = re.search(r'time=([^\s]+)', line, re.I) + if m5: + self.encode_time = m5.group(1).strip() + m6 = re.search(r'size=\s*(\d+)\s*kB', line, re.I) + if m6: + self.size_kb = int(m6.group(1)) + m7 = re.search(r'q=([\d.]+)', line, re.I) + if m7: + self.quality_q = float(m7.group(1)) + self.last_update_ts = time.time() + return True + + def parse_lag_event(self, line: str) -> bool: + """检测源端延迟事件 Resumed reading at pts""" + if "resumed reading at pts" in line.lower(): + with self._lock: + self.lag_events += 1 + return True + return False + + def _read_net_bytes(self): + """读取 /proc/net/dev 总字节数 (Linux)""" + try: + with open("/proc/net/dev", "r") as f: + content = f.read() + except Exception: + return None, None + rx_total = tx_total = 0 + for line in content.split("\n")[2:]: # 跳过表头 + line = line.strip() + if not line or ":" not in line: + continue + parts = line.split() + if len(parts) >= 10: + try: + rx_total += int(parts[1]) + tx_total += int(parts[9]) + except (ValueError, IndexError): + pass + return rx_total, tx_total + + def update_network(self): + """更新网络速度 (Linux)""" + if platform.system() != "Linux": + return + rx, tx = self._read_net_bytes() + if rx is None: + return + now = time.time() + with self._lock: + if self._net_last is not None: + dt = now - self._net_last_ts + if dt > 0.5: # 至少 0.5 秒间隔 + drx = (rx - self._net_last[0]) / (1024 * 1024) / dt + dtx = (tx - self._net_last[1]) / (1024 * 1024) / dt + self.rx_mbps = round(drx, 2) + self.tx_mbps = round(dtx, 2) + self._net_last = (rx, tx) + self._net_last_ts = now + + def format_summary(self) -> str: + """生成状态摘要字符串""" + self.update_network() + with self._lock: + elapsed = time.time() - self.start_ts + parts = [ + f"推流时长 {int(elapsed)}s", + f"帧率 {self.fps:.1f} fps", + f"已推帧 {self.frame}", + f"码率 {self.bitrate_kbps:.0f} kbps", + f"编码速度 {self.speed:.2f}x", + ] + if self.lag_events > 0: + parts.append(f"源抖动 {self.lag_events} 次") + if platform.system() == "Linux": + parts.append(f"下行 {self.rx_mbps:.2f} MB/s 上行 {self.tx_mbps:.2f} MB/s") + return " | ".join(parts) + + def start_periodic_printer(self): + """启动周期性打印线程""" + + def _run(): + while not self._stop.is_set(): + self._stop.wait(self.interval_sec) + if self._stop.is_set(): + break + if self.last_update_ts > 0 and (time.time() - self.last_update_ts) < 60: + print(f"[推流状态] {self.format_summary()}") + + self._printer_thread = threading.Thread(target=_run, daemon=True) + self._printer_thread.start() + + def stop(self): + """停止周期性打印""" + self._stop.set() + + +def parse_and_maybe_print(line: str, stats: StreamStats, prefix: str = "[FFmpeg]", verbose=True) -> bool: + """ + 解析一行 FFmpeg 输出，更新 stats。 + 若为进度行则更新不打印；若为 lag 事件则更新并可选打印；其他行原样打印。 + verbose: 是否打印非进度行 + """ + line_stripped = line.strip() if isinstance(line, str) else line.decode("utf-8", errors="replace").strip() + if not line_stripped: + return False + # 进度行：更新 stats，不重复打印（由周期摘要代替） + if stats.parse_ffmpeg_progress(line_stripped): + return True + # 源端延迟事件 + if stats.parse_lag_event(line_stripped): + if verbose: + print(f"{prefix} {line_stripped}") + return True + if verbose: + print(f"{prefix} {line_stripped}") + return False