# douyin_youtube_ffplay.py import queue import unicodedata import platform import traceback import time import datetime import subprocess import threading import re import os import json import configparser from pathlib import Path from web2_proxy_config import merge_child_env from web2_push_proxy import ( apply_push_proxy_to_ffmpeg_cmd, ffmpeg_env_with_push_proxy, normalize_push_proxy_mode, redact_proxy_url, resolve_youtube_push_proxy, ) from web2_youtube_encoder_cache import get_cached_encoder, store_cached_encoder from web2_youtube_runtime import resolve_live_video_encoder from web2_youtube_relay_heartbeat import clear_relay_heartbeat, write_relay_heartbeat from web2_youtube_relay_slot import release_relay_slot class StreamEnded(Exception): """主播真实下播的专用异常""" pass class SourceStalled(Exception): """源流卡死/无进度，交回监控循环或切换备用源""" pass def _ffmpeg_speed_value(line: str) -> float | None: match = re.search(r"speed\s*=\s*([0-9.]+)x", line, re.IGNORECASE) if not match: return None try: return float(match.group(1)) except ValueError: return None def _ffmpeg_progress_values(line: str) -> tuple[int | None, float | None]: frame = None media_time = None frame_match = re.search(r"frame\s*=\s*(\d+)", line, re.IGNORECASE) if frame_match: frame = int(frame_match.group(1)) time_match = re.search(r"time\s*=\s*(\d{2}):(\d{2}):(\d{2}(?:\.\d+)?)", line, re.IGNORECASE) if time_match: h, m, s = time_match.groups() media_time = int(h) * 3600 + int(m) * 60 + float(s) return frame, media_time def _looks_like_invalid_ts_drop(line: str) -> bool: return "invalid dropping" in line or "non monotonically increasing dts" in line def sanitize_title(text: str, max_len: int = 120) -> str: if not text: return "" text = unicodedata.normalize("NFKC", text) text = ''.join(ch for ch in text if unicodedata.category(ch)[0] != 'C') text = ''.join(ch for ch in text if ord(ch) <= 0xFFFF) return text.strip()[:max_len] def _stderr_reader_thread(proc, q): try: while True: line = proc.stderr.readline() if not line: break q.put(line) except Exception as e: try: q.put(f"__ERR__READER__:{e}") except Exception: pass finally: try: q.put(None) except Exception: pass def cleanup_proc(proc): try: if proc and proc.poll() is None: try: proc.terminate() proc.wait(timeout=5) except Exception: if proc.poll() is None: try: proc.kill() except Exception: pass except Exception: pass def start_douyin_youtube_ffplay( title: str, anchor_name: str, real_url: str, rec_info: str, record_name: str, recording: set, recording_time_list: dict, running_list: list, monitoring: list, clear_record_info, color_obj, *, record_url: str = "", config_file: str = "", fallback_urls: list[str] | None = None, ): """ 抖音 → YouTube 推流主函数（单路推流） 只从 config/youtube.ini 读取一个 stream key（取第一个有效值） """ # ====================== 读取 YouTube 配置 ====================== config_file = (config_file or os.environ.get("YOUTUBE_INI_PATH") or "config/youtube.ini").strip() rtmps_base = "rtmps://a.rtmp.youtube.com/live2/" rtmp_base = "rtmp://a.rtmp.youtube.com/live2/" use_rtmps = True # 720p 竖屏 YouTube 建议 4–6 Mbps；1080p 竖屏适当提高码率 b_v, maxrate, bufsize = "4000k", "4500k", "8000k" stream_key = None fast_audio = False # True=轻量音频链（降噪关） output_resolution = "720p" force_software_video = False audio_denoise_on = False has_audio_denoise_key = False if os.path.exists(config_file): config = configparser.ConfigParser() config.read(config_file, encoding="utf-8") if config.has_section("youtube"): for k, v in config.items("youtube"): v = (v or "").strip() if k == "rtmps" and v.lower() in ("0", "false", "否"): use_rtmps = False elif k == "output_resolution" and v: output_resolution = v.lower() elif k == "use_gpu" and v: force_software_video = v.lower() in ("否", "0", "false", "无", "关", "off") elif k == "audio_denoise": has_audio_denoise_key = True audio_denoise_on = v.lower() in ("1", "true", "yes", "是", "开", "on") elif k == "fast_audio" and v.lower() in ("1", "true", "yes", "是", "开", "on"): fast_audio = True elif k == "bitrate" and v: try: n = int(v) b_v, maxrate, bufsize = f"{n}k", f"{min(n * 12 // 10, 8000)}k", f"{n * 2}k" except ValueError: pass elif (k == "key" or k == "stream_key") and v and not v.startswith("#"): stream_key = v if has_audio_denoise_key: fast_audio = not audio_denoise_on if "1080" in output_resolution: b_v, maxrate, bufsize = "6500k", "7800k", "13000k" if stream_key: if stream_key.startswith("rtmp://") or stream_key.startswith("rtmps://"): youtube_rtmp = stream_key else: youtube_rtmp = (rtmps_base if use_rtmps else rtmp_base) + stream_key print(f"[INFO] 从 {config_file} 读取 YouTube 推流地址") else: raise ValueError(f"请在 {config_file} 中配置 key=你的stream_key") # ====================== 常量配置 ====================== _script_dir = os.path.dirname(os.path.abspath(__file__)) _relay_pm2 = ( os.environ.get("LIVE_PM2_PROCESS_NAME") or os.environ.get("PM2_SERVICE_NAME") or "youtube" ).strip() or "youtube" def _emit_relay_heartbeat(state: str, **extra: object) -> None: write_relay_heartbeat( _relay_pm2, { "state": state, "stream_title": stream_title, "source_url": real_url, "codec": codec_v, "record_name": record_name, "anchor_name": anchor_name, **extra, }, base_dir=_script_dir, ) MAX_RETRY_DELAY = 90 INITIAL_RETRY_DELAY = 3 STDERR_QUEUE_TIMEOUT = 1.0 NO_FRAME_TIMEOUT = 120 START_CHECK_AFTER = 25 STALE_OUTPUT_SECONDS = 40 RECONNECT_DELAY_MAX = 10 FAST_SPEED_THRESHOLD = 1.08 FAST_SPEED_STRIKES_NEEDED = 3 INVALID_DROP_WINDOW_SECONDS = 30 INVALID_DROP_LIMIT = 80 TERMINAL_KEYWORDS = [ "live has ended", "server returned 404", "ingestion error", "access denied", "server returned 403", ] TRANSIENT_KEYWORDS = [ "connection reset by peer", "failed to write trailer", "invalid data found", "server returned 500", "packet too short", "no route to host", "connection timed out", "connection refused", "error writing output", "broken pipe", "i/o error", "client disconnected", "error opening output file", "error opening output files", "error opening output", ] douyin_headers = ( "User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15\r\n" "Referer: https://live.douyin.com/\r\n" "Origin: https://live.douyin.com\r\n" ) # ====================== 标题处理 ====================== timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) clean_title = sanitize_title(title) stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}") print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}") try: print( "[RELAY_META] " + json.dumps( { "ts": int(time.time()), "anchor": anchor_name, "douyinHint": (real_url or record_name or "")[:400], "streamTitle": stream_title, "youtubeKeySuffix": (stream_key[-8:] if stream_key and len(stream_key) > 8 else stream_key), }, ensure_ascii=False, ), flush=True, ) except Exception: pass try: from web2_relay_log import on_relay_push_start, relay_db_path ini_text = "" if os.path.exists(config_file): with open(config_file, encoding="utf-8-sig") as ini_fp: ini_text = ini_fp.read() on_relay_push_start( process=_relay_pm2, anchor_name=anchor_name, record_url=record_url or record_name, stream_title=stream_title, real_url=real_url, youtube_ini_text=ini_text, db_path=relay_db_path(Path(_script_dir)), base_dir=Path(_script_dir), ) except Exception as relay_log_exc: print(f"[WARN] relay session log start skipped: {relay_log_exc}") source_candidates: list[str] = [] _seen_source: set[str] = set() for item in [real_url, *(fallback_urls or [])]: u = (item or "").strip() if not u or u in _seen_source: continue if not u.startswith("http"): continue _seen_source.add(u) source_candidates.append(u) if not source_candidates: source_candidates = [real_url] source_index = 0 push_proxy_mode = normalize_push_proxy_mode() push_proxy = resolve_youtube_push_proxy() if push_proxy_mode != "off" else "" if push_proxy: print(f"[INFO] YouTube push proxy ({push_proxy_mode}): {redact_proxy_url(push_proxy)}") if "1080" in output_resolution: _sw, _sh = "1080", "1920" else: _sw, _sh = "720", "1280" scale_vf = ( f"fps=30,scale={_sw}:{_sh}:force_original_aspect_ratio=decrease:flags=bilinear," f"pad={_sw}:{_sh}:(ow-iw)/2:(oh-ih)/2:black" ) # ====================== NVENC / 编码器选择 ====================== codec_v = "libx264" video_args = ["-preset", "ultrafast", "-tune", "zerolatency", "-profile:v", "high", "-level", "4.0", "-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize] _machine = platform.machine().lower() _want_enc = (os.environ.get("LIVE_VIDEO_ENCODER") or "").strip() try: if platform.system().lower() == "linux": _cached_enc = get_cached_encoder(_machine, "ffmpeg", base_dir=_script_dir) if _cached_enc: codec_v = _cached_enc if _cached_enc == "h264_nvenc": video_args = ["-profile:v", "high", "-level", "4.0", "-tune", "ll", "-rc", "cbr", "-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize] print(f"[INFO] 使用编码器缓存: {codec_v}") elif not force_software_video: enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=6) blob = (enc.stdout or "") + (enc.stderr or "") selected, note = resolve_live_video_encoder(_want_enc, blob, _machine) if note: print(f"[INFO] {note}") if selected: codec_v = selected if selected == "h264_nvenc": video_args = ["-profile:v", "high", "-level", "4.0", "-tune", "ll", "-rc", "cbr", "-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize] store_cached_encoder(_machine, "ffmpeg", codec_v, base_dir=_script_dir) elif ( not force_software_video and platform.system().lower() in ["windows", "linux"] ): check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=5) 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" video_args = ["-profile:v", "high", "-level", "4.0", "-tune", "ll", "-rc", "cbr", "-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize] print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") elif force_software_video: print("[INFO] use_gpu=否/无：强制软件编码 libx264") except Exception as e: print(f"[WARN] 编码器检测失败，回退软件编码: {e}") # ====================== 开源音频处理：背景音乐弱化 + 防版权 ====================== _ffmpeg_cwd = None if fast_audio: print("[INFO] fast_audio=开：轻量音频链（无 arnndn/长 EQ），减轻 CPU 以利 speed≥1.0") _af_chain = "afftdn=nf=-26:tn=1,aresample=async=1:first_pts=0" else: # 优先 RNNoise(arnndn)，无模型则用 afftdn；aphaser 相位扰动可干扰 Content ID 指纹 _arnndn_paths = [ os.path.join(_script_dir, "arnndn-models", "cb.rnnn"), os.path.join(os.path.dirname(_script_dir), "arnndn-models", "cb.rnnn"), "arnndn-models/cb.rnnn", ] _arnndn_model = None for _p in _arnndn_paths: if os.path.isfile(_p): _arnndn_model = os.path.normpath(_p) break if _arnndn_model: print(f"[INFO] 使用 RNNoise 降噪模型: {_arnndn_model}") _arnndn_base = os.path.dirname(os.path.dirname(os.path.abspath(_arnndn_model))) _af_denoise = "arnndn=m=arnndn-models/cb.rnnn:mix=0.82," _ffmpeg_cwd = _arnndn_base else: print("[INFO] 未找到 arnndn 模型，使用 afftdn 降噪（可克隆 richardpl/arnndn-models 到 arnndn-models/）") _af_denoise = "afftdn=nf=-26:tn=1," # 强化版：收紧带通 + 频率偏移 + 音高微调 + 多段EQ + 相位扰动（纯 FFmpeg 理论上限 ~70%） _af_chain = ( _af_denoise + "highpass=f=200,lowpass=f=3800," + "equalizer=f=150:width_type=o:width=2.5:g=-14," + "equalizer=f=350:width_type=o:width=2:g=-10," + "equalizer=f=600:width_type=o:width=1.8:g=-8," + "equalizer=f=1000:width_type=o:width=1.5:g=-6," + "equalizer=f=2500:width_type=o:width=1.2:g=-4," + "acompressor=threshold=-26dB:ratio=5:attack=8:release=80:makeup=5," + "afreqshift=shift=22," + "asetrate=48000*1.035,aresample=48000," + "aphaser=type=t:decay=0.4:delay=3:speed=0.7," + "volume=1.12," + "aresample=async=1:first_pts=0" ) # 720p 竖屏，符合 YouTube 推荐：4–6 Mbps、2s 关键帧、AAC 128k _video_mid = ["-vsync", "cfr", "-g", "60", "-keyint_min", "60", "-r", "30", "-bf", "0", "-sc_threshold", "0"] if codec_v == "libx264": _video_mid += ["-x264-params", "force-cfr=1:scenecut=0:open_gop=0:sync-lookahead=0"] def _build_ffmpeg_command(input_url: str) -> list[str]: cmd = [ "ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-stats", "-stats_period", "1", "-rw_timeout", "50000000", "-reconnect", "1", "-reconnect_at_eof", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", str(RECONNECT_DELAY_MAX), "-fflags", "+genpts+discardcorrupt", "-err_detect", "ignore_err", "-max_delay", "500000", "-thread_queue_size", "2048", "-filter_threads", "4", "-headers", douyin_headers, "-re", "-i", input_url, "-vf", scale_vf, "-c:v", codec_v, *video_args, *_video_mid, "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", "-af", _af_chain, "-flags", "+global_header", "-flvflags", "no_duration_filesize", "-f", "flv", youtube_rtmp, ] return apply_push_proxy_to_ffmpeg_cmd(cmd, push_proxy, push_proxy_mode, base_dir=_script_dir) ffmpeg_command, _uses_proxychains = _build_ffmpeg_command(source_candidates[source_index]) _emit_relay_heartbeat("starting") proc = None stderr_q = None reader_t = None retry_delay = INITIAL_RETRY_DELAY fast_speed_strikes = 0 invalid_drop_count = 0 invalid_drop_window_start = 0.0 last_progress_frame = 0 last_progress_time = 0.0 try: while True: try: if retry_delay > INITIAL_RETRY_DELAY: print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...") time.sleep(retry_delay) cleanup_proc(proc) stderr_q = queue.Queue() print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}") _popen_kw = dict( stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True, ) if _ffmpeg_cwd: _popen_kw["cwd"] = _ffmpeg_cwd base_env = merge_child_env() _popen_kw["env"] = ffmpeg_env_with_push_proxy( push_proxy, push_proxy_mode, use_proxychains=_uses_proxychains, base_env=base_env, ) proc = subprocess.Popen(ffmpeg_command, **_popen_kw) proc.last_out_ts = time.time() reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q)) reader_t.daemon = True reader_t.start() start_time = last_frame_time = time.time() fast_speed_strikes = 0 while True: try: try: line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT) except queue.Empty: if time.time() - getattr(proc, "last_out_ts", time.time()) > STALE_OUTPUT_SECONDS: print("[WARN] 长时间无任何日志，强制重启进程") cleanup_proc(proc) raise SourceStalled() continue if line is None: if proc.poll() is not None: break continue line = line.strip() if line.startswith("__ERR__READER__"): print(f"[ERROR] stderr reader 异常: {line}") continue if line: proc.last_out_ts = time.time() print(f"[FFmpeg] {line}") line_lower = line.lower() is_progress_line = "frame=" in line_lower and "speed=" in line_lower if any(kw in line_lower for kw in TERMINAL_KEYWORDS): print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流") cleanup_proc(proc) raise StreamEnded() if any(kw in line_lower for kw in TRANSIENT_KEYWORDS): print("[WARN] 检测到瞬时网络/IO 错误，准备重连") cleanup_proc(proc) retry_delay = min(max(retry_delay * 2, INITIAL_RETRY_DELAY), MAX_RETRY_DELAY) break now_ts = time.time() if _looks_like_invalid_ts_drop(line_lower): if now_ts - invalid_drop_window_start > INVALID_DROP_WINDOW_SECONDS: invalid_drop_window_start = now_ts invalid_drop_count = 0 invalid_drop_count += 1 if invalid_drop_count >= INVALID_DROP_LIMIT: print("[WARN] 源流时间戳异常过多，重启 FFmpeg") cleanup_proc(proc) raise SourceStalled() if is_progress_line: speed = _ffmpeg_speed_value(line_lower) frame, media_time = _ffmpeg_progress_values(line_lower) _emit_relay_heartbeat( "streaming", frame=frame, media_time=media_time, speed=speed, source_index=source_index, ) if speed is not None and time.time() - start_time > START_CHECK_AFTER: if speed > FAST_SPEED_THRESHOLD: fast_speed_strikes += 1 if fast_speed_strikes >= FAST_SPEED_STRIKES_NEEDED: print(f"[WARN] 推流 speed={speed:.2f}x 异常，重启 FFmpeg") cleanup_proc(proc) raise SourceStalled() else: fast_speed_strikes = 0 frame, media_time = _ffmpeg_progress_values(line_lower) if frame is not None or media_time is not None: frame_advanced = frame is not None and frame > last_progress_frame time_advanced = media_time is not None and media_time > last_progress_time + 0.01 if frame_advanced or time_advanced: last_frame_time = time.time() retry_delay = INITIAL_RETRY_DELAY if frame is not None: last_progress_frame = frame if media_time is not None: last_progress_time = media_time if time.time() - start_time > START_CHECK_AFTER: if time.time() - last_frame_time > NO_FRAME_TIMEOUT: print(f"[WARN] 连续 {NO_FRAME_TIMEOUT}s 无新帧，交回监控循环重新获取直播源") cleanup_proc(proc) raise SourceStalled() except (StreamEnded, SourceStalled): raise except Exception as e: print(f"[ERROR] 内层异常: {e}") traceback.print_exc() cleanup_proc(proc) retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) break continue except StreamEnded: print(f"[INFO] 主播已下播，停止 YouTube 转推 → {record_name}") break except SourceStalled: if source_index + 1 < len(source_candidates): source_index += 1 ffmpeg_command, _uses_proxychains = _build_ffmpeg_command(source_candidates[source_index]) print( f"[WARN] 切换备用源 URL ({source_index + 1}/{len(source_candidates)}): " f"{source_candidates[source_index][:120]}" ) _emit_relay_heartbeat("source_failover", source_index=source_index) retry_delay = INITIAL_RETRY_DELAY continue print(f"[WARN] 当前直播源无进度，返回监控循环刷新源地址 → {record_name}") break except Exception as e: print(f"[ERROR] 外层异常: {e}") traceback.print_exc() retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) continue except Exception as e: print(f"[FATAL] YouTube 推流最外层异常: {e}") traceback.print_exc() finally: cleanup_proc(proc) clear_relay_heartbeat(_relay_pm2, base_dir=_script_dir) release_relay_slot(record_name, pm2=_relay_pm2, base_dir=_script_dir) try: from web2_relay_log import on_relay_push_end, relay_db_path on_relay_push_end( _relay_pm2, anchor_name=anchor_name, reason="push_stopped", db_path=relay_db_path(Path(_script_dir)), ) except Exception as relay_log_exc: print(f"[WARN] relay session log end skipped: {relay_log_exc}") try: recording.discard(record_name) recording_time_list.pop(record_name, None) if real_url in running_list: running_list.remove(real_url) monitoring[0] = max(0, monitoring[0] - 1) clear_record_info(record_name, real_url) except Exception: pass color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)