diff --git a/backup/7777 b/backup/7777 new file mode 100644 index 0000000..b6c717d --- /dev/null +++ b/backup/7777 @@ -0,0 +1,203 @@ +else: + # ====================== 抖音→YouTube 2025（终极精简版） ====================== + import queue + import time + import datetime + import subprocess + import threading + import unicodedata + import platform + import re + import traceback + + # --------------------- 配置区（可外部传入） --------------------- + YT_STREAM_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" + YOUTUBE_RTMP = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}" + + MAX_RETRY_DELAY = 90 + INITIAL_RETRY_DELAY = 3 + NO_FRAME_TIMEOUT = 120 # 超过多久无新帧视为下播 + START_CHECK_AFTER = 25.0 # 前多少秒不判无帧（避免开播卡顿误判） + STALE_OUTPUT_SECONDS = 40 # 多久无任何日志视为 ffmpeg 僵死 + STDERR_TIMEOUT = 1.0 + + 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 timed out", "connection refused" + ] + + 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" + ) + + # --------------------- 工具函数 --------------------- + def sanitize_title(text: str, max_len: int = 120) -> str: + if not text: + return "" + text = unicodedata.normalize("NFKC", text) + text = ''.join(c for c in text if unicodedata.category(c)[0] != 'C') + text = ''.join(c for c in text if ord(c) <= 0xFFFF) + return text.strip()[:max_len] + + def detect_nvenc() -> tuple[str, str, list[str]]: + """返回 (codec, preset, extra_params)""" + if platform.system().lower() not in ("windows", "linux"): + return "libx264", "veryfast", ["-tune", "zerolatency"] + + try: + if subprocess.run(["nvidia-smi"], capture_output=True, timeout=5).returncode != 0: + return "libx264", "veryfast", ["-tune", "zerolatency"] + + result = subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], + capture_output=True, text=True, timeout=8 + ) + if "h264_nvenc" in result.stdout: + print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") + return "h264_nvenc", "p5", ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"] + except Exception as e: + print(f"[WARN] NVENC 检测失败，回落软件编码: {e}") + return "libx264", "veryfast", ["-tune", "zerolatency"] + + def build_ffmpeg_command(real_url: str): + codec_v, preset_v, tune_params = detect_nvenc() + return [ + "ffmpeg", "-y", "-loglevel", "info", + "-rw_timeout", "15000000", + "-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_params, + "-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 + ] + + # --------------------- 标题生成 --------------------- + timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) + clean_title = sanitize_title(title_in_name) + stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}") + print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}") + + # --------------------- 主逻辑（单层循环 + 单一 finally） --------------------- + proc = None + reader_thread = None + should_exit = False # 替代 StreamEnded 异常 + retry_delay = INITIAL_RETRY_DELAY + + def cleanup_resources(): + """统一清理（只调用一次）""" + nonlocal proc, reader_thread + if proc and proc.poll() is None: + try: proc.terminate(); proc.wait(timeout=5) + except: proc.kill() + if reader_thread and reader_thread.is_alive(): + reader_thread.join(timeout=1) + + # 全局状态清理（只执行一次） + try: + recording.discard(record_name) + recording_time_list.pop(record_name, None) + if record_url in running_list: + running_list.remove(record_url) + global monitoring + monitoring = max(0, monitoring - 1) + clear_record_info(record_name, record_url) + except: pass + + try: + ffmpeg_cmd = build_ffmpeg_command(real_url) + + while not should_exit: + if retry_delay > INITIAL_RETRY_DELAY: + print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接...") + time.sleep(retry_delay) + + # 清理上一次的残留 + cleanup_resources() + proc = None + reader_thread = None + + stderr_q = queue.Queue() + + print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}") + proc = subprocess.Popen( + ffmpeg_cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + universal_newlines=True + ) + reader_thread = threading.Thread( + target=lambda p=proc, q=stderr_q: ( + [q.put(l) for l in iter(p.stderr.readline, "")] or q.put(None) + ) + ) + reader_thread.daemon = True + reader_thread.start() + + start_time = last_frame_time = last_log_time = time.time() + + while True: + try: + line = stderr_q.get(timeout=STDERR_TIMEOUT) + except queue.Empty: + # 超时检查僵死 + if time.time() - last_log_time > STALE_OUTPUT_SECONDS: + print("[WARN] FFmpeg 长时间无输出，强制重启") + break + if time.time() - start_time > START_CHECK_AFTER and time.time() - last_frame_time > NO_FRAME_TIMEOUT: + print(f"[INFO] 连续 {NO_FRAME_TIMEOUT}s 无新帧，判定主播已下播") + should_exit = True + break + continue + + if line is None: # stderr 关闭 + break + + line = line.rstrip() + if line: + last_log_time = time.time() + print(f"[FFmpeg] {line}") + + lower = line.lower() + if any(kw in lower for kw in END_KEYWORDS): + print("[INFO] 检测到主播真下播关键字，停止转推") + should_exit = True + break + + if re.search(r'frame=\s*\d+', lower) or "fps=" in lower: + last_frame_time = time.time() + + # 内层循环退出：判断是否真下播 + if should_exit: + break + + # 网络波动等情况 → 指数退避重试 + retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) + + # --------------------- 正常退出 --------------------- + color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN) + + except Exception as e: + print(f"[FATAL] YouTube 推流意外崩溃: {e}") + traceback.print_exc() + color_obj.print_colored(f"[{record_name}] YouTube 推流异常退出但已清理\n", color_obj.RED) + + finally: + # 无论如何只清理一次 + cleanup_resources() + return \ No newline at end of file diff --git a/main.py b/main.py index b7259c9..447db3d 100644 --- a/main.py +++ b/main.py @@ -1529,7 +1529,9 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: error_window.append(1) else: - # ====================== 抖音→YouTube 2025 终极永不死版（已修复所有已知致命问题） ====================== + + + # ====================== 抖音→YouTube 2025（重构：单层重试 + 统一清理） ====================== import queue, unicodedata, platform, traceback class StreamEnded(Exception): @@ -1546,12 +1548,19 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: try: while True: line = proc.stderr.readline() - if not line: break + if not line: + break q.put(line) except Exception as e: - q.put(f"__ERR__READER__:{e}") + try: + q.put(f"__ERR__READER__:{e}") + except Exception: + pass finally: - q.put(None) + try: + q.put(None) + except Exception: + pass def cleanup_proc(proc): try: @@ -1561,29 +1570,24 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: proc.wait(timeout=5) except Exception: if proc.poll() is None: - proc.kill() + try: + proc.kill() + except Exception: + pass except Exception: pass - # ========== 参数配置 ========== - # YT_STREAM_KEY = "ue78-1c3e-mr9g-14mz-9r4z" # 没有ypp的美食 - - - YT_STREAM_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" # vlog频道 + # ========== 参数配置（保留你原始配置） ========== + YT_STREAM_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" youtube_rtmp = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}" - # 内网SRS服务器 - # YT_STREAM_KEY = "douyin" # vlog频道 - # youtube_rtmp = f"rtmp://192.168.31.184/live/{YT_STREAM_KEY}" - MAX_RETRY_DELAY = 90 INITIAL_RETRY_DELAY = 3 - NO_FRAME_TIMEOUT = 120 # 120秒无新帧才判真下播（完美契合YouTube 25~30秒容忍） + NO_FRAME_TIMEOUT = 120 START_CHECK_AFTER = 25.0 - STALE_OUTPUT_SECONDS = 40 # 40秒无任何日志才重启（更宽容） + STALE_OUTPUT_SECONDS = 40 STDERR_QUEUE_TIMEOUT = 1.0 - # 2025年12月实测最准的下播关键词（一个都不能少！） END_KEYWORDS = [ "connection reset by peer", "failed to write trailer", @@ -1599,13 +1603,12 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: "connection timed out", ] - # 标题清洗 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 检测 + # NVENC 检测（原样保留） codec_v, preset_v, tune_param = "libx264", "veryfast", ["-tune", "zerolatency"] try: if platform.system().lower() in ["windows", "linux"]: @@ -1618,14 +1621,13 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: except Exception as e: print(f"[WARN] NVENC 检测失败，回退软件编码: {e}") - # 伪装 headers + # headers & ffmpeg command（保持原来参数） 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 命令（关键！bufsize=1） ffmpeg_command = [ "ffmpeg", "-loglevel", "info", "-rw_timeout", "15000000", @@ -1646,8 +1648,10 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: "-f", "flv", youtube_rtmp ] + # ---- 单层重试循环（清晰可控） ---- proc = None - stderr_q = queue.Queue() + stderr_q = None + reader_t = None retry_delay = INITIAL_RETRY_DELAY try: @@ -1657,7 +1661,15 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...") time.sleep(retry_delay) - cleanup_proc(proc) + # 每次重连前先清理上一次的进程/线程 + try: + cleanup_proc(proc) + except Exception: + pass + if stderr_q: + # 垃圾化旧队列以便 GC + stderr_q = None + stderr_q = queue.Queue() print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}") proc = subprocess.Popen( @@ -1666,33 +1678,37 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, - bufsize=1, # 关键！必须是 1 + bufsize=1, universal_newlines=True ) - proc.last_out_ts = time.time() # 给 proc 加个属性记录最后输出时间 + proc.last_out_ts = time.time() + # 启动 stderr reader 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() + # ---- 监控循环（单一职责：处理 stderr 日志并决定是否结束/重连） ---- while True: try: try: line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT) except queue.Empty: - # 超时检查僵死 - if time.time() - proc.last_out_ts > STALE_OUTPUT_SECONDS: + # 超时检查：如果太久没有任何日志，判为僵死，重启 proc（跳出内层，进行重试） + if time.time() - getattr(proc, "last_out_ts", time.time()) > STALE_OUTPUT_SECONDS: print("[WARN] 长时间无任何日志，强制重启进程") cleanup_proc(proc) break continue - if line is None: # reader 结束 + if line is None: + # reader 线程结束；如果 proc 已退出，跳出去重试 if proc.poll() is not None: break - continue + else: + continue line = line.strip() if line.startswith("__ERR__READER__"): @@ -1700,22 +1716,21 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: continue if line: - proc.last_out_ts = time.time() # 任意日志都更新时间 + 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] 检测到主播真下播或被明确拒绝，停止推流") cleanup_proc(proc) raise StreamEnded() - # 帧检测（只更新 last_frame_time） + # 帧检测：更新 frame 时间戳 if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower: last_frame_time = time.time() - # 120秒无新帧 = 真下播（完美契合 YouTube 25~30秒容忍） + # 超时无新帧 => 认为下播 if time.time() - start_time > START_CHECK_AFTER: if time.time() - last_frame_time > NO_FRAME_TIMEOUT: print(f"[WARN] 连续 {NO_FRAME_TIMEOUT}s 无新帧，判定主播已下播") @@ -1723,43 +1738,71 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None: 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 + break # 跳出监控循环 -> 外层 while 会继续并重试 + + # 如果监控循环走到这里并不是因为 StreamEnded（例如 break），则重试连接（loop continue） + # 将在 while True 的下一次循环中进行等待并重连 + continue except StreamEnded: + # 真实的主播下播，停止并彻底退出本录制线程 print(f"[INFO] 主播已下播，停止 YouTube 转推 → {record_name}") break + except Exception as e: + # 外层致命异常：打印并退出循环（会在 finally 里做统一清理） print(f"[ERROR] 外层异常: {e}") traceback.print_exc() retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) + # 等待并尝试重连（除非你希望直接退出，这里保留重试策略） continue - finally: - cleanup_proc(proc) - # ============ 彻底清理 ============ + # ============ 正常或 StreamEnded 后的统一清理（在这里 break 到这里） ============ cleanup_proc(proc) - recording.discard(record_name) - recording_time_list.pop(record_name, None) - if record_url in running_list: - running_list.remove(record_url) - global monitoring - monitoring = max(0, monitoring - 1) - clear_record_info(record_name, record_url) + # 以下清理放在 try/except 里以防止清理过程出错导致漏逻辑 + try: + recording.discard(record_name) + recording_time_list.pop(record_name, None) + if record_url in running_list: + running_list.remove(record_url) + global monitoring + monitoring = max(0, monitoring - 1) + clear_record_info(record_name, record_url) + except Exception: + pass color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN) + return except Exception as e: + # 最外层异常也要统一清理（保证不会留下僵尸状态） print(f"[FATAL] YouTube 推流最外层异常: {e}") traceback.print_exc() - cleanup_proc(proc) + try: + cleanup_proc(proc) + except Exception: + pass + try: + recording.discard(record_name) + recording_time_list.pop(record_name, None) + if record_url in running_list: + running_list.remove(record_url) + # global monitoring + monitoring = max(0, monitoring - 1) + clear_record_info(record_name, record_url) + except Exception: + pass color_obj.print_colored(f"[{record_name}] YouTube 推流异常退出但已清理\n", color_obj.RED) + return count_time = time.time()