From 350eb6d7984c269f59dfcdef60260323d7c086e0 Mon Sep 17 00:00:00 2001 From: eric Date: Mon, 1 Dec 2025 18:36:38 +0800 Subject: [PATCH] 's' --- main.py | 456 +++++++++++++++----------------------------------------- 1 file changed, 120 insertions(+), 336 deletions(-) diff --git a/main.py b/main.py index 40630a0..3cfc049 100644 --- a/main.py +++ b/main.py @@ -1530,372 +1530,156 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: else: - import platform - # ====================== 抖音→YouTube 转推 2025 终极永不死版（已修复所有已知Bug） ====================== - # 功能：把当前抖音直播实时无损转推到 YouTube，做到真正“永不死” - # 实测：2025年12月，单线程一天重启 ≤ 2 次（大多数情况 0 次） + # ====================== 终极永不死 YouTube 转推增强版 ====================== - # ======= 以下为替换用的稳定版 YouTube 转推核心逻辑 ======= - import threading - import queue - import shlex - import html - import unicodedata - import subprocess - import traceback + import queue,traceback, unicodedata, platform - # 自定义异常，表示源端已结束或明确拒绝（不使用 SystemExit） class StreamEnded(Exception): pass - # 辅助：清洗标题（移除不可见与 emoji） def sanitize_title(text: str, max_len: int = 120) -> str: - if not text: - return "" - # 正常化（去控制字符） + 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) - text = text.strip() - if len(text) > max_len: - text = text[:max_len] - return text + return text.strip()[:max_len] - # stderr 读取线程函数：逐行读 stderr 并发送到队列 def _stderr_reader_thread(proc, q): try: for raw in iter(proc.stderr.readline, ''): - if not raw: - break + if not raw: break q.put(raw) - except Exception as _e: - # 读取线程出错也不要崩溃主线程 - q.put(f"__ERR__READER__:{_e}") + except Exception as e: + q.put(f"__ERR__READER__:{e}") finally: - q.put(None) # 结束信号 + q.put(None) - try: - - # 可调参数 - YT_STREAM_KEY = "x04z-564w-aks7-embw-30y4" # 请替换为你自己的 key - youtube_rtmp = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}" - MAX_RETRY_DELAY = 90 - INITIAL_RETRY_DELAY = 3 - NO_FRAME_THRESHOLD = 5 # 连续无帧次数判定下播 - NO_FRAME_INTERVAL = 12.0 # 无帧计数阈值（秒） - START_CHECK_AFTER = 25.0 # 启动后多久开始无帧检测（秒） - STALE_OUTPUT_SECONDS = 20.0 # stderr 无输出判定僵死（秒） - STDERR_QUEUE_TIMEOUT = 1.0 # 从 stderr 队列读取超时（秒） - - # 下播/致命错误关键字（极低误判版本，只判断真正的“结束/拒绝”） - END_KEYWORDS = [ - "live has ended", - "ingestion error", - "主播已下播", - "房间已关闭", - ] - - - - - # 开始构造 stream_title（并清洗） - timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) - title_suffix = f"_{title_in_name}" if title_in_name else "" - _clean_title_in_name = sanitize_title(title_in_name) - stream_title = sanitize_title(f"{anchor_name}{('_' + _clean_title_in_name) if _clean_title_in_name else ''}_{timestamp_str}") - - print(f"{rec_info} 开始推流到 YouTube: {stream_title}") - - # NVENC 检测（保持，但加更短 timeout 并安全捕获） - 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=3) - if check.returncode == 0: - enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=3) - 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("检测到 NVIDIA GPU → 使用 NVENC 硬件加速（质量更高更省CPU）") - except Exception: - # 任何检测失败都安全回退到软件编码 - pass - - # 伪装 headers（保持） - 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" - ) - - # 构造 ffmpeg 命令（保持画面处理与参数） - ffmpeg_command = [ - "ffmpeg", - "-rw_timeout", "10000000", # 10s，减少长等待（可按需微调） - "-reconnect", "1", - "-reconnect_at_eof", "1", - "-reconnect_streamed", "1", - "-reconnect_delay_max", "15", - "-fflags", "+genpts+discardcorrupt", - "-err_detect", "ignore_err", - "-thread_queue_size", "8192", - "-headers", douyin_headers, - "-i", real_url, - "-vf", "scale=720:1280:force_original_aspect_ratio=decrease,pad=720:1280:(ow-iw)/2:(oh-ih)/2:black", - "-c:v", codec_v, "-preset", preset_v, *tune_param, - "-b:v", "2500k", "-maxrate", "2500k", "-bufsize", "5000k", - "-g", "50", "-r", "25", "-pix_fmt", "yuv420p", - "-c:a", "aac", "-b:a", "128k", "-ar", "44100", "-ac", "2", - "-af", "aresample=async=1:first_pts=0", - "-flags", "+global_header", - "-f", "flv", youtube_rtmp - ] - - # 用队列和线程安全读取 stderr（避免 readline 阻塞） - stderr_q = queue.Queue() - proc = None - retry_delay = INITIAL_RETRY_DELAY - - # 外层循环：控制重连（区分真正下播与其他异常） - while True: - try: - if retry_delay > INITIAL_RETRY_DELAY: - print(f"【防风控】等待 {retry_delay}s 后重新连接抖音源...") - time.sleep(retry_delay) - - # 清理残留进程 - if proc and proc.poll() is None: - try: - proc.terminate() - proc.wait(timeout=5) - except Exception: - try: - if proc and proc.poll() is None: - proc.kill() - except Exception: - pass - - print(f"[{datetime.datetime.now():%H:%M:%S}] 启动YouTube推流 → {stream_title}") - # 启动 ffmpeg（将 stdout 设为 DEVNULL，stderr 用管道） - proc = subprocess.Popen( - ffmpeg_command, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - bufsize=1, - universal_newlines=True - ) - - # 启动 stderr 读取线程 - reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q)) - reader_t.daemon = True - reader_t.start() - - # 监控变量 - proc.last_out_ts = time.time() - start_time = time.time() - last_frame_time = time.time() - no_frame_count = 0 - retry_delay = INITIAL_RETRY_DELAY # 成功连接后恢复最小延迟 - - # 内层循环：从 queue 读取 stderr 并做判断；若超时也检查僵死 - while True: - try: - item = None - try: - item = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT) - except queue.Empty: - # 没有新 stderr 行，检查僵死 - if time.time() - proc.last_out_ts > STALE_OUTPUT_SECONDS: - print("【僵死检测】长时间无 stderr 输出，强制重启进程") - # 强制重启 - if proc and proc.poll() is None: - try: - proc.kill() - except Exception: - pass - break - # 继续等待 - continue - - # 读取到 None = reader 结束（proc stderr EOF） - if item is None: - # 如果进程已经退出，跳出内层 - if proc.poll() is not None: - break - else: - # 没有更多 stderr 但进程还活着：短暂等待 - time.sleep(0.5) - continue - - # 读取错误行 - line = str(item).strip() - proc.last_out_ts = time.time() - - # 出现读取线程异常标记 - if line.startswith("__ERR__READER__"): - print(f"FFmpeg stderr reader 异常: {line}") - # 作为非致命警告，继续监控 - continue - - if line: - # 简单解析 frame/fps 更新 last_frame_time（适配各种 ffmpeg 输出） - line_lower = line.lower() - # 使用更稳妥的 frame 匹配，不使用 \b - if "frame=" in line_lower or "fps=" in line_lower or re.search(r'frame=\s*\d+', line_lower): - last_frame_time = time.time() - no_frame_count = 0 - print(f"FFmpeg: {line}") - - # 致命下播/拒绝关键词检测（包含 http/rtmp 错误） - # 更严格匹配 HTTP code 或关键短语，避免误判 FFmpeg 的正常输出 - lowered = line_lower - if re.search(r'\b(401|403|404|410|502|503|504)\b', lowered) or any(kw in lowered for kw in END_KEYWORDS): - print("【致命错误】检测到可能的下播或访问被拒，停止YouTube推流") - # 彻底结束（外层判断 StreamEnded） - if proc and proc.poll() is None: - try: - proc.kill() - except Exception: - pass - raise StreamEnded() - - # 启动 START_CHECK_AFTER 秒后开始无帧检测 - if time.time() - start_time > START_CHECK_AFTER: - if time.time() - last_frame_time > NO_FRAME_INTERVAL: - no_frame_count += 1 - print(f"【YouTube救命】{time.time()-last_frame_time:.1f}s 无新帧，连续 {no_frame_count} 次") - if no_frame_count >= NO_FRAME_THRESHOLD: - print("【真下播确认】连续无帧次数超过阈值，判定主播已下播") - if proc and proc.poll() is None: - try: - proc.kill() - except Exception: - pass - raise StreamEnded() - # 触发一次短重启以尝试恢复 - if proc and proc.poll() is None: - try: - proc.kill() - except Exception: - pass - break - - # 内层处理结束，继续循环 - except StreamEnded: - # 主播真下播或被拒，跳出全部并不再重连 - raise - except Exception as inner_e: - print(f"推流内部异常: {inner_e}") - traceback.print_exc() - # 将视作连接问题，准备指数退避后重连 - retry_delay = min(int(retry_delay * 2), MAX_RETRY_DELAY) - # 确保 proc 被杀 - try: - if proc and proc.poll() is None: - proc.kill() - except Exception: - pass - break # 跳出内层，回到外层重连逻辑 - - except StreamEnded: - # 真下播（来自上层 raise StreamEnded） - print(f"检测到主播已下播（或被明确拒绝），将停止当前 YouTube 转推线程 → {record_name}") - break # 彻底退出外层循环 - except Exception as e: - # 外层异常（网络/执行异常），指数退避重连 - print(f"推流进程异常: {e}") - traceback.print_exc() - retry_delay = min(retry_delay * 2 if retry_delay else INITIAL_RETRY_DELAY, MAX_RETRY_DELAY) - # 在异常后继续外层 while，等待 retry_delay 后重连 - continue - finally: - # 彻底清理 proc 与 reader（保证不会遗留子进程） - try: - if 'proc' in locals() and proc: - if proc.poll() is None: - try: - proc.terminate() - proc.wait(timeout=5) - except Exception: - try: - if proc.poll() is None: - proc.kill() - except Exception: - pass - except Exception: - pass - - # ============ 彻底退出清理（和你原先逻辑保持一致） ============ + def cleanup_proc(proc): try: if proc and proc.poll() is None: - try: - proc.kill() + try: proc.terminate(); proc.wait(timeout=5) except Exception: - pass + if proc.poll() is None: proc.kill() except Exception: pass - # 保持你原来的记录清理逻辑，但对 monitoring 做全局声明与安全处理 + # ========== 参数配置 ========== + YT_STREAM_KEY = "x04z-564w-aks7-embw-30y4" + youtube_rtmp = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}" + + MAX_RETRY_DELAY = 90 + INITIAL_RETRY_DELAY = 3 + NO_FRAME_INTERVAL = 20 + NO_FRAME_TIMEOUT = 120 + START_CHECK_AFTER = 25.0 + STALE_OUTPUT_SECONDS = 30 + STDERR_QUEUE_TIMEOUT = 1.0 + END_KEYWORDS = ["live has ended", "主播已下播", "房间已关闭"] + + # 构造标题 + timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) + _clean_title_in_name = sanitize_title(title_in_name) + stream_title = sanitize_title(f"{anchor_name}{('_' + _clean_title_in_name) if _clean_title_in_name else ''}_{timestamp_str}") + print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}") + + # NVENC 检测 + codec_v, preset_v, tune_param = "libx264", "veryfast", ["-tune", "zerolatency"] + try: + if platform.system().lower() in ["windows", "linux"]: + check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=3) + if check.returncode == 0: + enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=3) + if "h264_nvenc" in (enc.stdout or ""): + codec_v, preset_v, tune_param = "h264_nvenc", "p5", ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"] + print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") + except Exception: pass + + # FFmpeg 命令 + 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\nOrigin: https://live.douyin.com\r\n" + ) + + ffmpeg_command = [ + "ffmpeg", "-rw_timeout", "10000000", "-reconnect", "1", "-reconnect_at_eof", "1", + "-reconnect_streamed", "1", "-reconnect_delay_max", "15", "-fflags", "+genpts+discardcorrupt", + "-err_detect", "ignore_err", "-thread_queue_size", "8192", "-headers", douyin_headers, + "-i", real_url, + "-vf", "scale=720:1280:force_original_aspect_ratio=decrease,pad=720:1280:(ow-iw)/2:(oh-ih)/2:black", + "-c:v", codec_v, "-preset", preset_v, *tune_param, + "-b:v", "2500k", "-maxrate", "2500k", "-bufsize", "5000k", + "-g", "50", "-r", "25", "-pix_fmt", "yuv420p", + "-c:a", "aac", "-b:a", "128k", "-ar", "44100", "-ac", "2", "-af", "aresample=async=1:first_pts=0", + "-flags", "+global_header", "-f", "flv", youtube_rtmp + ] + + proc, stderr_q, retry_delay = None, queue.Queue(), INITIAL_RETRY_DELAY + + while True: try: - recording.discard(record_name) - except Exception: - pass - try: - recording_time_list.pop(record_name, None) - except Exception: - pass - try: - if record_url in running_list: + if retry_delay > INITIAL_RETRY_DELAY: print(f"[WARN] 防风控等待 {retry_delay}s 后重连...") + time.sleep(retry_delay) + cleanup_proc(proc) + + 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) + 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() + consecutive_failures = 0 + + while True: try: - running_list.remove(record_url) - except ValueError: - pass - except Exception: - pass + try: line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT) + except queue.Empty: + if time.time() - last_frame_time > STALE_OUTPUT_SECONDS: + print("[WARN] 长时间无 stderr 输出，重启进程") + cleanup_proc(proc); break + continue - # 确保 monitoring 在全局可用，安全减一 - try: - global monitoring - monitoring = max(0, monitoring - 1) if isinstance(monitoring, int) else 0 - except NameError: - monitoring = 0 + if line is None: + if proc.poll() is not None: break + else: time.sleep(0.5); continue - try: - clear_record_info(record_name, record_url) - except Exception: - pass + line = line.strip(); proc.last_out_ts = time.time() + if line.startswith("__ERR__READER__"): + print(f"[ERROR] stderr reader 异常: {line}"); continue - try: - color_obj.print_colored(f"[{record_name}] YouTube推流已彻底停止并清理完毕\n", color_obj.GREEN) - except Exception: - print(f"[{record_name}] YouTube推流已彻底停止并清理完毕") + print(f"[FFmpeg] {line}") + line_lower = line.lower() - # ======= 替换结束 ======= + if re.search(r'\b(401|403|404|410|429|502|503|504)\b', line_lower) or any(kw in line_lower for kw in END_KEYWORDS): + print("[ERROR] 检测到下播或访问被拒，停止推流"); cleanup_proc(proc); raise StreamEnded() - except Exception as e: - print(f"YouTube推流最外层异常: {e}") - traceback.print_exc() - # 修正这里的判断：如果 proc 存在且仍在运行才 kill - try: - if 'proc' in locals() and proc and proc.poll() is None: - try: - proc.kill() - except Exception: - pass - except Exception: - pass - try: - color_obj.print_colored(f"[{record_name}] YouTube推流异常退出但已清理\n", color_obj.RED) - except Exception: - print(f"[{record_name}] YouTube推流异常退出但已清理\n") + # frame 更新 last_frame_time + if time.time() - start_time > START_CHECK_AFTER and "frame=" in line_lower: + last_frame_time = time.time() + if time.time() - last_frame_time > NO_FRAME_TIMEOUT: + print("[WARN] 无帧超时，判定主播已下播"); cleanup_proc(proc); raise StreamEnded() + + except StreamEnded: raise + except Exception as inner_e: + print(f"[ERROR] 内部异常: {inner_e}"); traceback.print_exc() + cleanup_proc(proc) + consecutive_failures += 1 + retry_delay = min(INITIAL_RETRY_DELAY * (2 ** consecutive_failures), MAX_RETRY_DELAY) + break + + except StreamEnded: + print(f"[INFO] 主播已下播或明确拒绝 → {record_name} 推流停止") + break + except Exception as e: + print(f"[ERROR] 外层异常: {e}"); traceback.print_exc() + consecutive_failures += 1 + retry_delay = min(INITIAL_RETRY_DELAY * (2 ** consecutive_failures), MAX_RETRY_DELAY) + continue + finally: + cleanup_proc(proc) + + count_time = time.time()