'..'

main.py

@@ -29,7 +29,7 @@ from src.utils import logger, Color, check_md5
 from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
 
 # =============== 基础环境 ===============
-version = "relay-parallel-2.0.1-logonly"
+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'
@@ -84,7 +84,7 @@ 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(): 
+def now_str():
     return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
 # =============== 读取运行配置 ===============
@@ -127,7 +127,7 @@ def parse_urls_from_file(path: str) -> List[str]:
     with open(path, "r", encoding=text_encoding, errors='ignore') as f:
         for line in f:
             s = line.strip()
-            if not s or s.startswith("#"): 
+            if not s or s.startswith("#"):
                 continue
             parts = re.split(r"[，,]\s*", s)
             cand = None
@@ -260,9 +260,11 @@ def launch_ffmpeg(input_url: str,
     user_agent = ("Mozilla/5.0 (Linux; Android 11; SM-G973U) "
                   "AppleWebKit/537.36 (KHTML, like Gecko) "
                   "Chrome/87.0.4280.141 Mobile Safari/537.36")
+    # 输入侧容错/重连/探测
     base = [
         "ffmpeg", "-nostdin", "-hide_banner",
-        "-stats", "-stats_period", "60",
+        "-stats",
+        "-stats_period", "60",
         "-progress", "pipe:1",
         "-loglevel", "info",
 
@@ -274,34 +276,49 @@ def launch_ffmpeg(input_url: str,
         "-reconnect_at_eof", "1",
         "-reconnect_delay_max", "30",
 
+        # UA/协议白名单
         "-user_agent", user_agent,
         "-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy",
 
+        # 缓冲 & 探测
         "-thread_queue_size", "4096",
         "-analyzeduration", "10000000",
         "-probesize", "10000000",
     ]
+
+    # 平台 headers（referer/origin）要在 -i 之前
     if extra_headers:
         base += ["-headers", extra_headers]
 
+    # 代理：http_proxy/https_proxy 环境变量（同时保留 -http_proxy 兼容）
     env = os.environ.copy()
     if proxy_addr:
         env["http_proxy"] = proxy_addr
         env["https_proxy"] = proxy_addr
         base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:]
 
+    # 输入
     base += ["-re", "-i", input_url]
 
+    # 通用容错 + 时间戳修复（优化版）
     base += [
-        "-fflags", "+genpts+igndts+discardcorrupt+nobuffer",
-        "-xerror",
-        "-use_wallclock_as_timestamps", "1",
-        "-avioflags", "direct",
-        "-rtbufsize", "100M",
+        "-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
-    base += ["-f", "flv", out_url]
+
+    # 输出（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] + " ...")
@@ -417,12 +434,11 @@ class LogState:
 
 LOGSTATE = LogState()
 
-# =============== URL 协调器（包含 url 与 (url,key) 占用） ===============
+# =============== URL 协调器（只保留 url 独占） ===============
 class UrlCoordinator:
-    """ 
+    """
     - urls 顺序分配
     - in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用）
-    - in_use_pairs: (url,key) 级别独占（禁止同一 url 推到同一个 key）
     """
     def __init__(self, path: str):
         self.path = path
@@ -431,8 +447,7 @@ class UrlCoordinator:
         self.lock = threading.Lock()
         self.last_md5 = check_md5(path)
 
-        self.in_use_urls: Set[str] = set()
-        self.in_use_pairs: Set[Tuple[str, str]] = set()  # (url, key)
+        self.in_use_urls: Set[str] = set()  # 只保留 url 独占
 
     def refresh_if_changed(self):
         try:
@@ -445,60 +460,53 @@ class UrlCoordinator:
             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:
-            if not self.urls:
-                return None
-            tried = 0
-            pick = None
-            while tried < len(self.urls):
+            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) % len(self.urls)
-                tried += 1
-                # 已被其他路占用 / 同 url-key 已经在用
+                self.idx = (self.idx + 1) % total
+
+                # 只要 url 已占用就跳过
                 if url in self.in_use_urls:
+                    tried += 1
                     continue
-                if (url, worker_key) in self.in_use_pairs:
+
+            # 锁外解析
+            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
-                pick = url
-                break
-            if not pick:
-                return None
+                self.in_use_urls.add(url)
 
-        # 锁外解析
-        port = asyncio.run(fetch_port_info(pick, quality_code))
-        if not port.get("anchor_name"):
-            LOGSTATE.set_status(pick, "unknown",
-                                text=f"解析失败：{pick}",
-                                prefix="")
-            return None
-        if not port.get("is_live"):
-            anchor = port.get('anchor_name','')
-            LOGSTATE.set_status(pick, "offline",
-                                text=f"未开播：{anchor} | {pick}",
-                                prefix="")
-            return None
+            anchor = port.get('anchor_name', '')
+            LOGSTATE.set_status(url, "online", text=f"已开播：{anchor} | {url}")
+            return url, port
 
-        # 占用
-        with self.lock:
-            if (pick in self.in_use_urls) or ((pick, worker_key) in self.in_use_pairs):
-                return None
-            self.in_use_urls.add(pick)
-            self.in_use_pairs.add((pick, worker_key))
-
-        # 状态更新（online）
-        anchor = port.get('anchor_name','')
-        LOGSTATE.set_status(pick, "online",
-                            text=f"已开播：{anchor} | {pick}",
-                            prefix="")
-        return pick, port
+        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)
-            if worker_key:
-                self.in_use_pairs.discard((url, worker_key))
 
 # =============== Worker（保留前缀日志） ===============
 class StreamWorker(threading.Thread):
@@ -526,6 +534,7 @@ class StreamWorker(threading.Thread):
 
         self._last_progress_print = -9999
         self._stop_event = threading.Event()
+        self.start_time = time.time()
 
     def stop(self):
         """最小改动：收到停止请求时，立即尝试停掉当前 ffmpeg，尽快响应退出。"""
@@ -542,106 +551,110 @@ class StreamWorker(threading.Thread):
         self._p("启动")
         try:
             while not self._stop_event.is_set():
-                self.coordinator.refresh_if_changed()
+                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
+                    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
+                            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.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}")
+                            logger.error(f"[{self.worker_name}] 检测当前源失败：{e}")
+                            still_live = True
 
-                        # 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分钟未见输出，重启…")
+                        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)
-                            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)
+                    # —— 未在推流：尝试分配一个新源（保持原顺序轮询 + 占用） ——
+                    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
 
-                # —— 未在推流：尝试分配一个新源（保持原顺序轮询 + 占用） ——
-                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)
+                    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
 
-                LOGSTATE.clear_idle_flag(self.worker_name)
-                url, port = got
-                real_url = select_source_url(url, port)
-                if not real_url:
-                    self._p(f"未得到播放地址，跳过：{url}")
-                    self.coordinator.release(url, self.key)
-                    time.sleep(1)
-                    continue
-
-                self.current_platform = detect_platform(url)
-                self.current_anchor = port.get("anchor_name", "")
-                headers = headers_for_platform(self.current_platform, url)
-                out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution)
-                self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args, self.worker_name)
-                self.current_url = url
-                self.current_real_url = real_url
-                self.start_time = time.time()
-                self._last_progress_print = -9999
-
-                self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}")
-
-        except Exception as e:
-            logger.error(f"[{self.worker_name}] 主循环异常：{e}")
         finally:
             stop_ffmpeg(self.current_proc)
             self.coordinator.release(self.current_url, self.key)
@@ -745,7 +758,6 @@ def main():
         w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator)
         w.start()
         workers.append(w)
-        # ---- 新增：登记全局，供优雅退出用 ----
         ALL_WORKERS.append(w)
 
     try:
@@ -754,7 +766,6 @@ def main():
     except KeyboardInterrupt:
         graceful_shutdown()
     finally:
-        # 二次保险，确保全部退出
         graceful_shutdown()
         for w in workers:
             w.join(timeout=10)