diff --git a/douyin_youtube_ffplay.py b/douyin_youtube_ffplay.py index 6cd5b9e..76bfa40 100644 --- a/douyin_youtube_ffplay.py +++ b/douyin_youtube_ffplay.py @@ -11,6 +11,7 @@ import re import os import configparser + class StreamEnded(Exception): """主播真实下播的专用异常""" pass @@ -33,40 +34,31 @@ def _stderr_reader_thread(proc, q): break q.put(line) except Exception as e: - q.put(f"__ERR__READER__:{e}") - finally: - q.put(None) - - -def _cleanup_proc(proc): - if proc and proc.poll() is None: try: - proc.terminate() - proc.wait(timeout=5) - except: - try: - proc.kill() - except: - pass + q.put(f"__ERR__READER__:{e}") + except Exception: + pass + finally: + try: + q.put(None) + except Exception: + pass -def _detect_nvenc(): +def cleanup_proc(proc): try: - if platform.system().lower() not in ["windows", "linux"]: - return "libx264", "veryfast", ["-tune", "zerolatency"] - - check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=5) - if check.returncode != 0: - return "libx264", "veryfast", ["-tune", "zerolatency"] - - enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5) - if "h264_nvenc" in (enc.stdout or ""): - print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") - return "h264_nvenc", "p5", ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "5200k"] - return "libx264", "veryfast", ["-tune", "zerolatency"] - except Exception as e: - print(f"[WARN] NVENC 检测失败，回退软件编码: {e}") - return "libx264", "veryfast", ["-tune", "zerolatency"] + 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( @@ -78,18 +70,20 @@ def start_douyin_youtube_ffplay( recording: set, recording_time_list: dict, running_list: list, - monitoring: list, # 可变列表，如 [count]，用于并发计数 + monitoring: list, clear_record_info, color_obj, ): """ - 抖音 → YouTube 多路推流主函数（支持 config/youtube.ini 中只配置 stream key） + 抖音 → YouTube 推流主函数（单路推流） + 只从 config/youtube.ini 读取一个 stream key（取第一个有效值） """ + # ====================== 读取 YouTube stream key 配置 ====================== config_file = "config/youtube.ini" rtmp_base = "rtmp://a.rtmp.youtube.com/live2/" - default_key = "qxvb-r47b-r5ju-6ud3-6k7z" - rtmp_targets = [] + default_key = "qxvb-r47b-r5ju-6ud3-6k7z" # 备用默认 key + youtube_rtmp = rtmp_base + default_key # 默认值 if os.path.exists(config_file): config = configparser.ConfigParser() @@ -97,21 +91,13 @@ def start_douyin_youtube_ffplay( if config.has_section("youtube"): for key_name, value in config.items("youtube"): value = value.strip() - if not value: - continue - if value.startswith("rtmp://"): - rtmp_targets.append(value) - else: - rtmp_targets.append(rtmp_base + value) - - if not rtmp_targets: - rtmp_targets = [rtmp_base + default_key] - print(f"[WARN] 未在 {config_file} 中找到有效配置，使用默认 stream key 推流") - - num_targets = len(rtmp_targets) - print(f"[INFO] 加载了 {num_targets} 个 YouTube 推流目标") - - tee_outputs = "|".join(f"[f=flv:flvflags=no_duration_filesize]{url}" for url in rtmp_targets) + if value: + if value.startswith("rtmp://"): + youtube_rtmp = value + else: + youtube_rtmp = rtmp_base + value + print(f"[INFO] 从 {config_file} 读取 YouTube stream key: {youtube_rtmp}") + break # 只取第一个有效 key，保持单路推流（不加多路逻辑） # ====================== 常量配置 ====================== MAX_RETRY_DELAY = 90 @@ -122,9 +108,17 @@ def start_douyin_youtube_ffplay( STALE_OUTPUT_SECONDS = 40 END_KEYWORDS = [ - "connection reset by peer", "failed to write trailer", "invalid data found", - "server returned 403", "server returned 404", "ingestion error", "access denied", - "packet too short", "client disconnected", "live has ended", "no route to host", + "connection reset by peer", + "failed to write trailer", + "invalid data found", + "server returned 403", + "server returned 404", + "ingestion error", + "access denied", + "packet too short", + "client disconnected", + "live has ended", + "no route to host", "connection timed out", ] @@ -138,12 +132,26 @@ def start_douyin_youtube_ffplay( 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 ({num_targets} 路): {stream_title}") + print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}") - # ====================== 编码器检测 ====================== - codec_v, preset_v, extra_tune_params = _detect_nvenc() + # ====================== NVENC 检测 ====================== + codec_v = "libx264" + preset_v = "veryfast" + tune_param = ["-tune", "zerolatency"] + try: + if 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" + preset_v = "p5" + tune_param = ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"] + print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") + except Exception as e: + print(f"[WARN] NVENC 检测失败，回退软件编码: {e}") - # ====================== FFmpeg 命令（1080p + tee 多路） ====================== + # ====================== FFmpeg 命令（1080p 单路推流，原样保留） ====================== ffmpeg_command = [ "ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-re", "-stats", "-stats_period", "1", @@ -164,7 +172,7 @@ def start_douyin_youtube_ffplay( "-c:v", codec_v, "-preset", preset_v, - *extra_tune_params, + *tune_param, "-profile:v", "high", "-b:v", "5200k", "-maxrate", "5800k", "-bufsize", "10400k", @@ -180,109 +188,119 @@ def start_douyin_youtube_ffplay( "-af", "aresample=async=1:first_pts=0", "-flags", "+global_header", - "-f", "tee", tee_outputs + "-flvflags", "no_duration_filesize", + "-f", "flv", youtube_rtmp ] - # ====================== 主循环 ====================== + # ====================== 主重试循环（原样保留） ====================== proc = None - stderr_q = queue.Queue() + stderr_q = None reader_t = None retry_delay = INITIAL_RETRY_DELAY try: while True: - if retry_delay > INITIAL_RETRY_DELAY: - print(f"[WARN] 防风控等待 {retry_delay}s 后重连...") - time.sleep(retry_delay) + try: + if retry_delay > INITIAL_RETRY_DELAY: + print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...") + time.sleep(retry_delay) - _cleanup_proc(proc) - stderr_q = queue.Queue() + cleanup_proc(proc) + stderr_q = queue.Queue() - print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}") - proc = subprocess.Popen( - ffmpeg_command, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - bufsize=1, - universal_newlines=True - ) - proc.last_out_ts = time.time() + print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}") + proc = subprocess.Popen( + ffmpeg_command, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + universal_newlines=True + ) + 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() + 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() + start_time = last_frame_time = time.time() - while True: - try: + while True: try: - line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT) - except queue.Empty: - if time.time() - proc.last_out_ts > STALE_OUTPUT_SECONDS: - print("[WARN] 长时间无日志，强制重启 ffmpeg 进程") - _cleanup_proc(proc) - break - continue + 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) + break + continue - if line is None: - if proc.poll() is not None: - break - 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 + 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() + if line: + proc.last_out_ts = time.time() + print(f"[FFmpeg] {line}") + line_lower = line.lower() - if any(kw in line_lower for kw in END_KEYWORDS): - print("[ERROR] 检测到主播真实下播或被拒，停止推流") - raise StreamEnded() - - if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower: - last_frame_time = time.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 无新帧，判定主播下播") + if any(kw in line_lower for kw in END_KEYWORDS): + print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流") + cleanup_proc(proc) raise StreamEnded() - except StreamEnded: - raise - except Exception as e: - print(f"[ERROR] 监控循环异常: {e}") - traceback.print_exc() - retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) - break + if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower: + last_frame_time = time.time() - retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) - continue + 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 StreamEnded() + + except StreamEnded: + 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 Exception as e: + print(f"[ERROR] 外层异常: {e}") + traceback.print_exc() + retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) + continue - except StreamEnded: - print(f"[INFO] 主播已下播，停止 YouTube 转推 → {record_name}") - except KeyboardInterrupt: - print("[INFO] 用户中断，停止推流") except Exception as e: - print(f"[FATAL] 推流异常: {e}") + print(f"[FATAL] YouTube 推流最外层异常: {e}") traceback.print_exc() finally: - _cleanup_proc(proc) + cleanup_proc(proc) 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) # 与 douyin_srs_ffplay.py 保持一致 + 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) \ No newline at end of file + color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN) \ No newline at end of file