# 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 configparser class StreamEnded(Exception): """主播真实下播的专用异常""" pass 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, ): """ 抖音 → 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" # 备用默认 key youtube_rtmp = rtmp_base + default_key # 默认值 if os.path.exists(config_file): config = configparser.ConfigParser() config.read(config_file, encoding="utf-8") if config.has_section("youtube"): for key_name, value in config.items("youtube"): value = value.strip() 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 INITIAL_RETRY_DELAY = 3 STDERR_QUEUE_TIMEOUT = 1.0 NO_FRAME_TIMEOUT = 120 START_CHECK_AFTER = 25 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 timed out", ] 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}") # ====================== 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}") # 720p # ffmpeg_command = [ # "ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-re", # "-stats", "-stats_period", "1", # "-rw_timeout", "30000000", # "-reconnect", "1", "-reconnect_at_eof", "1", "-reconnect_streamed", "1", # "-reconnect_delay_max", "5", # "-fflags", "+genpts+discardcorrupt+nobuffer+flush_packets", # "-flags", "low_delay", # "-err_detect", "ignore_err", # "-max_delay", "80000", # "-thread_queue_size", "2048", # "-headers", douyin_headers, # "-i", real_url, # # 720p # "-vf", "fps=30,scale=720:1280:force_original_aspect_ratio=decrease:flags=lanczos,pad=720:1280:(ow-iw)/2:(oh-ih)/2:black", # "-c:v", "libx264", # "-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", # "-vsync", "cfr", # "-g", "60", "-keyint_min", "60", # "-r", "30", # "-bf", "0", # "-sc_threshold", "0", # # "-nal-hrd", "cbr", # 改2：强制 CBR # # 720p # # "-x264-params", "nal-hrd=cbr:force-cfr=1:scenecut=0:filler=1", # 改3：简化 params，锁死 CBR # "-x264-params", "force-cfr=1:scenecut=0:open_gop=0:sync-lookahead=0", # "-pix_fmt", "yuv420p", # "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", # 改4：1080p可音频上 128k（YouTube 推荐） # "-af", "aresample=async=1:first_pts=0", # "-flags", "+global_header", # "-flvflags", "no_duration_filesize", # "-f", "flv", youtube_rtmp # ] # 1080p ffmpeg_command = [ "ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-re", "-stats", "-stats_period", "1", "-rw_timeout", "30000000", "-reconnect", "1", "-reconnect_at_eof", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "5", "-fflags", "+genpts+discardcorrupt+nobuffer+flush_packets", "-flags", "low_delay", "-err_detect", "ignore_err", "-max_delay", "80000", "-thread_queue_size", "2048", "-headers", douyin_headers, "-i", real_url, # 1080p "-vf","fps=30,scale=1080:1920:force_original_aspect_ratio=decrease:flags=lanczos,pad=1080:1920:(ow-iw)/2:(oh-ih)/2:black", "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-profile:v", "high", # 1080p "-b:v", "5200k", "-maxrate", "5800k", "-bufsize", "10400k", # 改1：2× bufsize，真 CBR "-vsync", "cfr", "-g", "60", "-keyint_min", "60", "-r", "30", "-bf", "0", "-sc_threshold", "0", # "-nal-hrd", "cbr", # 改2：强制 CBR # 1080p "-x264-params","force-cfr=1:scenecut=0:open_gop=0:sync-lookahead=0", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-ar", "44100", "-ac", "2", "-af", "aresample=async=1:first_pts=0", "-flags", "+global_header", "-flvflags", "no_duration_filesize", "-f", "flv", youtube_rtmp ] # ====================== 主重试循环（原样保留） ====================== proc = None stderr_q = None reader_t = None retry_delay = INITIAL_RETRY_DELAY 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}") 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() start_time = last_frame_time = time.time() 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) 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 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] 检测到主播真下播或被明确拒绝，停止推流") cleanup_proc(proc) 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 无新帧，判定主播已下播") 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 Exception as e: print(f"[FATAL] YouTube 推流最外层异常: {e}") traceback.print_exc() finally: 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) clear_record_info(record_name, real_url) except Exception: pass color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)