diff --git a/config/youtube.ini b/config/youtube.ini index ec9204a..60514c6 100644 --- a/config/youtube.ini +++ b/config/youtube.ini @@ -7,5 +7,5 @@ audio_bitrate = 128 fps = 30 gop = 60 copy_when_safe = true -resolution = +resolution = 1080x1920 single_mode = true \ No newline at end of file diff --git a/main.py b/main.py index 23ed698..d2af416 100644 --- a/main.py +++ b/main.py @@ -28,7 +28,7 @@ from src.utils import logger, Color, check_md5 from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path # --------------------- 基础环境 --------------------- -version = "relay-1.0.0" +version = "relay-1.0.1" os.environ['PATH'] = ffmpeg_path + os.pathsep + current_env_path script_path = os.path.split(os.path.realpath(sys.argv[0]))[0] config_file = f'{script_path}/config/config.ini' @@ -186,7 +186,6 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: else: logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。") port_info = {"anchor_name": "", "is_live": False} - elif (".m3u8" in url) or (".flv" in url): # 自定义直链 port_info = { @@ -214,7 +213,7 @@ def build_youtube_output() -> Tuple[str, list, list]: secure = _yt_get("youtube", "secure", fallback=False, cast=bool) ingest = _yt_get("youtube", "ingest", fallback="a", cast=str) - res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() + res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() # 例如 "1080x1920" bv = _safe_int(_yt_get("youtube", "bitrate", fallback=5000, cast=int), 300, 20000, 5000) ba = _safe_int(_yt_get("youtube", "audio_bitrate", fallback=128, cast=int), 64, 384, 128) @@ -223,10 +222,11 @@ def build_youtube_output() -> Tuple[str, list, list]: out_url = _build_youtube_rtmp_url(key, secure, ingest) - # —— 默认强制转码（最稳），如果想省 CPU 可把 -c:v libx264 改为 copy —— # + # —— 默认强制转码（稳） —— # v_args = [ "-c:v", "libx264", "-preset", "veryfast", + "-tune", "zerolatency", "-profile:v", "high", "-level:v", "4.2", "-pix_fmt", "yuv420p", @@ -243,34 +243,46 @@ def build_youtube_output() -> Tuple[str, list, list]: "-ac", "2", ] - # 可选缩放/留黑边，保持 YT 吃得更稳 + # 可选缩放/留黑边 + 避免 1088x1920（加 crop 强制精确尺寸） if res_str: try: w, h = map(int, res_str.lower().split("x")) - vf = f"scale={w}:{h}:force_original_aspect_ratio=decrease,pad={w}:{h}:(ow-iw)/2:(oh-ih)/2,format=yuv420p" + vf = ( + f"scale={w}:{h}:force_original_aspect_ratio=decrease," + f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2," + f"crop={w}:{h}," + f"format=yuv420p" + ) v_args = ["-vf", vf] + v_args except Exception: logger.warning(f"resolution '{res_str}' 非法，忽略") return out_url, v_args, a_args -def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = None) -> subprocess.Popen: +def launch_ffmpeg(input_url: str, + out_url: str, + extra_headers: Optional[str] = None, + v_args: Optional[list] = None, + a_args: Optional[list] = None) -> subprocess.Popen: """ - 启动一个 ffmpeg 进程：输入 input_url，输出 RTMP 到 out_url。 + 启动一个 ffmpeg 进程：输入 input_url，输出 RTMP/RTMPS 到 out_url。 + - v_args / a_args 由 build_youtube_output() 生成并传入 + - 启动 stdout reader 线程以防缓冲阻塞 + - 设置代理环境变量以兼容 https/hls """ import threading - user_agent = ("Mozilla/5.0 (Linux; Android 11; SM-G973U) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/87.0.4280.141 Mobile Safari/537.36") - # —— 输入侧容错/重连/时间戳修复 —— - rw_timeout = "15000000" - analyzeduration = "20000000" - probesize = "10000000" - + # 输入侧容错/重连/探测 base = [ "ffmpeg", "-nostdin", "-hide_banner", + "-stats", # 输出统计信息 + "-stats_period", "60", # 每60秒一次 + "-progress", "pipe:1", # 状态写到stdout，便于reader捕获 + "-loglevel", "info", + # 输入容错 & 自动重连 "-rw_timeout", "15000000", "-timeout", "10000000", @@ -279,68 +291,69 @@ def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = N "-reconnect_at_eof", "1", "-reconnect_delay_max", "30", - # User-Agent 和协议 + # UA/协议白名单 "-user_agent", user_agent, "-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy", - # 缓冲 & 探测优化 + # 缓冲 & 探测 "-thread_queue_size", "4096", "-analyzeduration", "10000000", "-probesize", "10000000", + ] - # 输入流 - "-re", "-i", input_url, - # 容错标志：防止丢包、断音直接崩溃 + # 平台 headers（referer/origin）要在 -i 之前 + if extra_headers: + base += ["-headers", extra_headers] + + # 代理：http_proxy/https_proxy 环境变量（同时保留 -http_proxy 兼容） + env = os.environ.copy() + if proxy_addr: + env["http_proxy"] = proxy_addr + env["https_proxy"] = proxy_addr + base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:] + + # 输入 + base += ["-re", "-i", input_url] + + # 通用容错 + base += [ "-fflags", "+genpts+discardcorrupt+nobuffer", "-use_wallclock_as_timestamps", "1", - - # 视频编码 - "-c:v", "libx264", - "-preset", "veryfast", - "-tune", "zerolatency", - "-profile:v", "high", - "-level:v", "4.1", - "-pix_fmt", "yuv420p", - "-r", "30", - "-g", "60", - "-b:v", "5000k", - "-maxrate", "5000k", - "-bufsize", "10000k", - "-vf", "scale=1080:1920:force_original_aspect_ratio=decrease,pad=1080:1920:(ow-iw)/2:(oh-ih)/2", - - # 音频编码 - "-c:a", "aac", - "-b:a", "128k", - "-ar", "44100", - "-ac", "2", - - # 输出到 YouTube - "-f", "flv", out_url, - "-fflags", "+genpts+discardcorrupt+nobuffer+flush_packets", "-avioflags", "direct", "-rtbufsize", "100M", ] + # 视频/音频编码参数（使用 build_youtube_output 的结果） + v_args = v_args or [] + a_args = a_args or [] + base += v_args + a_args - # 平台需要自定义 header（如 referer/origin） - if extra_headers: - # 插到 -user_agent 后、-thread_queue_size 前 - i1 = base.index("-user_agent") + 2 - i2 = base.index("-thread_queue_size") - base[i1:i1] = ["-headers", extra_headers] + # 输出（flv 到 YT） + base += ["-f", "flv", out_url] - # 代理 - if proxy_addr: - base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:] + # 打印命令便于排障（截断） + cmd_str = " ".join(base) + print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...") - # 打印命令便于排障 - print("ffmpeg cmd:", " ".join(base if len(" ".join(base)) < 500 else base[:30] + ["..."])) + logger.info(f"启动推流：{input_url} -> {out_url}") - logger.info(f"启动推流：{input_url} -> YouTube") - p = subprocess.Popen(base, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1) + creationflags = 0 + if os.name == "nt": + # 防止子进程挂住控制台（可选） + creationflags = subprocess.CREATE_NEW_PROCESS_GROUP - # —— 启动 watchdog：若 20s 没有任何输出，判定卡死，主循环会重启它 —— + p = subprocess.Popen( + base, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + bufsize=0, # 避免行缓冲告警 & 尽快刷新 + env=env, + creationflags=creationflags + ) + + # 最近输出时间戳，用于 watchdog p.last_out_ts = time.time() def _reader(): @@ -348,12 +361,19 @@ def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = N for line in iter(p.stdout.readline, b''): if not line: break - p.last_out_ts = time.time() # 任何输出都视为“仍在工作” + text = line.decode(errors="ignore").strip() + p.last_out_ts = time.time() + # 常见错误提示 + if "Server disconnected" in text: + logger.error(f"[YouTube输出异常] {text}") + elif "not supported" in text or "Invalid" in text or "invalid" in text: + logger.error(f"[转码异常] {text}") except Exception: pass t = threading.Thread(target=_reader, daemon=True) t.start() + return p def stop_ffmpeg(p: Optional[subprocess.Popen]): @@ -367,9 +387,10 @@ def stop_ffmpeg(p: Optional[subprocess.Popen]): p.stdin.flush() except Exception: pass + p.wait(timeout=6) else: p.send_signal(signal.SIGINT) - p.wait(timeout=10) + p.wait(timeout=10) except Exception: try: p.kill() @@ -439,6 +460,10 @@ def main(): scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒 last_md5 = "" # URL 文件变化时热更新 + # watchdog 参数 + watchdog_grace = 60 # 推流开始 60s 内不做静默判定 + watchdog_silence = 300 # 5 分钟完全无输出才重启 + try: while True: # 动态热更新 URL 列表（URL_config.ini 有变更则重载） @@ -457,33 +482,22 @@ def main(): except Exception as e: logger.warning(f"URL 列表热更新失败：{e}") - - if current_proc and current_url: - # 已在推流：检测是否仍直播 & 进程状态 - if current_proc.poll() is not None: - color.print_colored(f"[{now_str()}] ffmpeg 进程退出，重启推流…", color.YELLOW) + retcode = current_proc.poll() + if retcode is not None: + color.print_colored(f"[{now_str()}] ffmpeg 已退出 (code={retcode})，切源…", color.YELLOW) stop_ffmpeg(current_proc) - current_proc = launch_ffmpeg(real_url, out_url, headers) - current_proc.last_out_ts = time.time() + current_proc = None + current_url = None continue - # 检测 ffmpeg 是否卡住（10 秒无输出） - no_out_for = time.time() - getattr(current_proc, "last_out_ts", time.time()) - if no_out_for > 20: # ✅ 改成 20 秒 - color.print_colored(f"\n[{now_str()}] ffmpeg 20s 无输出，尝试重启当前源…", color.YELLOW) - stop_ffmpeg(current_proc) - current_proc = launch_ffmpeg(real_url, out_url, headers) - current_proc.last_out_ts = time.time() - continue - - # 轻量检测：仍直播？ + # 检查是否停播（依赖 spider） try: port = asyncio.run(fetch_port_info(current_url, record_quality_code)) still_live = bool(port.get("is_live")) except Exception as e: logger.error(f"检测当前源失败：{e}") - still_live = True # ⚠️ 出错时默认仍在播，避免误切 + still_live = True # 容错：无法判断时先认为还活着 if still_live: elapsed = int(time.time() - start_time) @@ -491,10 +505,21 @@ def main(): m, s = divmod(m, 60) dur_str = f"{h:02d}:{m:02d}:{s:02d}" print(f"\r[{now_str()}] 正在转播({dur_str})：{current_platform} | {current_anchor} | {current_url}", end="") + + # ✅ watchdog（推流开始 > 60s 且 5 分钟完全无输出） + silence = time.time() - getattr(current_proc, "last_out_ts", 0) + if elapsed >= watchdog_grace and silence > watchdog_silence: + color.print_colored(f"\n[{now_str()}] 超过5分钟未见 ffmpeg 输出，判定异常，重启…", color.YELLOW) + stop_ffmpeg(current_proc) + current_proc = None + current_url = None + time.sleep(1) + continue + time.sleep(check_interval_live) continue else: - color.print_colored(f"\n[{now_str()}] 主播停播，才切换下一个源…", color.YELLOW) + color.print_colored(f"\n[{now_str()}] 主播停播，切换下一个源…", color.YELLOW) stop_ffmpeg(current_proc) current_proc = None current_url = None @@ -532,8 +557,8 @@ def main(): headers = headers_for_platform(current_platform, url) # 启动 ffmpeg - out_url, _, _ = build_youtube_output() - current_proc = launch_ffmpeg(real_url, out_url, headers) + out_url, v_args, a_args = build_youtube_output() + current_proc = launch_ffmpeg(real_url, out_url, headers, v_args=v_args, a_args=a_args) current_url = url start_time = time.time() color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN)