's'

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()

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