From 3c92b691dbd74077ab26afc3721d98954258e698 Mon Sep 17 00:00:00 2001 From: eric Date: Sun, 5 Oct 2025 11:06:47 +0800 Subject: [PATCH] 's' --- back/mult4 | 775 ++++++++++++++++++++++++++++++++++++++++++ config/URL_config.ini | 6 +- 2 files changed, 780 insertions(+), 1 deletion(-) create mode 100644 back/mult4 diff --git a/back/mult4 b/back/mult4 new file mode 100644 index 0000000..fc1df56 --- /dev/null +++ b/back/mult4 @@ -0,0 +1,775 @@ +# -*- encoding: utf-8 -*- + +""" +并行转推（日志优化版，保持原业务逻辑） +- 保留 Worker-1/2 等前缀 +- 日志状态变化才打印，避免刷屏 +- 禁止同一个 URL 推到同一个 YouTube key（(url,key) 级别防重复） +- 仍然避免同一个 URL 被多路并行占用（url 级别独占） +- 新增：Ctrl+C/TERM 时优雅停止所有 Worker 与 ffmpeg 子进程（仅最小增改） +""" + +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.3-optimized" +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() + +# ---- 新增：全局 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(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, defv): + try: + v = int(v) + if v < lo or v > hi: return defv + return v + except Exception: + return defv + +# =============== 工具函数 =============== +def _round_res(w: int, h: int) -> Tuple[int, int]: + # 偶数对齐，避免 1088x1920 类问题 + 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 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) + 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 +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 + 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 + +# =============== 构建输出参数（每个 key） =============== +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, + 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", + "-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+flush_packets", # 自动生成时间戳 + 忽略乱序 + 刷新缓冲 + "-use_wallclock_as_timestamps", "1", # 用系统时钟修正时间戳 + "-avioflags", "direct", # 减少延迟 + "-rtbufsize", "100M", # 输入缓冲 + "-err_detect", "ignore_err", # 忽略解码错误 + "-avoid_negative_ts", "make_zero", # 避免负时间戳 + ] + + # 视频/音频编码参数（使用 build_youtube_output 的结果） + v_args = v_args or [] + a_args = a_args or [] + base += v_args + a_args + + # 输出（flv 到 YT，更宽容） + base += ["-f", "flv", "-flvflags", "no_duration_filesize", out_url] + + + + cmd_str = " ".join(base) + 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": + 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() + + # ---- 新增：注册进程，便于 Ctrl+C 统一清理 ---- + with ALL_LOCK: + ALL_PROCS.add(p) + + 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() + + # 只保留关键统计/错误（前缀带 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"{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 + + 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 + finally: + # ---- 新增：从全局集合移除 ---- + 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 = { + '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 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 独占） =============== +class UrlCoordinator: + """ + - urls 顺序分配 + - in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用） + """ + 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() # 只保留 url 独占 + + 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]]: + """尝试遍历一轮 URL 表，找到首个在播的；否则返回 None。""" + with self.lock: + total = len(self.urls) + if total == 0: + return None + + tried = 0 + while tried < total: + with self.lock: + url = self.urls[self.idx] + self.idx = (self.idx + 1) % total + + # 只要 url 已占用就跳过 + if url in self.in_use_urls: + tried += 1 + continue + + # 锁外解析 + port = asyncio.run(fetch_port_info(url, quality_code)) + if not port.get("anchor_name"): + LOGSTATE.set_status(url, "unknown", text=f"解析失败：{url}") + tried += 1 + continue + if not port.get("is_live"): + anchor = port.get('anchor_name', '') + LOGSTATE.set_status(url, "offline", text=f"未开播：{anchor} | {url}") + tried += 1 + continue + + # 回到锁内占用 + with self.lock: + if url in self.in_use_urls: + tried += 1 + continue + self.in_use_urls.add(url) + + anchor = port.get('anchor_name', '') + LOGSTATE.set_status(url, "online", text=f"已开播：{anchor} | {url}") + return url, port + + return None + + def release(self, url: Optional[str], worker_key: Optional[str]): + if not url: + return + with self.lock: + self.in_use_urls.discard(url) + +# =============== 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() + self.start_time = time.time() + + 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(): + try: + 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}") + time.sleep(2) + continue + + 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("-----------------------------------------------------") + print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") + print("模式：并行检测 + 并行推流（保持原有业务逻辑）") + 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（并行路数 = 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) + + # 可选：每路分辨率 + per_res = _yt_get_list("youtube", "resolutions") # 如：1920x1080,1080x1920 + while per_res and len(per_res) < len(keys): + per_res.append(per_res[-1]) + + # 打印 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 not SHUTDOWN_EVENT.is_set(): + time.sleep(0.5) + except KeyboardInterrupt: + graceful_shutdown() + finally: + graceful_shutdown() + 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 7aaad34..76c19c3 100644 --- a/config/URL_config.ini +++ b/config/URL_config.ini @@ -1,4 +1,8 @@ -https://live.douyin.com/50627104055 +https://live.douyin.com/16531926455 +https://live.douyin.com/572442812068 +https://live.douyin.com/477961427964 +https://live.douyin.com/686507122231 +https://live.douyin.com/50627104055 https://live.douyin.com/339454333072 https://live.douyin.com/72212955083 https://live.douyin.com/621118100231