From 4c6b0408879462a8da318956e05646c018b20d52 Mon Sep 17 00:00:00 2001 From: eric Date: Sat, 27 Sep 2025 19:04:22 +0800 Subject: [PATCH] '-m' --- back/back4 | 617 +++++++++++++++++++++++++++++++++++++ back/mult1 | 636 ++++++++++++++++++++++++++++++++++++++ config/URL_config.ini | 34 ++- config/youtube.ini | 3 +- main.py | 687 +++++++++++++++++++++++++----------------- 5 files changed, 1704 insertions(+), 273 deletions(-) create mode 100644 back/back4 create mode 100644 back/mult1 diff --git a/back/back4 b/back/back4 new file mode 100644 index 0000000..e49c354 --- /dev/null +++ b/back/back4 @@ -0,0 +1,617 @@ +# -*- 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.1.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 _round_res(w: int, h: int) -> Tuple[int, int]: + # 强制偶数，避免 1088/1918 这类非标准尺寸 + return (w & ~1, h & ~1) + +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() # 例如 "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) + 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) + + # —— 默认强制转码（稳） —— # + v_args = [ + "-c:v", "libx264", + "-preset", "veryfast", + "-tune", "zerolatency", + "-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", + ] + + # 可选缩放/留黑边 + 避免 1088x1920（crop 强制精确尺寸）+ 偶数对齐 + if res_str: + try: + w, h = map(int, res_str.lower().split("x")) + w, h = _round_res(w, h) + 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, + v_args: Optional[list] = None, + a_args: Optional[list] = None) -> subprocess.Popen: + """ + 启动一个 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") + + # 输入侧容错/重连/探测 + base = [ + "ffmpeg", "-nostdin", "-hide_banner", + "-stats", + "-stats_period", "60", + "-progress", "pipe:1", + "-loglevel", "info", + + # 输入容错 & 自动重连 + "-rw_timeout", "15000000", + "-timeout", "10000000", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_at_eof", "1", + "-reconnect_delay_max", "30", + + # 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", + ] + + # 平台 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+igndts+discardcorrupt+nobuffer", + "-xerror", + "-use_wallclock_as_timestamps", "1", + "-avioflags", "direct", + "-rtbufsize", "100M", + ] + + # 视频/音频编码参数（使用 build_youtube_output 的结果） + v_args = v_args or [] + a_args = a_args or [] + base += v_args + a_args + + # 输出（flv 到 YT） + base += ["-f", "flv", out_url] + + # 打印命令便于排障（截断） + cmd_str = " ".join(base) + print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...") + + logger.info(f"启动推流：{input_url} -> {out_url}") + + creationflags = 0 + if os.name == "nt": + creationflags = subprocess.CREATE_NEW_PROCESS_GROUP + + 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(): + try: + for line in iter(p.stdout.readline, b''): + if not line: + break + text = line.decode(errors="ignore").strip() + p.last_out_ts = time.time() + + # 输出关键统计，便于观察码率/帧率 + if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text): + logger.info(f"[ffmpeg统计] {text}") + + # 常见错误提示 + if "Server disconnected" in text: + logger.error(f"[YouTube/RTMP输出异常] {text}") + elif "not supported" in text or "Invalid" in text or "invalid" in text: + logger.error(f"[转码异常] {text}") + except Exception: + pass + + import threading + 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 + p.wait(timeout=6) + 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: Optional[str] = None + current_real_url: Optional[str] = None # 实际播放地址（flv/m3u8） + 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 文件变化时热更新 + + # watchdog 参数（更稳） + watchdog_grace = 120 # 推流开始 120s 内不做静默判定 + watchdog_silence = 600 # 10 分钟完全无输出才重启 + refresh_real_url_every = 300 # 每 5 分钟刷新一次真实播放地址，避免 token 过期 + + 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 + current_real_url = None + except Exception as e: + logger.warning(f"URL 列表热更新失败：{e}") + + if current_proc and current_url: + 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 = None + current_url = None + current_real_url = None + 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 # 容错：无法判断时先认为还活着 + + 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="") + + # 定期刷新真实播放地址，避免 token 过期 + try: + if elapsed >= refresh_real_url_every and (elapsed % refresh_real_url_every) < check_interval_live: + new_port = asyncio.run(fetch_port_info(current_url, record_quality_code)) + new_real_url = select_source_url(current_url, new_port) + if new_real_url and new_real_url != current_real_url: + color.print_colored(f"\n[{now_str()}] 刷新播放地址（token 变化），重启 ffmpeg…", color.YELLOW) + stop_ffmpeg(current_proc) + out_url, v_args, a_args = build_youtube_output() + headers = headers_for_platform(current_platform, current_url) + current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args=v_args, a_args=a_args) + current_real_url = new_real_url + start_time = time.time() + time.sleep(check_interval_live) + continue + except Exception as e: + logger.warning(f"刷新播放地址失败：{e}") + + # ✅ watchdog（推流开始 > 120s 且 10 分钟完全无输出） + 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()}] 超过10分钟未见 ffmpeg 输出，判定异常，重启…", color.YELLOW) + stop_ffmpeg(current_proc) + current_proc = None + current_url = None + current_real_url = None + time.sleep(1) + continue + + 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 + current_real_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, 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 + current_real_url = real_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() diff --git a/back/mult1 b/back/mult1 new file mode 100644 index 0000000..9ce824c --- /dev/null +++ b/back/mult1 @@ -0,0 +1,636 @@ +# -*- encoding: utf-8 -*- + +""" +多路并行版： +- 同时启动 N 路（N = youtube.ini 的 keys 数量） +- 每路独立 ffmpeg 进程、watchdog、定期刷新抖音真实播放地址 +- 共享 URL 池，避免重复占用同一个主播 +""" + +import asyncio +import os +import sys +import signal +import time +import datetime +import re +import uuid +import threading +import subprocess +from typing import Any, List, Tuple, Optional, Dict, Set + +import configparser +import urllib.request + +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-parallel-2.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' +yt_ini_file = f"{script_path}/config/youtube.ini" +text_encoding = 'utf-8-sig' +color = Color() + +# ---------- YouTube 推流配置（读取 youtube.ini） ---------- +_yt_cfg = configparser.ConfigParser() +with open(yt_ini_file, "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 _yt_get_list(section: str, key: str) -> List[str]: + raw = _yt_get(section, key, fallback="", cast=str) or "" + # 支持逗号、空白、换行分隔 + parts = re.split(r"[,;\s]+", raw.strip()) + return [p for p in parts if p] + +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 _round_res(w: int, h: int) -> Tuple[int, int]: + # 强制偶数，避免 1088/1918 这类尺寸（H.264 对齐问题） + return (w & ~1, h & ~1) + +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]: + 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: + 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: + 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: + 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_for_key(key: str, resolution: Optional[str]) -> Tuple[str, list, list]: + if not key: + logger.error("错误: youtube.ini 未配置 key/keys") + sys.exit(1) + + secure = _yt_get("youtube", "secure", fallback=False, cast=bool) + ingest = _yt_get("youtube", "ingest", fallback="a", cast=str) + res_str = (resolution or _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) + + v_args = [ + "-c:v", "libx264", + "-preset", "veryfast", + "-tune", "zerolatency", + "-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", + ] + + if res_str: + try: + w, h = map(int, res_str.lower().split("x")) + w, h = _round_res(w, h) + 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 + +# --------------------- 启动 ffmpeg --------------------- +def launch_ffmpeg(input_url: str, + out_url: str, + extra_headers: Optional[str], + v_args: list, + a_args: list) -> subprocess.Popen: + 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") + base = [ + "ffmpeg", "-nostdin", "-hide_banner", + "-stats", "-stats_period", "60", + "-progress", "pipe:1", + "-loglevel", "info", + + # 输入容错 & 自动重连 + "-rw_timeout", "15000000", + "-timeout", "10000000", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_at_eof", "1", + "-reconnect_delay_max", "30", + + "-user_agent", user_agent, + "-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy", + + "-thread_queue_size", "4096", + "-analyzeduration", "10000000", + "-probesize", "10000000", + ] + if extra_headers: + base += ["-headers", extra_headers] + + 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+igndts+discardcorrupt+nobuffer", + "-xerror", + "-use_wallclock_as_timestamps", "1", + "-avioflags", "direct", + "-rtbufsize", "100M", + ] + + base += v_args + a_args + base += ["-f", "flv", out_url] + + cmd_str = " ".join(base) + print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...") + logger.info(f"启动推流：{input_url} -> {out_url}") + + creationflags = 0 + if os.name == "nt": + creationflags = subprocess.CREATE_NEW_PROCESS_GROUP + + p = subprocess.Popen( + base, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + bufsize=0, + env=env, + creationflags=creationflags + ) + p.last_out_ts = time.time() + + def _reader(): + try: + for line in iter(p.stdout.readline, b''): + if not line: + break + text = line.decode(errors="ignore").strip() + p.last_out_ts = time.time() + + if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text): + logger.info(f"[ffmpeg统计] {text}") + if "Server disconnected" in text: + logger.error(f"[输出异常] {text}") + elif "not supported" in text or "Invalid" in text or "invalid" in text: + logger.error(f"[转码异常] {text}") + except Exception: + pass + + threading.Thread(target=_reader, daemon=True).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 + p.wait(timeout=6) + else: + p.send_signal(signal.SIGINT) + p.wait(timeout=10) + except Exception: + try: p.kill() + except Exception: pass + +# --------------------- 平台头部 --------------------- +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 "未知平台" + +# --------------------- 并行协调器 --------------------- +class UrlCoordinator: + """ 管理 URL 池、热更新、分配给不同 worker，避免重复占用 """ + def __init__(self, path: str): + self.path = path + self.urls: List[str] = parse_urls_from_file(path) + self.idx: int = 0 + self.in_use: Set[str] = set() + self.lock = threading.Lock() + self.last_md5 = check_md5(path) + + def refresh_if_changed(self): + try: + md5 = check_md5(self.path) + if md5 != self.last_md5: + self.urls = parse_urls_from_file(self.path) or self.urls + self.last_md5 = md5 + print(f"[{now_str()}] 已重载 URL 列表，数量：{len(self.urls)}") + except Exception as e: + logger.warning(f"URL 列表热更新失败：{e}") + + def acquire_next_live(self, quality_code: str = "OD") -> Optional[Tuple[str, dict]]: + """ 顺序扫描，跳过 in_use，找到正在直播的返回并占用 """ + with self.lock: + if not self.urls: + return None + start = self.idx + tried = 0 + while tried < len(self.urls): + url = self.urls[self.idx] + self.idx = (self.idx + 1) % len(self.urls) + tried += 1 + if url in self.in_use: + continue + # 释放锁去解析，避免长时间占用 + pass_url = url + break + else: + return None + # 锁外解析 + port = asyncio.run(fetch_port_info(pass_url, quality_code)) + if not port.get("anchor_name") or not port.get("is_live"): + return None + with self.lock: + # 二次检查，确保没被别人抢占 + if pass_url in self.in_use: + return None + self.in_use.add(pass_url) + return pass_url, port + + def release(self, url: Optional[str]): + if not url: + return + with self.lock: + self.in_use.discard(url) + +def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + +# --------------------- Worker --------------------- +class StreamWorker(threading.Thread): + def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator): + super().__init__(daemon=True) + self.worker_id = worker_id + self.key = key + self.resolution = resolution + self.coordinator = coordinator + + self.current_url: Optional[str] = None + self.current_real_url: Optional[str] = None + self.current_proc: Optional[subprocess.Popen] = None + self.current_platform = "" + self.current_anchor = "" + self.record_quality_code = "OD" + + self.watchdog_grace = 120 + self.watchdog_silence = 600 + self.refresh_real_url_every = 300 + self.check_interval_live = 8 + self.scan_interval_when_idle = 2 + + self._stop_event = threading.Event() + + def stop(self): + self._stop_event.set() + + def run(self): + color.print_colored(f"[Worker-{self.worker_id}] 启动", color.GREEN) + try: + while not self._stop_event.is_set(): + # 刷新 URL 文件 + self.coordinator.refresh_if_changed() + + if self.current_proc and self.current_url: + retcode = self.current_proc.poll() + if retcode is not None: + color.print_colored(f"[{now_str()}][Worker-{self.worker_id}] ffmpeg 退出(code={retcode})，释放并切源…", color.YELLOW) + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url) + self._clear_state() + continue + + # 仍在推流：检查主播是否仍在直播 + still_live = True + try: + port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + still_live = bool(port.get("is_live")) + except Exception as e: + logger.error(f"[Worker-{self.worker_id}] 检测当前源失败：{e}") + still_live = True + + if still_live: + elapsed = int(time.time() - self.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()}][Worker-{self.worker_id}] 正在转播({dur_str})：{self.current_platform} | {self.current_anchor} | {self.current_url}", end="") + + # 定期刷新真实播放地址 + try: + if elapsed >= self.refresh_real_url_every and (elapsed % self.refresh_real_url_every) < self.check_interval_live: + new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + new_real_url = select_source_url(self.current_url, new_port) + if new_real_url and new_real_url != self.current_real_url: + color.print_colored(f"\n[{now_str()}][Worker-{self.worker_id}] 刷新播放地址（token 变化），重启 ffmpeg…", color.YELLOW) + stop_ffmpeg(self.current_proc) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + headers = headers_for_platform(self.current_platform, self.current_url) + self.current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args, a_args) + self.current_real_url = new_real_url + self.start_time = time.time() + time.sleep(self.check_interval_live) + continue + except Exception as e: + logger.warning(f"[Worker-{self.worker_id}] 刷新播放地址失败：{e}") + + # watchdog：心跳超时才重启 + silence = time.time() - getattr(self.current_proc, "last_out_ts", 0) + if elapsed >= self.watchdog_grace and silence > self.watchdog_silence: + color.print_colored(f"\n[{now_str()}][Worker-{self.worker_id}] 超过10分钟未见输出，重启…", color.YELLOW) + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + time.sleep(1) + continue + + time.sleep(self.check_interval_live) + continue + else: + color.print_colored(f"\n[{now_str()}][Worker-{self.worker_id}] 主播停播，释放并切源…", color.YELLOW) + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + continue + + # 未在推流：尝试分配新的直播 URL + got = self.coordinator.acquire_next_live(self.record_quality_code) + if not got: + print(f"[{now_str()}][Worker-{self.worker_id}] 无可用直播源，稍后再试…") + time.sleep(self.scan_interval_when_idle) + continue + + url, port = got + real_url = select_source_url(url, port) + if not real_url: + print(f"[{now_str()}][Worker-{self.worker_id}] 未得到播放地址，跳过：{url}") + self.coordinator.release(url) + time.sleep(1) + continue + + self.current_platform = detect_platform(url) + self.current_anchor = port.get("anchor_name", "") + headers = headers_for_platform(self.current_platform, url) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args) + self.current_url = url + self.current_real_url = real_url + self.start_time = time.time() + color.print_colored(f"[{now_str()}][Worker-{self.worker_id}] 开始转播：{self.current_platform} | {self.current_anchor} | {url}", color.GREEN) + + except Exception as e: + logger.error(f"[Worker-{self.worker_id}] 主循环异常：{e}") + finally: + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url) + print(f"[Worker-{self.worker_id}] 已退出。") + + def _clear_state(self, release: bool = False): + if release and self.current_url: + self.coordinator.release(self.current_url) + self.current_url = None + self.current_real_url = None + self.current_proc = None + self.current_platform = "" + self.current_anchor = "" + +# --------------------- 启动入口 --------------------- +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 main(): + print("-----------------------------------------------------") + print("| DouyinLiveRecorder - 多路并行转推版 |") + print("-----------------------------------------------------") + print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") + print("模式：并行检测 + 并行推流（每路独立 watchdog）") + 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) + + keys = _yt_get_list("youtube", "keys") + if not keys: + # 兼容单 key 配置 + single_key = _yt_get("youtube", "key", fallback="", cast=str) + if single_key: + keys = [single_key] + if not keys: + logger.error("youtube.ini 没有配置 [youtube].keys 或 key") + sys.exit(1) + + # 可选：为每个 key 指定不同分辨率（与 keys 顺序对应），否则使用全局 resolution + per_res = _yt_get_list("youtube", "resolutions") # 例如：1920x1080,1080x1920,1280x720 + while per_res and len(per_res) < len(keys): + per_res.append(per_res[-1]) # 不足时沿用最后一个 + + # 协调器（共享 URL 池） + coordinator = UrlCoordinator(url_config_file) + + # 启动多路 worker + workers: List[StreamWorker] = [] + for i, key in enumerate(keys, start=1): + res_for_this = per_res[i-1] if per_res else None + w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator) + w.start() + workers.append(w) + + # 主线程仅负责捕获 Ctrl+C 并优雅退出 + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("\n收到中断，正在退出…") + finally: + for w in workers: + w.stop() + for w in workers: + w.join(timeout=10) + print("全部 worker 已退出。") + +if __name__ == "__main__": + main() diff --git a/config/URL_config.ini b/config/URL_config.ini index e409898..1fda72a 100644 --- a/config/URL_config.ini +++ b/config/URL_config.ini @@ -1,6 +1,8 @@ -https://live.douyin.com/60379224194 +https://live.douyin.com/72212955083 https://live.douyin.com/621118100231 +https://live.douyin.com/347310028364 https://live.douyin.com/280422307519 +https://live.douyin.com/651293616593 https://live.douyin.com/114111936890 https://live.douyin.com/43665279648 https://live.douyin.com/122383435921 @@ -8,7 +10,10 @@ https://live.douyin.com/92876979328 https://live.douyin.com/136996589868 https://live.douyin.com/517715534931 https://live.douyin.com/460525712926 +https://live.douyin.com/456576000931 +https://live.douyin.com/745606325880 https://live.douyin.com/749687541944 +https://live.douyin.com/675343045974 https://live.douyin.com/673565298571 https://live.douyin.com/835571459859 https://live.douyin.com/152358755212 @@ -18,4 +23,29 @@ https://live.douyin.com/967381081829 https://live.douyin.com/15652099787 https://live.douyin.com/78012762575 https://live.douyin.com/80248655914 - +https://live.douyin.com/163813589919 +https://live.douyin.com/689189188452 +https://live.douyin.com/498368994814 +https://live.douyin.com/617848099204 +https://live.douyin.com/821525628944 +https://live.douyin.com/993102287144 +https://live.douyin.com/701547125568 +https://live.douyin.com/816130992220 +https://live.douyin.com/876468215361 +https://live.douyin.com/291403201249 +https://live.douyin.com/511335278313 +https://live.douyin.com/483160615952 +https://live.douyin.com/673565298571 +https://live.douyin.com/525514386431 +https://live.douyin.com/8687122573 +https://live.douyin.com/642534242822 +https://live.douyin.com/743565594721 +https://live.douyin.com/249578288248 +https://live.douyin.com/472140253414 +https://live.douyin.com/700846653732 +https://live.douyin.com/81849868631 +https://live.douyin.com/690114366322 +https://live.douyin.com/642534242822 +https://live.douyin.com/469980190666 +https://live.douyin.com/388066418744 +https://live.douyin.com/122383435921 \ No newline at end of file diff --git a/config/youtube.ini b/config/youtube.ini index 60514c6..829835a 100644 --- a/config/youtube.ini +++ b/config/youtube.ini @@ -1,5 +1,6 @@ [youtube] key = x04z-564w-aks7-embw-30y4 +keys = x04z-564w-aks7-embw-30y4,qxvb-r47b-r5ju-6ud3-6k7z secure = true ingest = a bitrate = 5000 @@ -7,5 +8,5 @@ audio_bitrate = 128 fps = 30 gop = 60 copy_when_safe = true -resolution = 1080x1920 +resolution = 720x1280 single_mode = true \ No newline at end of file diff --git a/main.py b/main.py index e49c354..749fd4c 100644 --- a/main.py +++ b/main.py @@ -1,9 +1,12 @@ # -*- encoding: utf-8 -*- """ -单进程 ffmpeg 推流：从 URL_config.ini 顺序检测直播源， -一旦当前源停播/异常，自动切到下一个正在直播的源并推流到 YouTube。 -仅做“检测+转推”，不做录制、分段、通知等。 +并行转推（日志优化版，保持原业务逻辑） +- 保留 Worker-1/2 等前缀 +- 日志状态变化才打印，避免刷屏 +- 禁止同一个 URL 推到同一个 YouTube key（(url,key) 级别防重复） +- 仍然避免同一个 URL 被多路并行占用（url 级别独占） +- 新增：Ctrl+C/TERM 时优雅停止所有 Worker 与 ffmpeg 子进程（仅最小增改） """ import asyncio @@ -14,31 +17,36 @@ import time import datetime import re import uuid +import threading import subprocess -from pathlib import Path -from typing import Any, List, Tuple, Optional +from typing import Any, List, Tuple, Optional, Dict, Set 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.1.0" +# =============== 基础环境 =============== +version = "relay-parallel-2.0.1-logonly" 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' +yt_ini_file = f"{script_path}/config/youtube.ini" text_encoding = 'utf-8-sig' color = Color() -# ---------- YouTube 推流配置（读取 youtube.ini） ---------- +# ---- 新增：全局 Worker & 进程注册（用于 Ctrl+C 干净退出） ---- +ALL_WORKERS: List["StreamWorker"] = [] +ALL_PROCS: Set[subprocess.Popen] = set() +SHUTDOWN_EVENT = threading.Event() +ALL_LOCK = threading.Lock() + +# =============== 读取 YouTube 配置 =============== _yt_cfg = configparser.ConfigParser() -with open(f"{script_path}/config/youtube.ini", "r", encoding="utf-8-sig") as f: +with open(yt_ini_file, "r", encoding="utf-8-sig") as f: _yt_cfg.read_file(f) def _yt_get(section: str, key: str, fallback=None, cast=str): @@ -51,17 +59,22 @@ def _yt_get(section: str, key: str, fallback=None, cast=str): except Exception: return fallback -def _safe_int(v, lo, hi, default): +def _yt_get_list(section: str, key: str) -> List[str]: + raw = _yt_get(section, key, fallback="", cast=str) or "" + parts = re.split(r"[,;\s]+", raw.strip()) + return [p for p in parts if p] + +def _safe_int(v, lo, hi, defv): try: v = int(v) - if v < lo or v > hi: - return default + if v < lo or v > hi: return defv return v except Exception: - return default + return defv +# =============== 工具函数 =============== def _round_res(w: int, h: int) -> Tuple[int, int]: - # 强制偶数，避免 1088/1918 这类非标准尺寸 + # 偶数对齐，避免 1088x1920 类问题 return (w & ~1, h & ~1) def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None): @@ -71,7 +84,10 @@ def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None): host = f"{node}.rtmps.youtube.com" if secure else f"{node}.rtmp.youtube.com" return f"{scheme}://{host}/live2/{key}" -# --------------------- 简化配置读取 --------------------- +def now_str(): + return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + +# =============== 读取运行配置 =============== def read_config_value(cfg: configparser.RawConfigParser, section: str, option: str, default_value: Any) -> Any: try: cfg.read(config_file, encoding=text_encoding) @@ -85,16 +101,14 @@ def read_config_value(cfg: configparser.RawConfigParser, section: str, option: s 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) @@ -103,28 +117,18 @@ except Exception: if not use_proxy: color.print_colored("提示：无法直连海外站，如需 TikTok 请在 config.ini 开启代理并填写代理地址。", color.YELLOW) -# --------------------- URL 列表读取 --------------------- +# =============== 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("#"): + if not s or s.startswith("#"): continue - # 拆分（最多3段，取包含 https 或域名的那段） parts = re.split(r"[，,]\s*", s) cand = None for p in parts: @@ -139,9 +143,8 @@ def parse_urls_from_file(path: str) -> List[str]: 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]: @@ -150,7 +153,6 @@ def select_source_url(link: str, stream_info: dict) -> Optional[str]: 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 "") @@ -163,17 +165,6 @@ def select_source_url(link: str, stream_info: dict) -> Optional[str]: 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: @@ -191,7 +182,6 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: 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, @@ -201,23 +191,21 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: 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) +# =============== 构建输出参数（每个 key） =============== +def build_youtube_output_for_key(key: str, resolution: Optional[str]) -> Tuple[str, list, list]: if not key: - logger.error("错误: config/youtube.ini 未配置 key") + logger.error("错误: youtube.ini 未配置 key/keys") 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() # 例如 "1080x1920" + res_str = (resolution or _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) @@ -226,7 +214,6 @@ def build_youtube_output() -> Tuple[str, list, list]: out_url = _build_youtube_rtmp_url(key, secure, ingest) - # —— 默认强制转码（稳） —— # v_args = [ "-c:v", "libx264", "-preset", "veryfast", @@ -247,7 +234,6 @@ def build_youtube_output() -> Tuple[str, list, list]: "-ac", "2", ] - # 可选缩放/留黑边 + 避免 1088x1920（crop 强制精确尺寸）+ 偶数对齐 if res_str: try: w, h = map(int, res_str.lower().split("x")) @@ -264,27 +250,19 @@ def build_youtube_output() -> Tuple[str, list, list]: return out_url, v_args, a_args +# =============== 启动/停止 ffmpeg =============== 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/RTMPS 到 out_url。 - - v_args / a_args 由 build_youtube_output() 生成并传入 - - 启动 stdout reader 线程以防缓冲阻塞 - - 设置代理环境变量以兼容 https/hls - """ - import threading + extra_headers: Optional[str], + v_args: list, + a_args: list, + worker_name: str) -> subprocess.Popen: 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") - - # 输入侧容错/重连/探测 base = [ "ffmpeg", "-nostdin", "-hide_banner", - "-stats", - "-stats_period", "60", + "-stats", "-stats_period", "60", "-progress", "pipe:1", "-loglevel", "info", @@ -296,31 +274,24 @@ def launch_ffmpeg(input_url: str, "-reconnect_at_eof", "1", "-reconnect_delay_max", "30", - # 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", ] - - # 平台 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+igndts+discardcorrupt+nobuffer", "-xerror", @@ -329,19 +300,12 @@ def launch_ffmpeg(input_url: str, "-rtbufsize", "100M", ] - # 视频/音频编码参数（使用 build_youtube_output 的结果） - v_args = v_args or [] - a_args = a_args or [] base += v_args + a_args - - # 输出（flv 到 YT） base += ["-f", "flv", out_url] - # 打印命令便于排障（截断） cmd_str = " ".join(base) - print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...") - - logger.info(f"启动推流：{input_url} -> {out_url}") + print(f"ffmpeg cmd ({worker_name}):", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...") + logger.info(f"{worker_name} 启动推流：{input_url} -> {out_url}") creationflags = 0 if os.name == "nt": @@ -352,14 +316,16 @@ def launch_ffmpeg(input_url: str, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - bufsize=0, # 立即刷新，避免行缓冲告警 + bufsize=0, env=env, creationflags=creationflags ) - - # 最近输出时间戳，用于 watchdog p.last_out_ts = time.time() + # ---- 新增：注册进程，便于 Ctrl+C 统一清理 ---- + with ALL_LOCK: + ALL_PROCS.add(p) + def _reader(): try: for line in iter(p.stdout.readline, b''): @@ -368,33 +334,26 @@ def launch_ffmpeg(input_url: str, text = line.decode(errors="ignore").strip() p.last_out_ts = time.time() - # 输出关键统计，便于观察码率/帧率 - if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text): - logger.info(f"[ffmpeg统计] {text}") - - # 常见错误提示 + # 只保留关键统计/错误（前缀带 Worker） + if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text) or ("libx264" in text): + logger.info(f"{worker_name} [ffmpeg统计] {text}") if "Server disconnected" in text: - logger.error(f"[YouTube/RTMP输出异常] {text}") - elif "not supported" in text or "Invalid" in text or "invalid" in text: - logger.error(f"[转码异常] {text}") + logger.error(f"{worker_name} [输出异常] {text}") + elif ("not supported" in text) or ("Invalid" in text) or ("invalid" in text): + logger.error(f"{worker_name} [转码异常] {text}") except Exception: pass - import threading - t = threading.Thread(target=_reader, daemon=True) - t.start() - + threading.Thread(target=_reader, daemon=True).start() return p def stop_ffmpeg(p: Optional[subprocess.Popen]): - if not p: - return + if not p: return try: if os.name == "nt": if p.stdin: try: - p.stdin.write(b"q") - p.stdin.flush() + p.stdin.write(b"q"); p.stdin.flush() except Exception: pass p.wait(timeout=6) @@ -402,24 +361,14 @@ def stop_ffmpeg(p: Optional[subprocess.Popen]): 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 + try: p.kill() + except Exception: pass finally: - return bool(check_ffmpeg()) - -def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + # ---- 新增：从全局集合移除 ---- + with ALL_LOCK: + ALL_PROCS.discard(p) +# =============== 平台 headers/识别 =============== def headers_for_platform(platform_name: str, live_url: str) -> Optional[str]: live_domain = '/'.join(live_url.split('/')[0:3]) record_headers = { @@ -441,12 +390,309 @@ def detect_platform(url: str) -> str: if url.endswith(".flv") or url.endswith(".m3u8"): return "自定义录制直播" return "未知平台" +# =============== 日志状态缓存（去重/限频） =============== +class LogState: + def __init__(self): + self.lock = threading.Lock() + self.url_status: Dict[str, str] = {} # url -> "unknown/offline/online/streaming" + self.no_source_worker: Dict[str, bool] = {} # worker_name -> idle printed flag + + def set_status(self, url: str, status: str, text: Optional[str] = None, prefix: str = ""): + with self.lock: + prev = self.url_status.get(url, "unknown") + if prev != status: + self.url_status[url] = status + if text: + print(f"[{now_str()}]{prefix} {text}") + + def mark_idle_once(self, worker_name: str): + with self.lock: + if not self.no_source_worker.get(worker_name, False): + print(f"[{now_str()}][{worker_name}] 当前无可用直播源，等待中…") + self.no_source_worker[worker_name] = True + + def clear_idle_flag(self, worker_name: str): + with self.lock: + self.no_source_worker[worker_name] = False + +LOGSTATE = LogState() + +# =============== URL 协调器（包含 url 与 (url,key) 占用） =============== +class UrlCoordinator: + """ + - urls 顺序分配 + - in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用） + - in_use_pairs: (url,key) 级别独占（禁止同一 url 推到同一个 key） + """ + def __init__(self, path: str): + self.path = path + self.urls: List[str] = parse_urls_from_file(path) + self.idx: int = 0 + self.lock = threading.Lock() + self.last_md5 = check_md5(path) + + self.in_use_urls: Set[str] = set() + self.in_use_pairs: Set[Tuple[str, str]] = set() # (url, key) + + def refresh_if_changed(self): + try: + md5 = check_md5(self.path) + if md5 != self.last_md5: + self.urls = parse_urls_from_file(self.path) or self.urls + self.last_md5 = md5 + print(f"[{now_str()}] 已重载 URL 列表，数量：{len(self.urls)}") + except Exception as e: + logger.warning(f"URL 列表热更新失败：{e}") + + def acquire_next_live(self, worker_key: str, quality_code: str = "OD") -> Optional[Tuple[str, dict]]: + with self.lock: + if not self.urls: + return None + tried = 0 + pick = None + while tried < len(self.urls): + url = self.urls[self.idx] + self.idx = (self.idx + 1) % len(self.urls) + tried += 1 + # 已被其他路占用 / 同 url-key 已经在用 + if url in self.in_use_urls: + continue + if (url, worker_key) in self.in_use_pairs: + continue + pick = url + break + if not pick: + return None + + # 锁外解析 + port = asyncio.run(fetch_port_info(pick, quality_code)) + if not port.get("anchor_name"): + LOGSTATE.set_status(pick, "unknown", + text=f"解析失败：{pick}", + prefix="") + return None + if not port.get("is_live"): + anchor = port.get('anchor_name','') + LOGSTATE.set_status(pick, "offline", + text=f"未开播：{anchor} | {pick}", + prefix="") + return None + + # 占用 + with self.lock: + if (pick in self.in_use_urls) or ((pick, worker_key) in self.in_use_pairs): + return None + self.in_use_urls.add(pick) + self.in_use_pairs.add((pick, worker_key)) + + # 状态更新（online） + anchor = port.get('anchor_name','') + LOGSTATE.set_status(pick, "online", + text=f"已开播：{anchor} | {pick}", + prefix="") + return pick, port + + def release(self, url: Optional[str], worker_key: Optional[str]): + if not url: + return + with self.lock: + self.in_use_urls.discard(url) + if worker_key: + self.in_use_pairs.discard((url, worker_key)) + +# =============== Worker（保留前缀日志） =============== +class StreamWorker(threading.Thread): + def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator): + super().__init__(daemon=True) + self.worker_id = worker_id + self.key = key + self.worker_name = f"Worker-{worker_id}" + self.resolution = resolution + self.coordinator = coordinator + + self.current_url: Optional[str] = None + self.current_real_url: Optional[str] = None + self.current_proc: Optional[subprocess.Popen] = None + self.current_platform = "" + self.current_anchor = "" + self.record_quality_code = "OD" + + # 与原逻辑一致的参数 + self.watchdog_grace = 120 + self.watchdog_silence = 600 + self.refresh_real_url_every = 300 + self.check_interval_live = 8 + self.scan_interval_when_idle = 2 + + self._last_progress_print = -9999 + self._stop_event = threading.Event() + + def stop(self): + """最小改动：收到停止请求时，立即尝试停掉当前 ffmpeg，尽快响应退出。""" + self._stop_event.set() + try: + stop_ffmpeg(self.current_proc) + except Exception: + pass + + def _p(self, text: str): + print(f"[{now_str()}][{self.worker_name}] {text}") + + def run(self): + self._p("启动") + try: + while not self._stop_event.is_set(): + self.coordinator.refresh_if_changed() + + if self.current_proc and self.current_url: + retcode = self.current_proc.poll() + if retcode is not None: + self._p(f"❌ ffmpeg 退出(code={retcode})，切换下一个源…") + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url, self.key) + self._clear_state() + LOGSTATE.clear_idle_flag(self.worker_name) + continue + + # 检测是否仍在直播（保持原逻辑） + still_live = True + try: + port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + still_live = bool(port.get("is_live")) + except Exception as e: + logger.error(f"[{self.worker_name}] 检测当前源失败：{e}") + still_live = True + + if still_live: + elapsed = int(time.time() - self.start_time) + # 每 60s 打印一次“正在转播” + if elapsed - self._last_progress_print >= 60: + h, m = divmod(elapsed, 3600) + m, s = divmod(m, 60) + dur_str = f"{h:02d}:{m:02d}:{s:02d}" + self._p(f"⏱ 正在转播({dur_str})：{self.current_platform} | {self.current_anchor}") + self._last_progress_print = elapsed + + # 定期刷新真实播放地址（token） + try: + if elapsed >= self.refresh_real_url_every and (elapsed % self.refresh_real_url_every) < self.check_interval_live: + new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + new_real_url = select_source_url(self.current_url, new_port) + if new_real_url and new_real_url != self.current_real_url: + self._p("🔄 刷新播放地址（token 变化），重启 ffmpeg…") + stop_ffmpeg(self.current_proc) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + headers = headers_for_platform(self.current_platform, self.current_url) + self.current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args, a_args, self.worker_name) + self.current_real_url = new_real_url + self.start_time = time.time() + self._last_progress_print = -9999 + time.sleep(self.check_interval_live) + continue + except Exception as e: + logger.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}") + + # watchdog + silence = time.time() - getattr(self.current_proc, "last_out_ts", 0) + if elapsed >= self.watchdog_grace and silence > self.watchdog_silence: + self._p("⚠️ 超过10分钟未见输出，重启…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + time.sleep(1) + continue + + time.sleep(self.check_interval_live) + continue + else: + self._p("📴 主播停播，切换下一个源…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + continue + + # —— 未在推流：尝试分配一个新源（保持原顺序轮询 + 占用） —— + got = self.coordinator.acquire_next_live(self.key, self.record_quality_code) + if not got: + # 仅首次空闲时提示一次 + LOGSTATE.mark_idle_once(self.worker_name) + time.sleep(self.scan_interval_when_idle) + continue + + LOGSTATE.clear_idle_flag(self.worker_name) + url, port = got + real_url = select_source_url(url, port) + if not real_url: + self._p(f"未得到播放地址，跳过：{url}") + self.coordinator.release(url, self.key) + time.sleep(1) + continue + + self.current_platform = detect_platform(url) + self.current_anchor = port.get("anchor_name", "") + headers = headers_for_platform(self.current_platform, url) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args, self.worker_name) + self.current_url = url + self.current_real_url = real_url + self.start_time = time.time() + self._last_progress_print = -9999 + + self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}") + + except Exception as e: + logger.error(f"[{self.worker_name}] 主循环异常：{e}") + finally: + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url, self.key) + self._p("已退出。") + + def _clear_state(self, release: bool = False): + if release and self.current_url: + self.coordinator.release(self.current_url, self.key) + self.current_url = None + self.current_real_url = None + self.current_proc = None + self.current_platform = "" + self.current_anchor = "" + +# =============== 启动入口（保持原行为） =============== +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 graceful_shutdown(signum=None, frame=None): + if not SHUTDOWN_EVENT.is_set(): + print("\n收到中断，正在停止所有 worker 与 ffmpeg …") + SHUTDOWN_EVENT.set() + # 先发 stop 给所有 worker（会内部 stop 当前 ffmpeg） + for w in list(ALL_WORKERS): + try: + w.stop() + except Exception: + pass + # 再兜底停止全局登记的 ffmpeg 子进程 + with ALL_LOCK: + procs = list(ALL_PROCS) + for p in procs: + try: + stop_ffmpeg(p) + except Exception: + pass + def main(): print("-----------------------------------------------------") - print("| DouyinLiveRecorder - 单进程转推版 |") + print("| DouyinLiveRecorder - 并行转推版 |") print("-----------------------------------------------------") print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") - print("模式：只检测 + 单进程推流（停播自动切下一个 URL）") + print("模式：并行检测 + 并行推流（保持原有业务逻辑）") print(".....................................................") if not check_ffmpeg_existence(): @@ -458,160 +704,61 @@ def main(): logger.error("URL_config.ini 中没有可用 URL。") sys.exit(1) - # 顺序轮询索引 - idx = 0 - current_url: Optional[str] = None - current_real_url: Optional[str] = None # 实际播放地址（flv/m3u8） - current_proc: Optional[subprocess.Popen] = None - current_platform = "" - current_anchor = "" - record_quality_code = "OD" # 固定“原画”；需要也可改为从 config.ini 读取 + # keys（并行路数 = keys 数量）；兼容单 key + keys = _yt_get_list("youtube", "keys") + if not keys: + single_key = _yt_get("youtube", "key", fallback="", cast=str) + if single_key: + keys = [single_key] + if not keys: + logger.error("youtube.ini 没有配置 [youtube].keys 或 key") + sys.exit(1) - check_interval_live = 8 # 推流中检测当前源是否仍在直播的间隔秒 - scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒 - last_md5 = "" # URL 文件变化时热更新 + # 可选：每路分辨率 + per_res = _yt_get_list("youtube", "resolutions") # 如：1920x1080,1080x1920 + while per_res and len(per_res) < len(keys): + per_res.append(per_res[-1]) - # watchdog 参数（更稳） - watchdog_grace = 120 # 推流开始 120s 内不做静默判定 - watchdog_silence = 600 # 10 分钟完全无输出才重启 - refresh_real_url_every = 300 # 每 5 分钟刷新一次真实播放地址，避免 token 过期 + # 打印 ffmpeg 版本（贴合旧风格） + try: + vline = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True).stdout.splitlines()[0] + print(vline) + except Exception: + pass + + coordinator = UrlCoordinator(url_config_file) + + # 信号处理（Ctrl+C / kill） + try: + signal.signal(signal.SIGINT, graceful_shutdown) + except Exception: + pass + try: + signal.signal(signal.SIGTERM, graceful_shutdown) + except Exception: + pass + + # 启动多路（Worker 前缀区分） + workers: List[StreamWorker] = [] + for i, key in enumerate(keys, start=1): + res_for_this = per_res[i-1] if per_res else None + w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator) + w.start() + workers.append(w) + # ---- 新增：登记全局，供优雅退出用 ---- + ALL_WORKERS.append(w) 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 - current_real_url = None - except Exception as e: - logger.warning(f"URL 列表热更新失败：{e}") - - if current_proc and current_url: - 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 = None - current_url = None - current_real_url = None - 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 # 容错：无法判断时先认为还活着 - - 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="") - - # 定期刷新真实播放地址，避免 token 过期 - try: - if elapsed >= refresh_real_url_every and (elapsed % refresh_real_url_every) < check_interval_live: - new_port = asyncio.run(fetch_port_info(current_url, record_quality_code)) - new_real_url = select_source_url(current_url, new_port) - if new_real_url and new_real_url != current_real_url: - color.print_colored(f"\n[{now_str()}] 刷新播放地址（token 变化），重启 ffmpeg…", color.YELLOW) - stop_ffmpeg(current_proc) - out_url, v_args, a_args = build_youtube_output() - headers = headers_for_platform(current_platform, current_url) - current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args=v_args, a_args=a_args) - current_real_url = new_real_url - start_time = time.time() - time.sleep(check_interval_live) - continue - except Exception as e: - logger.warning(f"刷新播放地址失败：{e}") - - # ✅ watchdog（推流开始 > 120s 且 10 分钟完全无输出） - 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()}] 超过10分钟未见 ffmpeg 输出，判定异常，重启…", color.YELLOW) - stop_ffmpeg(current_proc) - current_proc = None - current_url = None - current_real_url = None - time.sleep(1) - continue - - 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 - current_real_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, 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 - current_real_url = real_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 + while not SHUTDOWN_EVENT.is_set(): + time.sleep(0.5) except KeyboardInterrupt: - print("\n收到中断，正在退出…") - except Exception as e: - logger.error(f"主循环异常：{e}") + graceful_shutdown() finally: - stop_ffmpeg(current_proc) - print("已退出。") + # 二次保险，确保全部退出 + graceful_shutdown() + for w in workers: + w.join(timeout=10) + print("全部 worker 已退出。") if __name__ == "__main__": main()