From 019f68a91ab0a8fa8ccff6efa592065baf1ea265 Mon Sep 17 00:00:00 2001 From: eric Date: Wed, 15 Oct 2025 16:15:46 +0800 Subject: [PATCH] 's' --- back/mult5 | 907 ++++++++++++++++++++++++++++++++++++++++++ config/URL_config.ini | 7 +- main.py | 734 +++++++++++++++++++++++----------- 3 files changed, 1410 insertions(+), 238 deletions(-) create mode 100644 back/mult5 diff --git a/back/mult5 b/back/mult5 new file mode 100644 index 0000000..87aecb3 --- /dev/null +++ b/back/mult5 @@ -0,0 +1,907 @@ +# -*- encoding: utf-8 -*- + +""" +并行转推（稳定性增强版，保持原业务逻辑与对外接口） +- 核心修复：降低“误判下播”的概率（抖音仍在播但 API 偶发失败） + * fetch_port_info：加入重试（3 次，指数退避），不改签名/调用处 + * “下播防抖”：只有在“连续N次 is_live=False 且最近M秒没有任何活跃迹象”才判定下播 + * 交叉验证：若 is_live=False，但当前 real_url 可探测（探活成功），则判为仍在播 +- FFmpeg 看门狗：卡顿/长时间无输出 -> 先软重启（保留 URL 占用），再必要时释放 +- 细节增强：日志更明确，解释每次判定原因；优雅退出确保无僵尸进程 +- 兼容性：不改配置文件/外部模块（spider、stream 等）的接口与行为 +- 修复：确保函数定义顺序，避免 NameError +""" + +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 contextlib + +import configparser +import urllib.request +import urllib.error + +from src import spider, stream +from src.utils import logger, Color, check_md5 +from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path + +# =============== 构建输出参数（每个 key） ———— 确保在此定义，早于 StreamWorker =============== +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 + + +# =============== 基础环境 =============== +version = "vrelay-parallel-2.1.1-pro" +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]: + """ + 选源优先级保持不变：抖音/TikTok 优先 FLV（且避免 H265），否则 record_url/m3u8/flv。 + """ + 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") + +# =============== 辅助：探测播放 URL 是否可用（用于误判纠正） =============== +def _open_with_proxy(url: str, timeout: float = 5.0): + """ + 使用 urllib 在当前代理环境下打开 URL。 + 仅抓取少量字节用于“活跃性探测”，避免大流量。 + """ + handlers = [] + if proxy_addr: + handlers.append(urllib.request.ProxyHandler({"http": proxy_addr, "https": proxy_addr})) + opener = urllib.request.build_opener(*handlers) if handlers else urllib.request.build_opener() + req = urllib.request.Request( + url, + headers={ + "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", + "Accept": "*/*", + "Connection": "close", + } + ) + return opener.open(req, timeout=timeout) + +def probe_stream_url(real_url: Optional[str], timeout: float = 5.0) -> bool: + """ + 轻量级“探活”： + - m3u8: 取前若干字节并检查 "#EXTM3U" + - flv: 取前3字节是否 "FLV" + - 其它：只要 200 且有数据即可视为“活着” + 失败不抛异常，返回 False + """ + if not real_url: + return False + try: + with contextlib.closing(_open_with_proxy(real_url, timeout=timeout)) as resp: + code = getattr(resp, "status", None) or getattr(resp, "code", None) + if code and int(code) >= 400: + return False + head = resp.read(256) # 少量字节足够判定格式 + u = real_url.lower() + if ".m3u8" in u: + return b"#EXTM3U" in head.upper() + if ".flv" in u: + return head.startswith(b"FLV") + # 其它类型：有数据即认为仍在 + return bool(head) + except Exception: + return False + +# =============== 解析与选源（抖音/TikTok/自定义） =============== +# 修复点①：增强 fetch_port_info —— 重试 + 指数退避（不改签名/调用处） +async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: + proxy = proxy_addr + delay = 1.5 # 初始退避 + for attempt in range(1, 4): # 最多 3 次 + 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} + + # 返回前补充：若 is_live True 但未给出真实播放地址，也作为暂不可用 + if port_info.get("is_live") and not select_source_url(url, port_info): + logger.warning(f"[fetch_port_info] 在播但暂未拿到播放 URL，可能接口延迟：{url}") + return port_info + + except Exception as e: + logger.warning(f"[fetch_port_info] 第{attempt}次解析失败: {e}") + if attempt < 3: + try: + await asyncio.sleep(delay) + except Exception: + # 若在同步上下文内调用（意外），兜底 + time.sleep(delay) + delay *= 2 + else: + logger.error(f"[fetch_port_info] 多次失败，返回空状态: {url}") + return {"anchor_name": "", "is_live": False} + + +# =============== 启动/停止 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: + with contextlib.suppress(Exception): + p.stdin.write(b"q"); p.stdin.flush() + p.wait(timeout=6) + else: + p.send_signal(signal.SIGINT) + p.wait(timeout=10) + except Exception: + with contextlib.suppress(Exception): + p.kill() + 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() + + 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 + + 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 # 10 分钟无输出视为卡死 + self.refresh_real_url_every = 300 # 5 分钟刷新一次真实地址（token 可能过期） + 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() + + # 修复点②：新增下播防抖状态（连续多次 is_live=False 才判定） + self._offline_count = 0 + self._last_live_ts = time.time() + + # 软重启冷却（避免频繁重启） + self._last_soft_restart_ts = 0.0 + self._soft_restart_min_interval = 90.0 + + def stop(self): + """收到停止请求时，尽快停掉当前 ffmpeg。""" + self._stop_event.set() + with contextlib.suppress(Exception): + stop_ffmpeg(self.current_proc) + + def _p(self, text: str): + print(f"[{now_str()}][{self.worker_name}] {text}") + + def _soft_restart_ffmpeg(self, reason: str): + """ + 尝试“软重启”ffmpeg：仅重启进程，不释放 URL 占用，优先保持连续性。 + """ + now = time.time() + if now - self._last_soft_restart_ts < self._soft_restart_min_interval: + self._p(f"跳过软重启（冷却中）: {reason}") + return False + self._last_soft_restart_ts = now + + try: + stop_ffmpeg(self.current_proc) + except Exception: + pass + + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) # 修复：用 self.resolution，非 None + headers = headers_for_platform(self.current_platform, self.current_url or "") + if not self.current_real_url: + # 尝试重新解析一遍 + new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) if self.current_url else {} + self.current_real_url = select_source_url(self.current_url or "", new_port) + if not self.current_real_url: + self._p("软重启失败：无法获取播放 URL。") + return False + + self.current_proc = launch_ffmpeg(self.current_real_url, out_url, headers, v_args, a_args, self.worker_name) + self.start_time = time.time() + self._last_progress_print = -9999 + self._p(f"已软重启 ffmpeg：{reason}") + return True + + 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 + is_live_raw = True + try: + port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + is_live_raw = bool(port.get("is_live")) + except Exception as e: + logger.warning(f"[{self.worker_name}] 检测当前源异常：{e}") + is_live_raw = True # 网络/接口异常默认不下播 + + if not is_live_raw: + # 交叉探活：若当前 real_url 还能探测成功，认为仍在播 + alive_by_probe = probe_stream_url(self.current_real_url) + if alive_by_probe: + logger.info(f"[{self.worker_name}] API 返回下播但探活成功，视为仍在播。") + is_live_raw = True + + # 下播防抖：连续 N 次失败 且 距最后一次活跃已超过 M 秒 + if is_live_raw: + self._offline_count = 0 + self._last_live_ts = time.time() + still_live = True + else: + self._offline_count += 1 + if time.time() - self._last_live_ts < 120: + logger.info(f"[{self.worker_name}] 短期异常 is_live=False（{self._offline_count}），已忽略") + still_live = True + elif self._offline_count < 3: + logger.info(f"[{self.worker_name}] 疑似下播 {self._offline_count}/3 次，继续观望") + still_live = True + else: + still_live = False # 多次确认才下播 + + 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) # 修复：用 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}") + + # 看门狗：长时间无输出 -> 先软重启，如果软重启失败再释放 + silence = time.time() - getattr(self.current_proc, "last_out_ts", 0) + if elapsed >= self.watchdog_grace and silence > self.watchdog_silence: + self._p("⚠️ 超过10分钟未见输出，尝试软重启…") + ok = self._soft_restart_ffmpeg("watchdog silence") + if not ok: + self._p("软重启失败，释放当前源并切换…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + time.sleep(1) + else: + time.sleep(self.check_interval_live) + 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 + + # 新源上线前：探活（避免刚拿到 URL 就空推） + if not probe_stream_url(real_url): + self._p(f"播放 URL 探活失败（可能刚开播或接口延迟），暂缓此源：{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.resolution，非 None + 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._offline_count = 0 + self._last_live_ts = time.time() + + 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: + with contextlib.suppress(Exception): + 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]) # 修复：统一用 stdout + 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): + with contextlib.suppress(Exception): + w.stop() + # 再兜底停止全局登记的 ffmpeg 子进程 + with ALL_LOCK: + procs = list(ALL_PROCS) + for p in procs: + with contextlib.suppress(Exception): + stop_ffmpeg(p) + +def main(): + print("DouyinRelay", version) + 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） + with contextlib.suppress(Exception): + signal.signal(signal.SIGINT, graceful_shutdown) + with contextlib.suppress(Exception): + signal.signal(signal.SIGTERM, graceful_shutdown) + + # 启动多路（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() \ No newline at end of file diff --git a/config/URL_config.ini b/config/URL_config.ini index 76c19c3..7389d00 100644 --- a/config/URL_config.ini +++ b/config/URL_config.ini @@ -1,4 +1,8 @@ -https://live.douyin.com/16531926455 +https://live.douyin.com/782322393954 +https://live.douyin.com/16531926455 +https://live.douyin.com/739670432305 +https://live.douyin.com/486789870173 +https://live.douyin.com/15652099787 https://live.douyin.com/572442812068 https://live.douyin.com/477961427964 https://live.douyin.com/686507122231 @@ -14,6 +18,7 @@ https://live.douyin.com/122383435921 https://live.douyin.com/92876979328 https://live.douyin.com/517715534931 https://live.douyin.com/460525712926 +https://live.douyin.com/339454333072 https://live.douyin.com/456576000931 https://live.douyin.com/745606325880 https://live.douyin.com/749687541944 diff --git a/main.py b/main.py index fc1df56..6e9cffd 100644 --- a/main.py +++ b/main.py @@ -1,12 +1,15 @@ # -*- encoding: utf-8 -*- """ -并行转推（日志优化版，保持原业务逻辑） -- 保留 Worker-1/2 等前缀 -- 日志状态变化才打印，避免刷屏 -- 禁止同一个 URL 推到同一个 YouTube key（(url,key) 级别防重复） -- 仍然避免同一个 URL 被多路并行占用（url 级别独占） -- 新增：Ctrl+C/TERM 时优雅停止所有 Worker 与 ffmpeg 子进程（仅最小增改） +并行转推（稳定性+体验增强版，保持原业务逻辑与对外接口） +- 核心修复：降低“误判下播”的概率（抖音仍在播但 API 偶发失败） + * fetch_port_info：加入重试（3 次，指数退避），不改签名/调用处 + * “下播防抖”：只有在“连续N次 is_live=False 且最近M秒没有任何活跃迹象”才判定下播 + * 交叉验证：若 is_live=False，但当前 real_url 可探测（探活成功），则判为仍在播 +- FFmpeg 看门狗：卡顿/长时间无输出 -> 先软重启（保留 URL 占用），再必要时释放 +- 细节增强：日志更明确，解释每次判定原因；优雅退出确保无僵尸进程 +- 兼容性：不改配置文件/外部模块（spider、stream 等）的接口与行为 +- 修复：确保函数定义顺序，避免 NameError """ import asyncio @@ -20,49 +23,145 @@ import uuid import threading import subprocess from typing import Any, List, Tuple, Optional, Dict, Set - +import contextlib import configparser import urllib.request +import urllib.error +import io +import logging +from logging.handlers import RotatingFileHandler +# ======================= 颜色日志（Windows Cmder 友好） ======================= +def _enable_vt_on_windows(): + """在 Windows 控制台启用 ANSI VT（Cmder/CMD/PowerShell 颜色支持）""" + if os.name != "nt": + return + try: + import msvcrt # noqa: F401 + import ctypes + kernel32 = ctypes.windll.kernel32 + handle = kernel32.GetStdHandle(-11) # STD_OUTPUT_HANDLE = -11 + mode = ctypes.c_uint32() + if kernel32.GetConsoleMode(handle, ctypes.byref(mode)): + mode.value |= 0x0004 # ENABLE_VIRTUAL_TERMINAL_PROCESSING + kernel32.SetConsoleMode(handle, mode) + except Exception: + pass + +_enable_vt_on_windows() + +class ColorLog: + # ANSI colors + RESET = "\033[0m" + GREY = "\033[90m" + RED = "\033[91m" + GREEN = "\033[92m" + YELLOW = "\033[93m" + BLUE = "\033[94m" + MAGENTA = "\033[95m" + CYAN = "\033[96m" + WHITE = "\033[97m" + + ICONS = { + "INFO": "🟢", + "WARN": "🟡", + "ERROR": "🔴", + "DEBUG": "⚪", + "OK": "✅", + "SWAP": "🔁", + "LIVE": "📡", + "OFF": "📴", + "FFMPEG":"🎬", + "WATCH": "👀", + } + + COLORS = { + "INFO": GREEN, + "OK": GREEN, + "WARN": YELLOW, + "ERROR": RED, + "DEBUG": GREY, + "SWAP": CYAN, + "LIVE": BLUE, + "OFF": RED, + "FFMPEG": MAGENTA, + "WATCH": CYAN, + } + + def __init__(self, log_file_path: str = "logs/runtime.log", max_bytes: int = 5*1024*1024, backup: int = 3): + os.makedirs(os.path.dirname(log_file_path), exist_ok=True) + + self._logger = logging.getLogger("Runtime") + self._logger.setLevel(logging.INFO) + self._logger.propagate = False + # 控制台 Handler（给 src.utils.logger 让位，这里只用于文件） + if not self._logger.handlers: + file_handler = RotatingFileHandler(log_file_path, maxBytes=max_bytes, backupCount=backup, encoding="utf-8") + fmt = logging.Formatter("%(asctime)s | %(levelname)-5s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S") + file_handler.setFormatter(fmt) + self._logger.addHandler(file_handler) + + def _stamp(self) -> str: + return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + def _compose(self, level: str, msg: str, worker: Optional[str] = None) -> str: + icon = self.ICONS.get(level, "") + color = self.COLORS.get(level, self.WHITE) + prefix = f"[{self._stamp()}]" + if worker: + prefix += f"[{worker}]" + return f"{prefix} {color}[{icon} {level}]{self.RESET} {msg}" + + def info(self, msg: str, worker: Optional[str] = None): + text = self._compose("INFO", msg, worker) + print(text) + self._logger.info(msg) + + def ok(self, msg: str, worker: Optional[str] = None): + text = self._compose("OK", msg, worker) + print(text) + self._logger.info(msg) + + def warn(self, msg: str, worker: Optional[str] = None): + text = self._compose("WARN", msg, worker) + print(text) + self._logger.warning(msg) + + def error(self, msg: str, worker: Optional[str] = None): + text = self._compose("ERROR", msg, worker) + print(text, file=sys.stderr) + self._logger.error(msg) + + def debug(self, msg: str, worker: Optional[str] = None): + text = self._compose("DEBUG", msg, worker) + print(text) + self._logger.debug(msg) + + def swap(self, msg: str, worker: Optional[str] = None): + text = self._compose("SWAP", msg, worker) + print(text) + self._logger.info("[SWAP] " + msg) + + def live(self, msg: str, worker: Optional[str] = None): + text = self._compose("LIVE", msg, worker) + print(text) + self._logger.info("[LIVE] " + msg) + + def off(self, msg: str, worker: Optional[str] = None): + text = self._compose("OFF", msg, worker) + print(text) + self._logger.info("[OFF] " + msg) + +CLOG = ColorLog() # 全局彩色日志 + +# ======================= 这里开始是原有工程的 import =========================== 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] +# =============== 重要：先定义 build_youtube_output_for_key，避免 NameError =============== +# 读取 YouTube 配置（需要在函数内部访问 _yt_get，故先声明占位；稍后会正式赋值） +_yt_cfg: Optional[configparser.ConfigParser] = None def _safe_int(v, lo, hi, defv): try: @@ -72,7 +171,6 @@ def _safe_int(v, lo, hi, defv): except Exception: return defv -# =============== 工具函数 =============== def _round_res(w: int, h: int) -> Tuple[int, int]: # 偶数对齐，避免 1088x1920 类问题 return (w & ~1, h & ~1) @@ -84,121 +182,11 @@ 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) - 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]: + """ + ！！！此函数放在文件靠前位置，确保任何地方调用都不再出现 NameError。 + 与旧版保持一致；仅对异常进行更稳妥处理与注释增强。 + """ if not key: logger.error("错误: youtube.ini 未配置 key/keys") sys.exit(1) @@ -250,6 +238,222 @@ def build_youtube_output_for_key(key: str, resolution: Optional[str]) -> Tuple[s return out_url, v_args, a_args +# =============== 基础环境 =============== +version = "vrelay-parallel-2.1.1-pro" +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 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) + CLOG.info("你的网络似乎直连可用（无需代理）。") +except Exception: + if not use_proxy: + print("\033[93m提示：无法直连海外站，如需 TikTok 请在 config.ini 开启并填写代理地址。\033[0m") + +# =============== URL 列表 =============== +def parse_urls_from_file(path: str) -> List[str]: + urls, seen = [], set() + if not os.path.isfile(path): + os.makedirs(os.path.dirname(path), exist_ok=True) + 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]: + """ + 选源优先级保持不变：抖音/TikTok 优先 FLV（且避免 H265），否则 record_url/m3u8/flv。 + """ + 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") + +# =============== 辅助：探测播放 URL 是否可用（用于误判纠正） =============== +def _open_with_proxy(url: str, timeout: float = 5.0): + """ + 使用 urllib 在当前代理环境下打开 URL。 + 仅抓取少量字节用于“活跃性探测”，避免大流量。 + """ + handlers = [] + if proxy_addr: + handlers.append(urllib.request.ProxyHandler({"http": proxy_addr, "https": proxy_addr})) + opener = urllib.request.build_opener(*handlers) if handlers else urllib.request.build_opener() + req = urllib.request.Request( + url, + headers={ + "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", + "Accept": "*/*", + "Connection": "close", + } + ) + return opener.open(req, timeout=timeout) + +def probe_stream_url(real_url: Optional[str], timeout: float = 5.0) -> bool: + """ + 轻量级“探活”： + - m3u8: 取前若干字节并检查 "#EXTM3U" + - flv: 取前3字节是否 "FLV" + - 其它：只要 200 且有数据即可视为“活着” + 失败不抛异常，返回 False + """ + if not real_url: + return False + try: + with contextlib.closing(_open_with_proxy(real_url, timeout=timeout)) as resp: + code = getattr(resp, "status", None) or getattr(resp, "code", None) + if code and int(code) >= 400: + return False + head = resp.read(256) # 少量字节足够判定格式 + u = real_url.lower() + if ".m3u8" in u: + return b"#EXTM3U" in head.upper() + if ".flv" in u: + return head.startswith(b"FLV") + # 其它类型：有数据即认为仍在 + return bool(head) + except Exception: + return False + +# =============== 解析与选源（抖音/TikTok/自定义） =============== +# 修复点①：增强 fetch_port_info —— 重试 + 指数退避（不改签名/调用处） +async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: + proxy = proxy_addr + delay = 1.5 # 初始退避 + for attempt in range(1, 4): # 最多 3 次 + 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} + + # 返回前补充：若 is_live True 但未给出真实播放地址，也作为暂不可用（但不判下播） + if port_info.get("is_live") and not select_source_url(url, port_info): + logger.warning(f"[fetch_port_info] 在播但暂未拿到播放 URL，可能接口延迟：{url}") + return port_info + + except Exception as e: + logger.warning(f"[fetch_port_info] 第{attempt}次解析失败: {e}") + if attempt < 3: + try: + await asyncio.sleep(delay) + except Exception: + time.sleep(delay) + delay *= 2 + else: + logger.error(f"[fetch_port_info] 多次失败，返回空状态: {url}") + return {"anchor_name": "", "is_live": False} + # =============== 启动/停止 ffmpeg =============== def launch_ffmpeg(input_url: str, out_url: str, @@ -302,12 +506,12 @@ def launch_ffmpeg(input_url: str, # 通用容错 + 时间戳修复（优化版） base += [ - "-fflags", "+genpts+igndts+flush_packets", # 自动生成时间戳 + 忽略乱序 + 刷新缓冲 - "-use_wallclock_as_timestamps", "1", # 用系统时钟修正时间戳 - "-avioflags", "direct", # 减少延迟 - "-rtbufsize", "100M", # 输入缓冲 - "-err_detect", "ignore_err", # 忽略解码错误 - "-avoid_negative_ts", "make_zero", # 避免负时间戳 + "-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 的结果） @@ -318,10 +522,9 @@ def launch_ffmpeg(input_url: str, # 输出（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] + " ...") + # 控制台+文件日志 + CLOG.live(f"FFmpeg CMD({worker_name}): {cmd_str if len(cmd_str) < 500 else cmd_str[:500] + ' ...'}", worker=worker_name) logger.info(f"{worker_name} 启动推流：{input_url} -> {out_url}") creationflags = 0 @@ -339,7 +542,7 @@ def launch_ffmpeg(input_url: str, ) p.last_out_ts = time.time() - # ---- 新增：注册进程，便于 Ctrl+C 统一清理 ---- + # ---- 注册进程，便于 Ctrl+C 统一清理 ---- with ALL_LOCK: ALL_PROCS.add(p) @@ -353,11 +556,11 @@ def launch_ffmpeg(input_url: str, # 只保留关键统计/错误（前缀带 Worker） if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text) or ("libx264" in text): - logger.info(f"{worker_name} [ffmpeg统计] {text}") + CLOG.info(f"[ffmpeg统计] {text}", worker=worker_name) if "Server disconnected" in text: - logger.error(f"{worker_name} [输出异常] {text}") + CLOG.error(f"[输出异常] {text}", worker=worker_name) elif ("not supported" in text) or ("Invalid" in text) or ("invalid" in text): - logger.error(f"{worker_name} [转码异常] {text}") + CLOG.error(f"[转码异常] {text}", worker=worker_name) except Exception: pass @@ -369,19 +572,17 @@ def stop_ffmpeg(p: Optional[subprocess.Popen]): try: if os.name == "nt": if p.stdin: - try: + with contextlib.suppress(Exception): 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 + with contextlib.suppress(Exception): + p.kill() finally: - # ---- 新增：从全局集合移除 ---- + # ---- 从全局集合移除 ---- with ALL_LOCK: ALL_PROCS.discard(p) @@ -425,7 +626,7 @@ class LogState: 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}] 当前无可用直播源，等待中…") + CLOG.debug("当前无可用直播源，等待中…", worker=worker_name) self.no_source_worker[worker_name] = True def clear_idle_flag(self, worker_name: str): @@ -434,7 +635,7 @@ class LogState: LOGSTATE = LogState() -# =============== URL 协调器（只保留 url 独占） =============== +# =============== URL 协调器（url 级独占，保持原顺序） =============== class UrlCoordinator: """ - urls 顺序分配 @@ -446,8 +647,7 @@ class UrlCoordinator: self.idx: int = 0 self.lock = threading.Lock() self.last_md5 = check_md5(path) - - self.in_use_urls: Set[str] = set() # 只保留 url 独占 + self.in_use_urls: Set[str] = set() def refresh_if_changed(self): try: @@ -455,7 +655,7 @@ class UrlCoordinator: 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)}") + CLOG.info(f"已重载 URL 列表，数量：{len(self.urls)}") except Exception as e: logger.warning(f"URL 列表热更新失败：{e}") @@ -472,19 +672,18 @@ class UrlCoordinator: 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)) + anchor = port.get('anchor_name', '') 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 @@ -496,7 +695,6 @@ class UrlCoordinator: continue self.in_use_urls.add(url) - anchor = port.get('anchor_name', '') LOGSTATE.set_status(url, "online", text=f"已开播：{anchor} | {url}") return url, port @@ -508,7 +706,7 @@ class UrlCoordinator: with self.lock: self.in_use_urls.discard(url) -# =============== Worker（保留前缀日志） =============== +# =============== Worker（保持原业务逻辑框架，加入防抖与探活） =============== class StreamWorker(threading.Thread): def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator): super().__init__(daemon=True) @@ -525,53 +723,112 @@ class StreamWorker(threading.Thread): 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.watchdog_grace = 120 # 推流启动后宽限期 + self.watchdog_silence = 600 # 10 分钟无输出视为卡死 + self.refresh_real_url_every = 300 # 5 分钟刷新一次真实地址（token 可能过期） + 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() + # 修复点②：新增下播防抖状态（连续多次 is_live=False 才判定） + self._offline_count = 0 + self._last_live_ts = time.time() + + # 软重启冷却（避免频繁重启） + self._last_soft_restart_ts = 0.0 + self._soft_restart_min_interval = 90.0 + def stop(self): - """最小改动：收到停止请求时，立即尝试停掉当前 ffmpeg，尽快响应退出。""" + """收到停止请求时，尽快停掉当前 ffmpeg。""" self._stop_event.set() + with contextlib.suppress(Exception): + stop_ffmpeg(self.current_proc) + + def _soft_restart_ffmpeg(self, reason: str): + """ + 尝试“软重启”ffmpeg：仅重启进程，不释放 URL 占用，优先保持连续性。 + """ + now = time.time() + if now - self._last_soft_restart_ts < self._soft_restart_min_interval: + CLOG.warn(f"跳过软重启（冷却中）: {reason}", worker=self.worker_name) + return False + self._last_soft_restart_ts = now + try: stop_ffmpeg(self.current_proc) except Exception: pass - def _p(self, text: str): - print(f"[{now_str()}][{self.worker_name}] {text}") + 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 or "") + if not self.current_real_url: + # 尝试重新解析一遍 + new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) if self.current_url else {} + self.current_real_url = select_source_url(self.current_url or "", new_port) + if not self.current_real_url: + CLOG.error("软重启失败：无法获取播放 URL。", worker=self.worker_name) + return False + + self.current_proc = launch_ffmpeg(self.current_real_url, out_url, headers, v_args, a_args, self.worker_name) + self.start_time = time.time() + self._last_progress_print = -9999 + CLOG.ok(f"已软重启 ffmpeg：{reason}", worker=self.worker_name) + return True def run(self): - self._p("启动") + CLOG.ok("启动", worker=self.worker_name) 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})，切换下一个源…") + CLOG.error(f"ffmpeg 退出(code={retcode})，切换下一个源…", worker=self.worker_name) 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 + is_live_raw = True try: port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) - still_live = bool(port.get("is_live")) + is_live_raw = bool(port.get("is_live")) except Exception as e: - logger.error(f"[{self.worker_name}] 检测当前源失败：{e}") + CLOG.warn(f"检测当前源异常：{e}", worker=self.worker_name) + is_live_raw = True # 网络/接口异常默认不下播 + + if not is_live_raw: + # 交叉探活：若当前 real_url 还能探测成功，认为仍在播 + alive_by_probe = probe_stream_url(self.current_real_url) + if alive_by_probe: + CLOG.warn("API 返回下播但探活成功，视为仍在播。", worker=self.worker_name) + is_live_raw = True + + # 下播防抖：连续 N 次失败 且 距最后一次活跃已超过 M 秒 + if is_live_raw: + self._offline_count = 0 + self._last_live_ts = time.time() still_live = True + else: + self._offline_count += 1 + if time.time() - self._last_live_ts < 120: + CLOG.warn(f"短期异常 is_live=False（{self._offline_count}），已忽略", worker=self.worker_name) + still_live = True + elif self._offline_count < 3: + CLOG.warn(f"疑似下播 {self._offline_count}/3 次，继续观望", worker=self.worker_name) + still_live = True + else: + still_live = False # 多次确认才下播 if still_live: elapsed = int(time.time() - self.start_time) @@ -580,7 +837,7 @@ class StreamWorker(threading.Thread): 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}") + CLOG.live(f"正在转播({dur_str})：{self.current_platform} | {self.current_anchor}", worker=self.worker_name) self._last_progress_print = elapsed # 定期刷新真实播放地址（token） @@ -589,7 +846,7 @@ class StreamWorker(threading.Thread): 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…") + CLOG.swap("刷新播放地址（token 变化），软重启 ffmpeg…", worker=self.worker_name) 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) @@ -600,28 +857,33 @@ class StreamWorker(threading.Thread): time.sleep(self.check_interval_live) continue except Exception as e: - logger.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}") + CLOG.warn(f"刷新播放地址失败：{e}", worker=self.worker_name) - # 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) + CLOG.warn("超过10分钟未见输出，尝试软重启…", worker=self.worker_name) + ok = self._soft_restart_ffmpeg("watchdog silence") + if not ok: + CLOG.warn("软重启失败，释放当前源并切换…", worker=self.worker_name) + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + time.sleep(1) + else: + time.sleep(self.check_interval_live) continue time.sleep(self.check_interval_live) continue else: - self._p("📴 主播停播，切换下一个源…") + CLOG.off("多次确认主播下播，切换下一个源…", worker=self.worker_name) 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) @@ -632,7 +894,14 @@ class StreamWorker(threading.Thread): url, port = got real_url = select_source_url(url, port) if not real_url: - self._p(f"未得到播放地址，跳过：{url}") + CLOG.warn(f"未得到播放地址，跳过：{url}", worker=self.worker_name) + self.coordinator.release(url, self.key) + time.sleep(1) + continue + + # 新源上线前：探活（避免刚拿到 URL 就空推） + if not probe_stream_url(real_url): + CLOG.warn(f"播放 URL 探活失败（可能刚开播或接口延迟），暂缓此源：{url}", worker=self.worker_name) self.coordinator.release(url, self.key) time.sleep(1) continue @@ -646,19 +915,22 @@ class StreamWorker(threading.Thread): self.current_real_url = real_url self.start_time = time.time() self._last_progress_print = -9999 + self._offline_count = 0 + self._last_live_ts = time.time() - self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}") + CLOG.ok(f"开始转播：{self.current_platform} | {self.current_anchor} | {url}", worker=self.worker_name) except Exception as e: - # 🔧 关键：任何内部异常不让线程退出 - logger.error(f"[{self.worker_name}] 内部异常：{e}") + # 关键：任何内部异常不让线程退出 + CLOG.error(f"内部异常：{e}", worker=self.worker_name) time.sleep(2) continue finally: - stop_ffmpeg(self.current_proc) + with contextlib.suppress(Exception): + stop_ffmpeg(self.current_proc) self.coordinator.release(self.current_url, self.key) - self._p("已退出。") + CLOG.off("已退出。", worker=self.worker_name) def _clear_state(self, release: bool = False): if release and self.current_url: @@ -674,39 +946,31 @@ 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]) + 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: + with contextlib.suppress(Exception): w.stop() - except Exception: - pass # 再兜底停止全局登记的 ffmpeg 子进程 with ALL_LOCK: procs = list(ALL_PROCS) for p in procs: - try: + with contextlib.suppress(Exception): stop_ffmpeg(p) - except Exception: - pass def main(): - print("-----------------------------------------------------") - print("| DouyinLiveRecorder - 并行转推版 |") - print("-----------------------------------------------------") - print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") - print("模式：并行检测 + 并行推流（保持原有业务逻辑）") - print(".....................................................") + print("DouyinRelay", version) + print("-----------------------------------------") if not check_ffmpeg_existence(): logger.error("缺少 ffmpeg，程序退出") @@ -742,14 +1006,10 @@ def main(): coordinator = UrlCoordinator(url_config_file) # 信号处理（Ctrl+C / kill） - try: + with contextlib.suppress(Exception): signal.signal(signal.SIGINT, graceful_shutdown) - except Exception: - pass - try: + with contextlib.suppress(Exception): signal.signal(signal.SIGTERM, graceful_shutdown) - except Exception: - pass # 启动多路（Worker 前缀区分） workers: List[StreamWorker] = []