diff --git a/back/mult2 b/back/mult2 new file mode 100644 index 0000000..e526da3 --- /dev/null +++ b/back/mult2 @@ -0,0 +1,768 @@ +# -*- 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.1-logonly" +os.environ['PATH'] = ffmpeg_path + os.pathsep + current_env_path +script_path = os.path.split(os.path.realpath(sys.argv[0]))[0] +config_file = f'{script_path}/config/config.ini' +url_config_file = f'{script_path}/config/URL_config.ini' +yt_ini_file = f"{script_path}/config/youtube.ini" +text_encoding = 'utf-8-sig' +color = Color() + +# ---- 新增：全局 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", + + "-user_agent", user_agent, + "-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy", + + "-thread_queue_size", "4096", + "-analyzeduration", "10000000", + "-probesize", "10000000", + ] + if extra_headers: + base += ["-headers", extra_headers] + + env = os.environ.copy() + if proxy_addr: + env["http_proxy"] = proxy_addr + env["https_proxy"] = proxy_addr + base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:] + + base += ["-re", "-i", input_url] + + base += [ + "-fflags", "+genpts+igndts+discardcorrupt+nobuffer", + "-xerror", + "-use_wallclock_as_timestamps", "1", + "-avioflags", "direct", + "-rtbufsize", "100M", + ] + + base += v_args + a_args + base += ["-f", "flv", out_url] + + cmd_str = " ".join(base) + print(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 与 (url,key) 占用） =============== +class UrlCoordinator: + """ + - urls 顺序分配 + - in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用） + - in_use_pairs: (url,key) 级别独占（禁止同一 url 推到同一个 key） + """ + def __init__(self, path: str): + self.path = path + self.urls: List[str] = parse_urls_from_file(path) + self.idx: int = 0 + self.lock = threading.Lock() + self.last_md5 = check_md5(path) + + self.in_use_urls: Set[str] = set() + self.in_use_pairs: Set[Tuple[str, str]] = set() # (url, key) + + def refresh_if_changed(self): + try: + md5 = check_md5(self.path) + if md5 != self.last_md5: + self.urls = parse_urls_from_file(self.path) or self.urls + self.last_md5 = md5 + print(f"[{now_str()}] 已重载 URL 列表，数量：{len(self.urls)}") + except Exception as e: + logger.warning(f"URL 列表热更新失败：{e}") + + def acquire_next_live(self, worker_key: str, quality_code: str = "OD") -> Optional[Tuple[str, dict]]: + with self.lock: + if not self.urls: + return None + tried = 0 + pick = None + while tried < len(self.urls): + url = self.urls[self.idx] + self.idx = (self.idx + 1) % len(self.urls) + tried += 1 + # 已被其他路占用 / 同 url-key 已经在用 + if url in self.in_use_urls: + continue + if (url, worker_key) in self.in_use_pairs: + continue + pick = url + break + if not pick: + return None + + # 锁外解析 + port = asyncio.run(fetch_port_info(pick, quality_code)) + if not port.get("anchor_name"): + LOGSTATE.set_status(pick, "unknown", + text=f"解析失败：{pick}", + prefix="") + # ⚠️ 增加：释放 URL 占用，避免卡死 + self.release(pick, worker_key) + return None + if not port.get("is_live"): + anchor = port.get('anchor_name','') + LOGSTATE.set_status(pick, "offline", + text=f"未开播：{anchor} | {pick}", + prefix="") + # ⚠️ 增加：释放 URL 占用，避免卡死 + self.release(pick, worker_key) + return None + + # 占用 + with self.lock: + if (pick in self.in_use_urls) or ((pick, worker_key) in self.in_use_pairs): + return None + self.in_use_urls.add(pick) + self.in_use_pairs.add((pick, worker_key)) + + # 状态更新（online） + anchor = port.get('anchor_name','') + LOGSTATE.set_status(pick, "online", + text=f"已开播：{anchor} | {pick}", + prefix="") + return pick, port + + def release(self, url: Optional[str], worker_key: Optional[str]): + if not url: + return + with self.lock: + self.in_use_urls.discard(url) + if worker_key: + self.in_use_pairs.discard((url, worker_key)) + +# =============== Worker（保留前缀日志） =============== +class StreamWorker(threading.Thread): + def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator): + super().__init__(daemon=True) + self.worker_id = worker_id + self.key = key + self.worker_name = f"Worker-{worker_id}" + self.resolution = resolution + self.coordinator = coordinator + + self.current_url: Optional[str] = None + self.current_real_url: Optional[str] = None + self.current_proc: Optional[subprocess.Popen] = None + self.current_platform = "" + self.current_anchor = "" + self.record_quality_code = "OD" + + # 与原逻辑一致的参数 + self.watchdog_grace = 120 + self.watchdog_silence = 600 + self.refresh_real_url_every = 300 + self.check_interval_live = 8 + self.scan_interval_when_idle = 2 + + self._last_progress_print = -9999 + self._stop_event = threading.Event() + + def stop(self): + """最小改动：收到停止请求时，立即尝试停掉当前 ffmpeg，尽快响应退出。""" + self._stop_event.set() + try: + stop_ffmpeg(self.current_proc) + except Exception: + pass + + def _p(self, text: str): + print(f"[{now_str()}][{self.worker_name}] {text}") + + def run(self): + self._p("启动") + try: + while not self._stop_event.is_set(): + self.coordinator.refresh_if_changed() + + if self.current_proc and self.current_url: + retcode = self.current_proc.poll() + if retcode is not None: + self._p(f"❌ ffmpeg 退出(code={retcode})，切换下一个源…") + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url, self.key) + self._clear_state() + LOGSTATE.clear_idle_flag(self.worker_name) + continue + + # 检测是否仍在直播（保持原逻辑） + still_live = True + try: + port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + still_live = bool(port.get("is_live")) + except Exception as e: + logger.error(f"[{self.worker_name}] 检测当前源失败：{e}") + still_live = True + + if still_live: + elapsed = int(time.time() - self.start_time) + # 每 60s 打印一次“正在转播” + if elapsed - self._last_progress_print >= 60: + h, m = divmod(elapsed, 3600) + m, s = divmod(m, 60) + dur_str = f"{h:02d}:{m:02d}:{s:02d}" + self._p(f"⏱ 正在转播({dur_str})：{self.current_platform} | {self.current_anchor}") + self._last_progress_print = elapsed + + # 定期刷新真实播放地址（token） + try: + if elapsed >= self.refresh_real_url_every and (elapsed % self.refresh_real_url_every) < self.check_interval_live: + new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + new_real_url = select_source_url(self.current_url, new_port) + if new_real_url and new_real_url != self.current_real_url: + self._p("🔄 刷新播放地址（token 变化），重启 ffmpeg…") + stop_ffmpeg(self.current_proc) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + headers = headers_for_platform(self.current_platform, self.current_url) + self.current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args, a_args, self.worker_name) + self.current_real_url = new_real_url + self.start_time = time.time() + self._last_progress_print = -9999 + time.sleep(self.check_interval_live) + continue + except Exception as e: + logger.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}") + + # watchdog + silence = time.time() - getattr(self.current_proc, "last_out_ts", 0) + if elapsed >= self.watchdog_grace and silence > self.watchdog_silence: + self._p("⚠️ 超过10分钟未见输出，重启…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + time.sleep(1) + continue + + time.sleep(self.check_interval_live) + continue + else: + self._p("📴 主播停播，切换下一个源…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + continue + + # —— 未在推流：尝试分配一个新源（保持原顺序轮询 + 占用） —— + got = self.coordinator.acquire_next_live(self.key, self.record_quality_code) + if not got: + # 仅首次空闲时提示一次 + LOGSTATE.mark_idle_once(self.worker_name) + time.sleep(self.scan_interval_when_idle) + continue + + LOGSTATE.clear_idle_flag(self.worker_name) + url, port = got + real_url = select_source_url(url, port) + if not real_url: + self._p(f"未得到播放地址，跳过：{url}") + self.coordinator.release(url, self.key) + time.sleep(1) + continue + + self.current_platform = detect_platform(url) + self.current_anchor = port.get("anchor_name", "") + headers = headers_for_platform(self.current_platform, url) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args, self.worker_name) + self.current_url = url + self.current_real_url = real_url + self.start_time = time.time() + self._last_progress_print = -9999 + + self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}") + + except Exception as e: + logger.error(f"[{self.worker_name}] 主循环异常：{e}") + finally: + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url, self.key) + self._p("已退出。") + + def _clear_state(self, release: bool = False): + if release and self.current_url: + self.coordinator.release(self.current_url, self.key) + self.current_url = None + self.current_real_url = None + self.current_proc = None + self.current_platform = "" + self.current_anchor = "" + +# =============== 启动入口（保持原行为） =============== +def check_ffmpeg_existence() -> bool: + try: + result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True) + if result.returncode == 0: + print(result.stdout.splitlines()[0]) + except Exception: + pass + finally: + return bool(check_ffmpeg()) + +# ---- 新增：统一优雅退出 ---- +def graceful_shutdown(signum=None, frame=None): + if not SHUTDOWN_EVENT.is_set(): + print("\n收到中断，正在停止所有 worker 与 ffmpeg …") + SHUTDOWN_EVENT.set() + # 先发 stop 给所有 worker（会内部 stop 当前 ffmpeg） + for w in list(ALL_WORKERS): + try: + w.stop() + except Exception: + pass + # 再兜底停止全局登记的 ffmpeg 子进程 + with ALL_LOCK: + procs = list(ALL_PROCS) + for p in procs: + try: + stop_ffmpeg(p) + except Exception: + pass + +def main(): + print("-----------------------------------------------------") + print("| DouyinLiveRecorder - 并行转推版 |") + print("-----------------------------------------------------") + print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") + print("模式：并行检测 + 并行推流（保持原有业务逻辑）") + print(".....................................................") + + if not check_ffmpeg_existence(): + logger.error("缺少 ffmpeg，程序退出") + sys.exit(1) + + urls = parse_urls_from_file(url_config_file) + if not urls: + logger.error("URL_config.ini 中没有可用 URL。") + sys.exit(1) + + # keys（并行路数 = keys 数量）；兼容单 key + keys = _yt_get_list("youtube", "keys") + if not keys: + single_key = _yt_get("youtube", "key", fallback="", cast=str) + if single_key: + keys = [single_key] + if not keys: + logger.error("youtube.ini 没有配置 [youtube].keys 或 key") + sys.exit(1) + + # 可选：每路分辨率 + per_res = _yt_get_list("youtube", "resolutions") # 如：1920x1080,1080x1920 + while per_res and len(per_res) < len(keys): + per_res.append(per_res[-1]) + + # 打印 ffmpeg 版本（贴合旧风格） + try: + vline = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True).stdout.splitlines()[0] + print(vline) + except Exception: + pass + + coordinator = UrlCoordinator(url_config_file) + + # 信号处理（Ctrl+C / kill） + try: + signal.signal(signal.SIGINT, graceful_shutdown) + except Exception: + pass + try: + signal.signal(signal.SIGTERM, graceful_shutdown) + except Exception: + pass + + # 启动多路（Worker 前缀区分） + workers: List[StreamWorker] = [] + for i, key in enumerate(keys, start=1): + res_for_this = per_res[i-1] if per_res else None + w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator) + w.start() + workers.append(w) + # ---- 新增：登记全局，供优雅退出用 ---- + ALL_WORKERS.append(w) + + try: + while not SHUTDOWN_EVENT.is_set(): + time.sleep(0.5) + except KeyboardInterrupt: + graceful_shutdown() + finally: + # 二次保险，确保全部退出 + graceful_shutdown() + for w in workers: + w.join(timeout=10) + print("全部 worker 已退出。") + +if __name__ == "__main__": + main() diff --git a/back/mult3 b/back/mult3 new file mode 100644 index 0000000..f2068f0 --- /dev/null +++ b/back/mult3 @@ -0,0 +1,782 @@ +# -*- 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 与 (url,key) 占用） =============== +class UrlCoordinator: + """ + - urls 顺序分配 + - in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用） + - in_use_pairs: (url,key) 级别独占（禁止同一 url 推到同一个 key） + """ + def __init__(self, path: str): + self.path = path + self.urls: List[str] = parse_urls_from_file(path) + self.idx: int = 0 + self.lock = threading.Lock() + self.last_md5 = check_md5(path) + + self.in_use_urls: Set[str] = set() + self.in_use_pairs: Set[Tuple[str, str]] = set() # (url, key) + + def refresh_if_changed(self): + try: + md5 = check_md5(self.path) + if md5 != self.last_md5: + self.urls = parse_urls_from_file(self.path) or self.urls + self.last_md5 = md5 + print(f"[{now_str()}] 已重载 URL 列表，数量：{len(self.urls)}") + except Exception as e: + logger.warning(f"URL 列表热更新失败：{e}") + + def acquire_next_live(self, worker_key: str, quality_code: str = "OD") -> Optional[Tuple[str, dict]]: + """尝试遍历一轮 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 or (url, worker_key) in self.in_use_pairs: + 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 or (url, worker_key) in self.in_use_pairs: + tried += 1 + continue + self.in_use_urls.add(url) + self.in_use_pairs.add((url, worker_key)) + + # 状态更新（online） + 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) + if worker_key: + self.in_use_pairs.discard((url, worker_key)) + +# =============== Worker（保留前缀日志） =============== +class StreamWorker(threading.Thread): + def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator): + super().__init__(daemon=True) + self.worker_id = worker_id + self.key = key + self.worker_name = f"Worker-{worker_id}" + self.resolution = resolution + self.coordinator = coordinator + + self.current_url: Optional[str] = None + self.current_real_url: Optional[str] = None + self.current_proc: Optional[subprocess.Popen] = None + self.current_platform = "" + self.current_anchor = "" + self.record_quality_code = "OD" + + # 与原逻辑一致的参数 + self.watchdog_grace = 120 + self.watchdog_silence = 600 + self.refresh_real_url_every = 300 + self.check_interval_live = 8 + self.scan_interval_when_idle = 2 + + self._last_progress_print = -9999 + self._stop_event = threading.Event() + self.start_time = time.time() + + def stop(self): + """最小改动：收到停止请求时，立即尝试停掉当前 ffmpeg，尽快响应退出。""" + self._stop_event.set() + try: + stop_ffmpeg(self.current_proc) + except Exception: + pass + + def _p(self, text: str): + print(f"[{now_str()}][{self.worker_name}] {text}") + + def run(self): + self._p("启动") + try: + while not self._stop_event.is_set(): + try: + self.coordinator.refresh_if_changed() + + if self.current_proc and self.current_url: + retcode = self.current_proc.poll() + if retcode is not None: + self._p(f"❌ ffmpeg 退出(code={retcode})，切换下一个源…") + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url, self.key) + self._clear_state() + LOGSTATE.clear_idle_flag(self.worker_name) + continue + + # 检测是否仍在直播（保持原逻辑） + still_live = True + try: + port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + still_live = bool(port.get("is_live")) + except Exception as e: + logger.error(f"[{self.worker_name}] 检测当前源失败：{e}") + still_live = True + + if still_live: + elapsed = int(time.time() - self.start_time) + # 每 60s 打印一次“正在转播” + if elapsed - self._last_progress_print >= 60: + h, m = divmod(elapsed, 3600) + m, s = divmod(m, 60) + dur_str = f"{h:02d}:{m:02d}:{s:02d}" + self._p(f"⏱ 正在转播({dur_str})：{self.current_platform} | {self.current_anchor}") + self._last_progress_print = elapsed + + # 定期刷新真实播放地址（token） + try: + if elapsed >= self.refresh_real_url_every and (elapsed % self.refresh_real_url_every) < self.check_interval_live: + new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) + new_real_url = select_source_url(self.current_url, new_port) + if new_real_url and new_real_url != self.current_real_url: + self._p("🔄 刷新播放地址（token 变化），重启 ffmpeg…") + stop_ffmpeg(self.current_proc) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + headers = headers_for_platform(self.current_platform, self.current_url) + self.current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args, a_args, self.worker_name) + self.current_real_url = new_real_url + self.start_time = time.time() + self._last_progress_print = -9999 + time.sleep(self.check_interval_live) + continue + except Exception as e: + logger.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}") + + # watchdog + silence = time.time() - getattr(self.current_proc, "last_out_ts", 0) + if elapsed >= self.watchdog_grace and silence > self.watchdog_silence: + self._p("⚠️ 超过10分钟未见输出，重启…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + time.sleep(1) + continue + + time.sleep(self.check_interval_live) + continue + else: + self._p("📴 主播停播，切换下一个源…") + stop_ffmpeg(self.current_proc) + self._clear_state(release=True) + LOGSTATE.clear_idle_flag(self.worker_name) + continue + + # —— 未在推流：尝试分配一个新源（保持原顺序轮询 + 占用） —— + got = self.coordinator.acquire_next_live(self.key, self.record_quality_code) + if not got: + LOGSTATE.mark_idle_once(self.worker_name) + time.sleep(self.scan_interval_when_idle) + continue + + LOGSTATE.clear_idle_flag(self.worker_name) + url, port = got + real_url = select_source_url(url, port) + if not real_url: + self._p(f"未得到播放地址，跳过：{url}") + self.coordinator.release(url, self.key) + time.sleep(1) + continue + + self.current_platform = detect_platform(url) + self.current_anchor = port.get("anchor_name", "") + headers = headers_for_platform(self.current_platform, url) + out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) + self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args, self.worker_name) + self.current_url = url + self.current_real_url = real_url + self.start_time = time.time() + self._last_progress_print = -9999 + + self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}") + + except Exception as e: + # 🔧 关键：任何内部异常不让线程退出 + logger.error(f"[{self.worker_name}] 内部异常：{e}") + time.sleep(2) + continue + + finally: + stop_ffmpeg(self.current_proc) + self.coordinator.release(self.current_url, self.key) + self._p("已退出。") + + def _clear_state(self, release: bool = False): + if release and self.current_url: + self.coordinator.release(self.current_url, self.key) + self.current_url = None + self.current_real_url = None + self.current_proc = None + self.current_platform = "" + self.current_anchor = "" + +# =============== 启动入口（保持原行为） =============== +def check_ffmpeg_existence() -> bool: + try: + result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True) + if result.returncode == 0: + print(result.stdout.splitlines()[0]) + except Exception: + pass + finally: + return bool(check_ffmpeg()) + +# ---- 新增：统一优雅退出 ---- +def graceful_shutdown(signum=None, frame=None): + if not SHUTDOWN_EVENT.is_set(): + print("\n收到中断，正在停止所有 worker 与 ffmpeg …") + SHUTDOWN_EVENT.set() + # 先发 stop 给所有 worker（会内部 stop 当前 ffmpeg） + for w in list(ALL_WORKERS): + try: + w.stop() + except Exception: + pass + # 再兜底停止全局登记的 ffmpeg 子进程 + with ALL_LOCK: + procs = list(ALL_PROCS) + for p in procs: + try: + stop_ffmpeg(p) + except Exception: + pass + +def main(): + print("-----------------------------------------------------") + print("| DouyinLiveRecorder - 并行转推版 |") + print("-----------------------------------------------------") + print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") + print("模式：并行检测 + 并行推流（保持原有业务逻辑）") + print(".....................................................") + + if not check_ffmpeg_existence(): + logger.error("缺少 ffmpeg，程序退出") + sys.exit(1) + + urls = parse_urls_from_file(url_config_file) + if not urls: + logger.error("URL_config.ini 中没有可用 URL。") + sys.exit(1) + + # keys（并行路数 = keys 数量）；兼容单 key + keys = _yt_get_list("youtube", "keys") + if not keys: + single_key = _yt_get("youtube", "key", fallback="", cast=str) + if single_key: + keys = [single_key] + if not keys: + logger.error("youtube.ini 没有配置 [youtube].keys 或 key") + sys.exit(1) + + # 可选：每路分辨率 + per_res = _yt_get_list("youtube", "resolutions") # 如：1920x1080,1080x1920 + while per_res and len(per_res) < len(keys): + per_res.append(per_res[-1]) + + # 打印 ffmpeg 版本（贴合旧风格） + try: + vline = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True).stdout.splitlines()[0] + print(vline) + except Exception: + pass + + coordinator = UrlCoordinator(url_config_file) + + # 信号处理（Ctrl+C / kill） + try: + signal.signal(signal.SIGINT, graceful_shutdown) + except Exception: + pass + try: + signal.signal(signal.SIGTERM, graceful_shutdown) + except Exception: + pass + + # 启动多路（Worker 前缀区分） + workers: List[StreamWorker] = [] + for i, key in enumerate(keys, start=1): + res_for_this = per_res[i-1] if per_res else None + w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator) + w.start() + workers.append(w) + ALL_WORKERS.append(w) + + try: + while not SHUTDOWN_EVENT.is_set(): + time.sleep(0.5) + except KeyboardInterrupt: + graceful_shutdown() + finally: + graceful_shutdown() + for w in workers: + w.join(timeout=10) + print("全部 worker 已退出。") + +if __name__ == "__main__": + main() diff --git a/config/URL_config.ini b/config/URL_config.ini index 1fda72a..7aaad34 100644 --- a/config/URL_config.ini +++ b/config/URL_config.ini @@ -1,13 +1,13 @@ -https://live.douyin.com/72212955083 +https://live.douyin.com/50627104055 +https://live.douyin.com/339454333072 +https://live.douyin.com/72212955083 https://live.douyin.com/621118100231 -https://live.douyin.com/347310028364 https://live.douyin.com/280422307519 https://live.douyin.com/651293616593 https://live.douyin.com/114111936890 https://live.douyin.com/43665279648 https://live.douyin.com/122383435921 https://live.douyin.com/92876979328 -https://live.douyin.com/136996589868 https://live.douyin.com/517715534931 https://live.douyin.com/460525712926 https://live.douyin.com/456576000931 @@ -20,9 +20,7 @@ https://live.douyin.com/152358755212 https://live.douyin.com/990825651731 https://live.douyin.com/591442402624 https://live.douyin.com/967381081829 -https://live.douyin.com/15652099787 https://live.douyin.com/78012762575 -https://live.douyin.com/80248655914 https://live.douyin.com/163813589919 https://live.douyin.com/689189188452 https://live.douyin.com/498368994814 @@ -32,7 +30,6 @@ https://live.douyin.com/993102287144 https://live.douyin.com/701547125568 https://live.douyin.com/816130992220 https://live.douyin.com/876468215361 -https://live.douyin.com/291403201249 https://live.douyin.com/511335278313 https://live.douyin.com/483160615952 https://live.douyin.com/673565298571 diff --git a/main.py b/main.py index 749fd4c..fc1df56 100644 --- a/main.py +++ b/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)