diff --git a/backup/back8 b/backup/back8 new file mode 100644 index 0000000..7f65b92 --- /dev/null +++ b/backup/back8 @@ -0,0 +1,287 @@ +import os +import subprocess +import threading +import asyncio +import uuid +import time +import signal +import sys +import logging +import platform +import spider +import stream +import aiohttp + +# ----------------- Logging Setup ----------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S" +) + +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, max_minutes=8): + self.push_proc = None + self.record_proc = None + self.stop_flag = threading.Event() + self.bitrate = bitrate + self.audio_bitrate = audio_bitrate + self.resolution = resolution + self.fps = fps + self.gop = gop + self.start_time = None + self.room_name = None + self.pushing = False + self.out_file = None + self.max_minutes = max_minutes + + 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_stream_loop, + args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._monitor_thread, daemon=True).start() + + # ----------------- 构建推流命令 ----------------- + 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', + '-rw_timeout', '10000000', + '-analyzeduration', '1000000', + '-probesize', '5000000', + '-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', + '-i', real_url, + '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-b:v', f'{self.bitrate}k', + '-maxrate', f'{self.bitrate}k', + '-bufsize', f'{self.bitrate * 2}k', + '-r', str(self.fps), + '-g', str(self.gop), + '-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 _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: + 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.PIPE, stderr=subprocess.PIPE + ) + + # ---- 启动本地录制 ---- + record_cmd = self._build_record_cmd(real_url) + self.record_proc = subprocess.Popen( + record_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE + ) + + logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") + logging.info(f"[{self.room_name}] 推流进程 PID: {self.push_proc.pid}") + + # ---- 监控 stderr ---- + def monitor_proc(proc, name): + try: + for line in iter(proc.stderr.readline, b''): + if self.stop_flag.is_set(): + break + if line: + decoded = line.decode('utf-8', errors='ignore').strip() + if name == "push" and not self.pushing and "frame=" in decoded and "fps=" in decoded: + self.pushing = True + logging.info(f"[{self.room_name}] 推流到 YouTube 成功") + logging.debug(f"[{self.room_name} {name}]: {decoded}") + 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() + 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 or self.record_proc.poll() is not None: + logging.warning(f"[{self.room_name}] FFmpeg 进程退出，1秒后重启...") + break + time.sleep(2) + + except Exception as e: + logging.error(f"[{self.room_name}] FFmpeg异常: {e}") + finally: + self._terminate_procs() + self._finalize_recording() + if not self.stop_flag.is_set(): + time.sleep(1) + + 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(2) + + # ----------------- 停止 ----------------- + def stop(self): + logging.info(f"[{self.room_name}] 停止推流和录制...") + self.stop_flag.set() + self._terminate_procs() + self._finalize_recording() + + def _finalize_recording(self): + if self.out_file and os.path.exists(self.out_file): + try: + if self.record_proc and self.record_proc.poll() is None: + sig = signal.CTRL_BREAK_EVENT if IS_WINDOWS else signal.SIGINT + self.record_proc.send_signal(sig) + try: + self.record_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self.record_proc.kill() + except Exception: + pass + + elapsed = int(time.time() - self.start_time) if self.start_time else 0 + suffix = f"_{elapsed // 60}min" if elapsed >= 60 else f"_{elapsed}s" + new_name = self.out_file.replace(".mp4", f"{suffix}.mp4") + try: + os.rename(self.out_file, new_name) + except FileExistsError: + new_name = self.out_file.replace(".mp4", f"_{uuid.uuid4().hex[:4]}.mp4") + os.rename(self.out_file, new_name) + size_mb = os.path.getsize(new_name) / (1024 * 1024) + logging.info(f"[{self.room_name}] 本地录制已结束: {new_name} | 大小: {size_mb:.2f} MB") + +# ----------------- 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: + 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"初始化 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=proxy_addr, cookies=cookies)) + if not real_url: + logging.error("获取真实直播流失败") + return None + logging.info(f"实际直播流: {real_url}") + streamer = DouyinToYouTubeStreamer(bitrate=bitrate) + streamer.start_stream(real_url, youtube_key, room_name) + return streamer + +# ----------------- Main Program ----------------- +if __name__ == "__main__": + DOUYIN_URL = "https://live.douyin.com/321245312734" + YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" + + streamer_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) + if not streamer_instance: + logging.error("推流实例创建失败，程序退出") + sys.exit(1) + + def signal_handler(sig, frame): + if streamer_instance: + streamer_instance.stop() + sys.exit(0) + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + logging.info("程序已启动，低延迟推流中，按 Ctrl+C 停止...") + while True: + time.sleep(60) diff --git a/backup/back9 b/backup/back9 new file mode 100644 index 0000000..41de1a2 --- /dev/null +++ b/backup/back9 @@ -0,0 +1,267 @@ +import os +import subprocess +import threading +import asyncio +import uuid +import time +import signal +import sys +import logging +import platform +import spider +import stream +import aiohttp + +# ----------------- Logging Setup ----------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S" +) + +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, 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 + self.resolution = resolution + self.fps = fps + self.gop = gop + self.start_time = None + self.room_name = None + 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_stream_loop, + args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._monitor_thread, daemon=True).start() + + # ----------------- 构建推流命令 ----------------- + 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', + '-rw_timeout', '10000000', + '-analyzeduration', '1000000', + '-probesize', '5000000', + '-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', + '-i', real_url, + '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-b:v', f'{self.bitrate}k', + '-maxrate', f'{self.bitrate}k', + '-bufsize', f'{self.bitrate * 2}k', + '-r', str(self.fps), + '-g', str(self.gop), + '-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 _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: + 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) + 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) + logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") + + # ---- 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) + 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 + time.sleep(2) + + 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 _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}] 停止推流和录制...") + self.stop_flag.set() + 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: + 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"初始化 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)) + if not real_url: + logging.error("获取真实直播流失败") + return None + logging.info(f"实际直播流: {real_url}") + 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" + YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" + + # 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 streamer_instance: + streamer_instance.stop() + sys.exit(0) + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + logging.info("程序已启动，低延迟推流中，按 Ctrl+C 停止...") + while True: + time.sleep(60) diff --git a/live.py b/live.py index 7f65b92..39970ae 100644 --- a/live.py +++ b/live.py @@ -6,11 +6,10 @@ import uuid import time import signal import sys +import shutil import logging -import platform import spider import stream -import aiohttp # ----------------- Logging Setup ----------------- logging.basicConfig( @@ -19,14 +18,10 @@ logging.basicConfig( datefmt="%H:%M:%S" ) -IS_WINDOWS = platform.system() == "Windows" - -# ----------------- Douyin to YouTube Streamer ----------------- +# ----------------- Low-Latency Douyin to YouTube Streamer ----------------- class DouyinToYouTubeStreamer: - def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', - fps=30, gop=60, max_minutes=8): - self.push_proc = None - self.record_proc = None + def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', fps=30, gop=60): + self.proc = None self.stop_flag = threading.Event() self.bitrate = bitrate self.audio_bitrate = audio_bitrate @@ -35,227 +30,153 @@ class DouyinToYouTubeStreamer: self.gop = gop self.start_time = None self.room_name = None - self.pushing = False - self.out_file = None - self.max_minutes = max_minutes - - os.makedirs("records", exist_ok=True) + self.semaphore = threading.Semaphore(1) + self.pushing = False # 是否正在推流 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_stream_loop, - args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._run_ffmpeg_loop, args=(real_url, youtube_key), daemon=True).start() threading.Thread(target=self._monitor_thread, daemon=True).start() - # ----------------- 构建推流命令 ----------------- - 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") + 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") headers = "Referer: https://www.douyin.com/\r\nOrigin: https://www.douyin.com\r\n" + cmd = [ - 'ffmpeg', '-y', + 'ffmpeg', '-y', '-v', 'error', '-hide_banner', '-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', + '-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, '-b:v', f'{self.bitrate}k', '-maxrate', f'{self.bitrate}k', '-bufsize', f'{self.bitrate * 2}k', - '-r', str(self.fps), - '-g', str(self.gop), - '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', + '-c:a', 'aac', + '-b:a', f'{self.audio_bitrate}k', + '-ar', '44100', + '-ac', '2', + '-strict', '-2', + '-avoid_negative_ts', 'make_zero', '-f', 'flv', f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}' ] return cmd - # ----------------- 构建录制命令 ----------------- - 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): + def _run_ffmpeg_loop(self, real_url, youtube_key): while not self.stop_flag.is_set(): try: - self.start_time = time.time() - self.pushing = False + 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 + logging.info(f"[{self.room_name}] FFmpeg 启动成功 (PID: {self.proc.pid})") - # ---- 启动推流 ---- - push_cmd = self._build_push_cmd(real_url, youtube_key) - self.push_proc = subprocess.Popen( - push_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE - ) + 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}") - # ---- 启动本地录制 ---- - record_cmd = self._build_record_cmd(real_url) - self.record_proc = subprocess.Popen( - record_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE - ) - - logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") - logging.info(f"[{self.room_name}] 推流进程 PID: {self.push_proc.pid}") - - # ---- 监控 stderr ---- - def monitor_proc(proc, name): - try: - for line in iter(proc.stderr.readline, b''): - if self.stop_flag.is_set(): - break - if line: - decoded = line.decode('utf-8', errors='ignore').strip() - if name == "push" and not self.pushing and "frame=" in decoded and "fps=" in decoded: - self.pushing = True - logging.info(f"[{self.room_name}] 推流到 YouTube 成功") - logging.debug(f"[{self.room_name} {name}]: {decoded}") - 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() - 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 or self.record_proc.poll() is not None: - logging.warning(f"[{self.room_name}] FFmpeg 进程退出，1秒后重启...") - break - time.sleep(2) + 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}") + logging.error(f"[{self.room_name}] FFmpeg 启动异常: {e}") finally: - self._terminate_procs() - self._finalize_recording() + 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.proc = None if not self.stop_flag.is_set(): time.sleep(1) - 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): + term_width, _ = shutil.get_terminal_size() + bar_width = min(50, term_width - 30) 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="") + 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 stop(self): - logging.info(f"[{self.room_name}] 停止推流和录制...") + logging.info(f"[{self.room_name}] 停止推流...") self.stop_flag.set() - self._terminate_procs() - self._finalize_recording() - - def _finalize_recording(self): - if self.out_file and os.path.exists(self.out_file): + if self.proc and self.proc.poll() is None: + self.proc.terminate() try: - if self.record_proc and self.record_proc.poll() is None: - sig = signal.CTRL_BREAK_EVENT if IS_WINDOWS else signal.SIGINT - self.record_proc.send_signal(sig) - try: - self.record_proc.wait(timeout=5) - except subprocess.TimeoutExpired: - self.record_proc.kill() - except Exception: - pass + self.proc.wait(timeout=1) + except subprocess.TimeoutExpired: + self.proc.kill() - elapsed = int(time.time() - self.start_time) if self.start_time else 0 - suffix = f"_{elapsed // 60}min" if elapsed >= 60 else f"_{elapsed}s" - new_name = self.out_file.replace(".mp4", f"{suffix}.mp4") - try: - os.rename(self.out_file, new_name) - except FileExistsError: - new_name = self.out_file.replace(".mp4", f"_{uuid.uuid4().hex[:4]}.mp4") - os.rename(self.out_file, new_name) - size_mb = os.path.getsize(new_name) / (1024 * 1024) - logging.info(f"[{self.room_name}] 本地录制已结束: {new_name} | 大小: {size_mb:.2f} MB") # ----------------- 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: - 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 + 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") except Exception as e: - logging.error(f"初始化 aiohttp session 失败: {e}") + logging.error(f"获取抖音真实流失败: {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=proxy_addr, cookies=cookies)) +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)) if not real_url: logging.error("获取真实直播流失败") return None @@ -264,19 +185,20 @@ def start_douyin_to_youtube(record_url, youtube_key, room_name=None, streamer.start_stream(real_url, youtube_key, room_name) return streamer + # ----------------- Main Program ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/321245312734" + DOUYIN_URL = "https://live.douyin.com/851742623054" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" - streamer_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) - if not streamer_instance: + hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) + if not hls_instance: logging.error("推流实例创建失败，程序退出") sys.exit(1) def signal_handler(sig, frame): - if streamer_instance: - streamer_instance.stop() + if hls_instance: + hls_instance.stop() sys.exit(0) signal.signal(signal.SIGINT, signal_handler)