From 8395418a1262c757c73c23a8ef9c3d4d46a1d937 Mon Sep 17 00:00:00 2001 From: eric Date: Tue, 16 Sep 2025 21:29:47 +0800 Subject: [PATCH] 's --- backup/back9 | 44 ++++---- live.py | 294 +++++++++++++++++++++++++++++++-------------------- 2 files changed, 206 insertions(+), 132 deletions(-) diff --git a/backup/back9 b/backup/back9 index 41de1a2..c13ec99 100644 --- a/backup/back9 +++ b/backup/back9 @@ -8,9 +8,10 @@ import signal import sys import logging import platform -import spider -import stream import aiohttp +import select +import spider # 你原来的 spider 模块 +import stream # 你原来的 stream 模块 # ----------------- Logging Setup ----------------- logging.basicConfig( @@ -38,7 +39,7 @@ class DouyinToYouTubeStreamer: self.pushing = False self.out_file = None self.max_minutes = max_minutes - self.enable_recording = enable_recording # 控制录制开关 + self.enable_recording = enable_recording os.makedirs("records", exist_ok=True) @@ -73,6 +74,7 @@ class DouyinToYouTubeStreamer: '-bufsize', f'{self.bitrate * 2}k', '-r', str(self.fps), '-g', str(self.gop), + '-s', self.resolution, '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-f', 'flv', f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}' @@ -98,7 +100,7 @@ class DouyinToYouTubeStreamer: ] return cmd - # ----------------- 推流循环 ----------------- + # ----------------- 推流循环（优化版） ----------------- def _run_stream_loop(self, real_url, youtube_key): while not self.stop_flag.is_set(): try: @@ -107,26 +109,31 @@ class DouyinToYouTubeStreamer: # ---- 启动推流 ---- push_cmd = self._build_push_cmd(real_url, youtube_key) - self.push_proc = subprocess.Popen(push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + self.push_proc = subprocess.Popen( + push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + bufsize=1, universal_newlines=True + ) logging.info(f"[{self.room_name}] 推流进程 PID: {self.push_proc.pid}") - # ---- 启动本地录制（根据开关决定） ---- + # ---- 启动本地录制 ---- if self.enable_recording: record_cmd = self._build_record_cmd(real_url) - self.record_proc = subprocess.Popen(record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + self.record_proc = subprocess.Popen( + record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + bufsize=1, universal_newlines=True + ) logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") - # ---- monitor stderr ---- + # ---- monitor stderr 非阻塞 ---- def monitor_proc(proc, name): try: while not self.stop_flag.is_set() and proc and proc.poll() is None: - data = proc.stderr.read(1024) - if data and name == "push" and not self.pushing: - decoded = data.decode('utf-8', errors='ignore') - if "frame=" in decoded and "fps=" in decoded: - self.pushing = True - logging.info(f"[{self.room_name}] 推流到 YouTube 成功") - time.sleep(0.05) + line = proc.stderr.readline() + if not line: + continue + if name == "push" and not self.pushing and "frame=" in line and "fps=" in line: + self.pushing = True + logging.info(f"[{self.room_name}] 推流到 YouTube 成功") except Exception as e: logging.error(f"[{self.room_name} {name} stderr异常: {e}") @@ -134,13 +141,14 @@ class DouyinToYouTubeStreamer: if self.enable_recording: threading.Thread(target=monitor_proc, args=(self.record_proc, "record"), daemon=True).start() - # ---- 循环检查 ---- + # ---- 循环检查进程状态 ---- while not self.stop_flag.is_set(): elapsed = time.time() - self.start_time if elapsed >= self.max_minutes * 60: logging.warning(f"[{self.room_name}] 超过 {self.max_minutes} 分钟，自动停止推流和录制") self.stop() break + if self.push_proc.poll() is not None: logging.warning(f"[{self.room_name}] 推流进程退出，5秒后重启...") time.sleep(5) @@ -149,7 +157,6 @@ class DouyinToYouTubeStreamer: logging.warning(f"[{self.room_name}] 录制进程退出，5秒后重启...") time.sleep(5) break - time.sleep(2) except Exception as e: logging.error(f"[{self.room_name}] FFmpeg异常: {e}") @@ -231,6 +238,7 @@ async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=No logging.error(f"初始化 aiohttp session 失败: {e}") return None + def start_douyin_to_youtube(record_url, youtube_key, room_name=None, proxy_addr=None, cookies='', bitrate=5000, enable_recording=False): real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) @@ -245,7 +253,7 @@ def start_douyin_to_youtube(record_url, youtube_key, room_name=None, # ----------------- Main Program ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/851742623054" + DOUYIN_URL = "https://live.douyin.com/722725928418" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" # enable_recording=True 可开启本地录制 diff --git a/live.py b/live.py index 39970ae..c13ec99 100644 --- a/live.py +++ b/live.py @@ -6,10 +6,12 @@ import uuid import time import signal import sys -import shutil import logging -import spider -import stream +import platform +import aiohttp +import select +import spider # 你原来的 spider 模块 +import stream # 你原来的 stream 模块 # ----------------- Logging Setup ----------------- logging.basicConfig( @@ -18,10 +20,14 @@ logging.basicConfig( datefmt="%H:%M:%S" ) -# ----------------- Low-Latency Douyin to YouTube Streamer ----------------- +IS_WINDOWS = platform.system() == "Windows" + +# ----------------- Douyin to YouTube Streamer ----------------- class DouyinToYouTubeStreamer: - def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', fps=30, gop=60): - self.proc = None + def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', + fps=30, gop=60, max_minutes=8, enable_recording=False): + self.push_proc = None + self.record_proc = None self.stop_flag = threading.Event() self.bitrate = bitrate self.audio_bitrate = audio_bitrate @@ -30,175 +36,235 @@ class DouyinToYouTubeStreamer: self.gop = gop self.start_time = None self.room_name = None - self.semaphore = threading.Semaphore(1) - self.pushing = False # 是否正在推流 + self.pushing = False + self.out_file = None + self.max_minutes = max_minutes + self.enable_recording = enable_recording + os.makedirs("records", exist_ok=True) + + # ----------------- 启动推流 ----------------- def start_stream(self, real_url, youtube_key, room_name=None): self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" logging.info(f"[{self.room_name}] 启动低延迟推流，目标 YouTube Key: {youtube_key}") - threading.Thread(target=self._run_ffmpeg_loop, args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._run_stream_loop, + args=(real_url, youtube_key), daemon=True).start() threading.Thread(target=self._monitor_thread, daemon=True).start() - def _build_ffmpeg_cmd(self, real_url, youtube_key): - user_agent = ("Mozilla/5.0 (Linux; Android 11; SAMSUNG SM-G973U) AppleWebKit/537.36 " - "(KHTML, like Gecko) SamsungBrowser/14.2 Chrome/87.0.4280.141 Mobile Safari/537.36") + # ----------------- 构建推流命令 ----------------- + def _build_push_cmd(self, real_url, youtube_key): + user_agent = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36") headers = "Referer: https://www.douyin.com/\r\nOrigin: https://www.douyin.com\r\n" - cmd = [ - 'ffmpeg', '-y', '-v', 'error', '-hide_banner', + 'ffmpeg', '-y', '-rw_timeout', '10000000', '-analyzeduration', '1000000', '-probesize', '5000000', - '-thread_queue_size', '512', '-user_agent', user_agent, '-headers', headers, '-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto', '-fflags', '+discardcorrupt+genpts', '-flags', 'low_delay', '-http_seekable', '0', - '-avioflags', 'direct', - '-reconnect', '1', - '-reconnect_at_eof', '1', - '-reconnect_streamed', '1', - '-reconnect_delay_max', '5', - '-flush_packets', '1', '-i', real_url, - '-c:v', 'libx264', - '-preset', 'veryfast', - '-tune', 'zerolatency', - '-crf', '23', - '-pix_fmt', 'yuv420p', - '-profile:v', 'main', - '-level', '3.1', - '-x264-params', f'keyint={self.gop}:scenecut=0:min-keyint={self.gop//2}', - '-g', str(self.gop), - '-r', str(self.fps), - '-s', self.resolution, + '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', '-b:v', f'{self.bitrate}k', '-maxrate', f'{self.bitrate}k', '-bufsize', f'{self.bitrate * 2}k', - '-c:a', 'aac', - '-b:a', f'{self.audio_bitrate}k', - '-ar', '44100', - '-ac', '2', - '-strict', '-2', - '-avoid_negative_ts', 'make_zero', + '-r', str(self.fps), + '-g', str(self.gop), + '-s', self.resolution, + '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-f', 'flv', f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}' ] return cmd - def _run_ffmpeg_loop(self, real_url, youtube_key): + # ----------------- 构建录制命令 ----------------- + def _build_record_cmd(self, real_url): + base_name = f"{self.room_name}_{int(time.time())}.mp4" + self.out_file = os.path.join("records", base_name).replace("\\", "/") + cmd = [ + 'ffmpeg', '-y', + '-i', real_url, + '-c:v', 'libx264', + '-preset', 'veryfast', + '-tune', 'zerolatency', + '-c:a', 'aac', + '-b:a', f'{self.audio_bitrate}k', + '-ar', '44100', + '-ac', '2', + '-movflags', '+faststart+frag_keyframe', + self.out_file + ] + return cmd + + # ----------------- 推流循环（优化版） ----------------- + def _run_stream_loop(self, real_url, youtube_key): while not self.stop_flag.is_set(): try: - with self.semaphore: - cmd = self._build_ffmpeg_cmd(real_url, youtube_key) - self.proc = subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, - bufsize=0, universal_newlines=False + self.start_time = time.time() + self.pushing = False + + # ---- 启动推流 ---- + push_cmd = self._build_push_cmd(real_url, youtube_key) + self.push_proc = subprocess.Popen( + push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + bufsize=1, universal_newlines=True + ) + logging.info(f"[{self.room_name}] 推流进程 PID: {self.push_proc.pid}") + + # ---- 启动本地录制 ---- + if self.enable_recording: + record_cmd = self._build_record_cmd(real_url) + self.record_proc = subprocess.Popen( + record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + bufsize=1, universal_newlines=True ) - self.start_time = time.time() - self.pushing = False - logging.info(f"[{self.room_name}] FFmpeg 启动成功 (PID: {self.proc.pid})") + logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") - def read_stderr(): - try: - for line in iter(self.proc.stderr.readline, b''): - if self.stop_flag.is_set(): - break - if line: - decoded = line.decode('utf-8', errors='ignore').strip() - logging.info(f"[{self.room_name} FFmpeg]: {decoded}") - # 判断推流是否成功 - if not self.pushing and "frame=" in decoded and "fps=" in decoded: - self.pushing = True - logging.info(f"[{self.room_name}] 推流到 YouTube 成功") - except Exception as e: - logging.error(f"[{self.room_name}] stderr线程异常: {e}") - - stderr_thread = threading.Thread(target=read_stderr, daemon=True) - stderr_thread.start() - - self.proc.wait() - if not self.stop_flag.is_set(): - logging.warning(f"[{self.room_name}] FFmpeg 退出 (返回码: {self.proc.returncode})，1秒后重启...") - self.pushing = False - except Exception as e: - logging.error(f"[{self.room_name}] FFmpeg 启动异常: {e}") - finally: - if self.proc and self.proc.poll() is None: - self.proc.terminate() + # ---- monitor stderr 非阻塞 ---- + def monitor_proc(proc, name): try: - self.proc.wait(timeout=1) - except subprocess.TimeoutExpired: - self.proc.kill() - self.proc = None + while not self.stop_flag.is_set() and proc and proc.poll() is None: + line = proc.stderr.readline() + if not line: + continue + if name == "push" and not self.pushing and "frame=" in line and "fps=" in line: + self.pushing = True + logging.info(f"[{self.room_name}] 推流到 YouTube 成功") + except Exception as e: + logging.error(f"[{self.room_name} {name} stderr异常: {e}") + + threading.Thread(target=monitor_proc, args=(self.push_proc, "push"), daemon=True).start() + if self.enable_recording: + threading.Thread(target=monitor_proc, args=(self.record_proc, "record"), daemon=True).start() + + # ---- 循环检查进程状态 ---- + while not self.stop_flag.is_set(): + elapsed = time.time() - self.start_time + if elapsed >= self.max_minutes * 60: + logging.warning(f"[{self.room_name}] 超过 {self.max_minutes} 分钟，自动停止推流和录制") + self.stop() + break + + if self.push_proc.poll() is not None: + logging.warning(f"[{self.room_name}] 推流进程退出，5秒后重启...") + time.sleep(5) + break + if self.enable_recording and self.record_proc and self.record_proc.poll() is not None: + logging.warning(f"[{self.room_name}] 录制进程退出，5秒后重启...") + time.sleep(5) + break + + except Exception as e: + logging.error(f"[{self.room_name}] FFmpeg异常: {e}") + finally: + self._terminate_procs() if not self.stop_flag.is_set(): time.sleep(1) - def _monitor_thread(self): - term_width, _ = shutil.get_terminal_size() - bar_width = min(50, term_width - 30) - while not self.stop_flag.is_set(): - if self.proc: - uptime = time.time() - self.start_time if self.start_time else 0 - uptime_str = f"{int(uptime // 60):02d}:{int(uptime % 60):02d}" - status = "推流中" if self.pushing else "启动中" - bar_len = bar_width if self.pushing else 0 - bar = "#" * bar_len + "-" * (bar_width - bar_len) - print(f"\r[{self.room_name}] [{bar}] {status} | 运行时间: {uptime_str} | 码率: {self.bitrate}k", end="") - else: - print(f"\r[{self.room_name}] FFmpeg 未运行", end="") - time.sleep(2) + def _terminate_procs(self): + for proc in [self.push_proc, self.record_proc]: + if proc and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + proc.kill() + self.push_proc, self.record_proc = None, None + # ----------------- 监控线程 ----------------- + def _monitor_thread(self): + while not self.stop_flag.is_set(): + uptime = int(time.time() - self.start_time) if self.start_time else 0 + m, s = divmod(uptime, 60) + status = "推流中" if self.pushing else "启动中" + print(f"\r[{self.room_name}] {status} | 运行时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k", end="") + time.sleep(5) + + # ----------------- 停止 ----------------- def stop(self): - logging.info(f"[{self.room_name}] 停止推流...") + logging.info(f"[{self.room_name}] 停止推流和录制...") self.stop_flag.set() - if self.proc and self.proc.poll() is None: - self.proc.terminate() - try: - self.proc.wait(timeout=1) - except subprocess.TimeoutExpired: - self.proc.kill() + self._terminate_procs() # ----------------- Douyin Stream Parsing ----------------- async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=None, cookies=''): + headers = { + "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"), + "Referer": "https://www.douyin.com/", + "Origin": "https://www.douyin.com", + "Accept": "application/json, text/javascript, */*; q=0.01", + } + + session_args = {"headers": headers} + if proxy_addr: + session_args["proxy"] = proxy_addr + try: - if 'v.douyin.com' not in record_url and '/user/' not in record_url: - json_data = await spider.get_douyin_stream_data(url=record_url, proxy_addr=proxy_addr, cookies=cookies) - else: - json_data = await spider.get_douyin_app_stream_data(url=record_url, proxy_addr=proxy_addr, cookies=cookies) - port_info = await stream.get_douyin_stream_url(json_data, record_quality, proxy_addr) - return port_info.get("record_url") + async with aiohttp.ClientSession(**session_args) as session: + for attempt in range(3): + try: + async with session.get(record_url, cookies=cookies) as resp: + text = await resp.text() + if not text or "captcha" in text.lower(): + logging.warning(f"第{attempt+1}次尝试获取流失败，返回内容为空或验证码，重试...") + await asyncio.sleep(1) + continue + if 'v.douyin.com' not in record_url and '/user/' not in record_url: + json_data = await spider.get_douyin_stream_data( + url=record_url, proxy_addr=proxy_addr, cookies=cookies) + else: + json_data = await spider.get_douyin_app_stream_data( + url=record_url, proxy_addr=proxy_addr, cookies=cookies) + + port_info = await stream.get_douyin_stream_url( + json_data, record_quality, proxy_addr) + if port_info and port_info.get("record_url"): + return port_info.get("record_url") + else: + logging.warning(f"第{attempt+1}次尝试解析失败，port_info为空") + await asyncio.sleep(1) + except Exception as e: + logging.warning(f"第{attempt+1}次尝试抓取抖音流异常: {e}") + await asyncio.sleep(1) + logging.error("多次尝试后仍获取抖音真实流失败") + return None except Exception as e: - logging.error(f"获取抖音真实流失败: {e}") + logging.error(f"初始化 aiohttp session 失败: {e}") return None -def start_douyin_to_youtube(record_url, youtube_key, room_name=None, proxy_addr=None, cookies='', bitrate=5000): - real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr, cookies)) + +def start_douyin_to_youtube(record_url, youtube_key, room_name=None, + proxy_addr=None, cookies='', bitrate=5000, enable_recording=False): + real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) if not real_url: logging.error("获取真实直播流失败") return None logging.info(f"实际直播流: {real_url}") - streamer = DouyinToYouTubeStreamer(bitrate=bitrate) + streamer = DouyinToYouTubeStreamer(bitrate=bitrate, enable_recording=enable_recording) streamer.start_stream(real_url, youtube_key, room_name) return streamer # ----------------- Main Program ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/851742623054" + DOUYIN_URL = "https://live.douyin.com/722725928418" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" - hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) - if not hls_instance: + # enable_recording=True 可开启本地录制 + streamer_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, enable_recording=False) + if not streamer_instance: logging.error("推流实例创建失败，程序退出") sys.exit(1) def signal_handler(sig, frame): - if hls_instance: - hls_instance.stop() + if streamer_instance: + streamer_instance.stop() sys.exit(0) signal.signal(signal.SIGINT, signal_handler)