# -*- encoding: utf-8 -*- """ 单进程 ffmpeg 推流：从 URL_config.ini 顺序检测直播源， 一旦当前源停播/异常，自动切到下一个正在直播的源并推流到 YouTube。 仅做“检测+转推”，不做录制、分段、通知等。 """ import asyncio import os import sys import signal import time import datetime import re import uuid import subprocess from pathlib import Path from typing import Any, List, Tuple, Optional import configparser import urllib.request from urllib.error import URLError, HTTPError import httpx from src import spider, stream from src.utils import logger, Color, check_md5 from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path # --------------------- 基础环境 --------------------- version = "relay-1.0.0" 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' url_config_file = f'{script_path}/config/URL_config.ini' text_encoding = 'utf-8-sig' color = Color() # ---------- YouTube 推流配置（读取 youtube.ini） ---------- _yt_cfg = configparser.ConfigParser() with open(f"{script_path}/config/youtube.ini", "r", encoding="utf-8-sig") as f: _yt_cfg.read_file(f) def _yt_get(section: str, key: str, fallback=None, cast=str): try: if cast is bool: return _yt_cfg.getboolean(section, key, fallback=fallback) if cast is int: return _yt_cfg.getint(section, key, fallback=fallback) return _yt_cfg.get(section, key, fallback=fallback) except Exception: return fallback def _safe_int(v, lo, hi, default): try: v = int(v) if v < lo or v > hi: return default return v except Exception: return default def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None): node = (ingest or "a").strip().lower() node = node if node.isalpha() and len(node) == 1 else "a" scheme = "rtmps" if secure else "rtmp" host = f"{node}.rtmps.youtube.com" if secure else f"{node}.rtmp.youtube.com" return f"{scheme}://{host}/live2/{key}" # --------------------- 简化配置读取 --------------------- def read_config_value(cfg: configparser.RawConfigParser, section: str, option: str, default_value: Any) -> Any: try: cfg.read(config_file, encoding=text_encoding) if section not in cfg.sections(): cfg.add_section(section) return cfg.get(section, option) except (configparser.NoSectionError, configparser.NoOptionError): cfg.set(section, option, str(default_value)) with open(config_file, 'w', encoding=text_encoding) as f: cfg.write(f) return default_value _CN_BOOL = {"是": True, "否": False} cfg = configparser.RawConfigParser() use_proxy = _CN_BOOL.get(read_config_value(cfg, '录制设置', '是否使用代理ip(是/否)', "否"), False) proxy_addr = read_config_value(cfg, '录制设置', '代理地址', "") proxy_addr = proxy_addr if use_proxy and proxy_addr.strip() else None # Cookie（按需） dy_cookie = read_config_value(cfg, 'Cookie', '抖音cookie', '') tiktok_cookie = read_config_value(cfg, 'Cookie', 'tiktok_cookie', '') # 可选：启动时检测系统代理（仅提示） try: if not use_proxy: urllib.request.urlopen("https://www.google.com/", timeout=5) print("提示：你的网络似乎直连可用（无需代理）。") except Exception: if not use_proxy: color.print_colored("提示：无法直连海外站，如需 TikTok 请在 config.ini 开启代理并填写代理地址。", color.YELLOW) # --------------------- URL 列表读取 --------------------- def parse_urls_from_file(path: str) -> List[str]: """ 读取 URL_config.ini，返回去重且保持相对顺序的 URL 列表。 支持行内格式： - 纯 URL - quality,url - url,主播: xxx - # 注释行 忽略 """ urls, seen = [], set() if not os.path.isfile(path): open(path, 'w', encoding=text_encoding).close() print("URL_config.ini 为空，请写入直播间地址后再运行。") return urls with open(path, "r", encoding=text_encoding, errors='ignore') as f: for line in f: s = line.strip() if not s or s.startswith("#"): continue # 拆分（最多3段，取包含 https 或域名的那段） parts = re.split(r"[，,]\s*", s) cand = None for p in parts: if '://' in p or re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", p): cand = p break if not cand: continue url = cand if '://' in cand else f"https://{cand}" if url not in seen: seen.add(url) urls.append(url) return urls # --------------------- 直播解析与选源 --------------------- def is_flv_preferred_platform(link: str) -> bool: # 抖音/TikTok 优先走 FLV；但 H265 不走 FLV return any(i in link for i in ["douyin", "tiktok"]) def select_source_url(link: str, stream_info: dict) -> Optional[str]: if not stream_info: return None if is_flv_preferred_platform(link): flv = stream_info.get("flv_url") if flv: # 若 codec=h265 则不走 flv try: from urllib.parse import urlparse, parse_qs qs = parse_qs(urlparse(flv).query or "") if (qs.get("codec") or [""])[0].lower() == "h265": logger.warning("FLV 不支持 h265，改用 HLS。") else: return flv except Exception: return flv return stream_info.get("record_url") or stream_info.get("m3u8_url") or stream_info.get("flv_url") async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: """ 调用项目现有 spider/stream，返回： { "anchor_name": str, "is_live": bool, "title": str?, "record_url": "...", "flv_url": "...", "m3u8_url": "..." } """ proxy = proxy_addr port_info = {} try: if "douyin.com/" in url: if 'v.douyin.com' not in url and '/user/' not in url: j = await spider.get_douyin_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie) else: j = await spider.get_douyin_app_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie) port_info = await stream.get_douyin_stream_url(j, record_quality_code, proxy) elif url.startswith("https://www.tiktok.com/"): if proxy: j = await spider.get_tiktok_stream_data(url=url, proxy_addr=proxy, cookies=tiktok_cookie) port_info = await stream.get_tiktok_stream_url(j, record_quality_code, proxy) else: logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。") port_info = {"anchor_name": "", "is_live": False} elif (".m3u8" in url) or (".flv" in url): # 自定义直链 port_info = { "anchor_name": "自定义_" + str(uuid.uuid4())[:8], "is_live": True, } if url.endswith(".flv"): port_info["flv_url"] = url else: port_info["m3u8_url"] = url else: # 其它站点可按需扩展 port_info = {"anchor_name": "", "is_live": False} except Exception as e: logger.error(f"解析失败: {e}") port_info = {"anchor_name": "", "is_live": False} return port_info # --------------------- ffmpeg 单进程推流 --------------------- def build_youtube_output() -> Tuple[str, list, list]: key = _yt_get("youtube", "key", fallback="", cast=str) if not key: logger.error("错误: config/youtube.ini 未配置 key") sys.exit(1) 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() 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) fps = _safe_int(_yt_get("youtube", "fps", fallback=30, cast=int), 10, 60, 30) gop = _safe_int(_yt_get("youtube", "gop", fallback=2*fps, cast=int), fps, 4*fps, 2*fps) out_url = _build_youtube_rtmp_url(key, secure, ingest) # —— 默认强制转码（最稳），如果想省 CPU 可把 -c:v libx264 改为 copy —— # v_args = [ "-c:v", "libx264", "-preset", "veryfast", "-profile:v", "high", "-level:v", "4.2", "-pix_fmt", "yuv420p", "-r", str(fps), "-g", str(gop), "-b:v", f"{bv}k", "-maxrate", f"{bv}k", "-bufsize", f"{max(2*bv, 4000)}k", ] a_args = [ "-c:a", "aac", "-b:a", f"{ba}k", "-ar", "44100", "-ac", "2", ] # 可选缩放/留黑边，保持 YT 吃得更稳 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" 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: """ 启动一个 ffmpeg 进程：输入 input_url，输出 RTMP 到 out_url。 """ 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", # 输入容错 & 自动重连 "-rw_timeout", "15000000", "-timeout", "10000000", "-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_at_eof", "1", "-reconnect_delay_max", "30", # User-Agent 和协议 "-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, # 容错标志：防止丢包、断音直接崩溃 "-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", ] # 平台需要自定义 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] # 代理 if proxy_addr: base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:] # 打印命令便于排障 print("ffmpeg cmd:", " ".join(base if len(" ".join(base)) < 500 else base[:30] + ["..."])) logger.info(f"启动推流：{input_url} -> YouTube") p = subprocess.Popen(base, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1) # —— 启动 watchdog：若 20s 没有任何输出，判定卡死，主循环会重启它 —— p.last_out_ts = time.time() def _reader(): try: for line in iter(p.stdout.readline, b''): if not line: break p.last_out_ts = time.time() # 任何输出都视为“仍在工作” except Exception: pass t = threading.Thread(target=_reader, daemon=True) t.start() return p def stop_ffmpeg(p: Optional[subprocess.Popen]): if not p: return try: if os.name == "nt": if p.stdin: try: p.stdin.write(b"q") p.stdin.flush() except Exception: pass else: p.send_signal(signal.SIGINT) p.wait(timeout=10) except Exception: try: p.kill() except Exception: pass # --------------------- 主循环（单进程转推+切源） --------------------- def check_ffmpeg_existence() -> bool: try: result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True) if result.returncode == 0: print(result.stdout.splitlines()[0]) except Exception: pass finally: return bool(check_ffmpeg()) def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def headers_for_platform(platform_name: str, live_url: str) -> Optional[str]: live_domain = '/'.join(live_url.split('/')[0:3]) record_headers = { 'PandaTV': 'origin:https://www.pandalive.co.kr', 'WinkTV': 'origin:https://www.winktv.co.kr', 'PopkonTV': 'origin:https://www.popkontv.com', 'FlexTV': 'origin:https://www.flextv.co.kr', '千度热播': 'referer:https://qiandurebo.com', '17Live': 'referer:https://17.live/en/live/6302408', '浪Live': 'referer:https://www.lang.live', 'shopee': f'origin:{live_domain}', 'Blued直播': 'referer:https://app.blued.cn' } return record_headers.get(platform_name) def detect_platform(url: str) -> str: if "douyin.com" in url: return "抖音直播" if "tiktok.com" in url: return "TikTok直播" if url.endswith(".flv") or url.endswith(".m3u8"): return "自定义录制直播" return "未知平台" def main(): print("-----------------------------------------------------") print("| DouyinLiveRecorder - 单进程转推版 |") print("-----------------------------------------------------") print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") print("模式：只检测 + 单进程推流（停播自动切下一个 URL）") print(".....................................................") if not check_ffmpeg_existence(): logger.error("缺少 ffmpeg，程序退出") sys.exit(1) urls = parse_urls_from_file(url_config_file) if not urls: logger.error("URL_config.ini 中没有可用 URL。") sys.exit(1) # 顺序轮询索引 idx = 0 current_url = None current_proc: Optional[subprocess.Popen] = None current_platform = "" current_anchor = "" record_quality_code = "OD" # 固定“原画”；需要也可改为从 config.ini 读取 check_interval_live = 8 # 推流中检测当前源是否仍在直播的间隔秒 scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒 last_md5 = "" # URL 文件变化时热更新 try: while True: # 动态热更新 URL 列表（URL_config.ini 有变更则重载） try: md5 = check_md5(url_config_file) if md5 != last_md5: urls = parse_urls_from_file(url_config_file) or urls last_md5 = md5 print(f"[{now_str()}] 已重载 URL 列表，数量：{len(urls)}") # 若当前 URL 已不在列表，强制切换 if current_url and current_url not in urls: print(f"[{now_str()}] 当前URL已被移除，停止并重新选择。") stop_ffmpeg(current_proc) current_proc = None current_url = None 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) stop_ffmpeg(current_proc) current_proc = launch_ffmpeg(real_url, out_url, headers) current_proc.last_out_ts = time.time() 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 # 轻量检测：仍直播？ 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 # ⚠️ 出错时默认仍在播，避免误切 if still_live: elapsed = int(time.time() - start_time) h, m = divmod(elapsed, 3600) 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="") time.sleep(check_interval_live) continue else: color.print_colored(f"\n[{now_str()}] 主播停播，才切换下一个源…", color.YELLOW) stop_ffmpeg(current_proc) current_proc = None current_url = None else: # 未在推流：从 urls 顺序找第一个 is_live 的 if not urls: time.sleep(scan_interval_when_idle) continue tried = 0 found = False start_idx = idx while tried < len(urls): url = urls[idx] idx = (idx + 1) % len(urls) tried += 1 port = asyncio.run(fetch_port_info(url, record_quality_code)) if not port.get("anchor_name"): print(f"[{now_str()}] 获取失败，跳过：{url}") continue if not port.get("is_live"): print(f"[{now_str()}] 未开播：{port.get('anchor_name','')} | {url}") continue # 选取真实播放地址 real_url = select_source_url(url, port) if not real_url: print(f"[{now_str()}] 未得到播放地址，跳过：{url}") continue current_platform = detect_platform(url) current_anchor = port.get("anchor_name", "") headers = headers_for_platform(current_platform, url) # 启动 ffmpeg out_url, _, _ = build_youtube_output() current_proc = launch_ffmpeg(real_url, out_url, headers) current_url = url start_time = time.time() color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN) found = True break if not found: # 没有任何源在播，过会儿再试 print(f"[{now_str()}] 当前无可用直播源，等待中…") time.sleep(scan_interval_when_idle) continue except KeyboardInterrupt: print("\n收到中断，正在退出…") except Exception as e: logger.error(f"主循环异常：{e}") finally: stop_ffmpeg(current_proc) print("已退出。") if __name__ == "__main__": main()