diff --git a/live.py b/live.py index c75402c..23caf84 100644 --- a/live.py +++ b/live.py @@ -19,51 +19,68 @@ class AdaptiveHLS2YouTube: self.delay_sec = delay_sec self.segment_time = segment_time self.check_interval = check_interval - self.buffer_queue = queue.Queue(maxsize=500) # 增大队列，防止阻塞 + self.buffer_queue = queue.Queue() self.max_bitrate = max_bitrate self.min_bitrate = min_bitrate self.audio_bitrate = audio_bitrate self.current_bitrate = max_bitrate - def start_stream(self, real_url, youtube_key, room_name=None): - room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" - print(f"[{room_name}] 启动推流，目标 YouTube Key: {youtube_key}") - threading.Thread(target=self._pull_thread, args=(real_url,), daemon=True).start() - threading.Thread(target=self._push_thread, args=(youtube_key, room_name), daemon=True).start() - threading.Thread(target=self._abr_thread, daemon=True).start() - + # ----------------- 稳定拉流线程 ----------------- def _pull_thread(self, real_url): - cmd = [ - "ffmpeg", - "-rw_timeout", "50000000", - "-user_agent", "Mozilla/5.0", - "-headers", "Referer: https://www.douyin.com/", - "-fflags", "+nobuffer+genpts", - "-flags", "low_delay", - "-reconnect", "1", - "-reconnect_at_eof", "1", - "-reconnect_streamed", "1", - "-reconnect_delay_max", "10", - "-i", real_url, - "-c", "copy", - "-f", "mpegts", - "-" - ] - try: - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) - print("[拉流] FFmpeg 启动成功") - while not self.stop_flag.is_set(): - data = proc.stdout.read(188*2000) # 大块读取，减少空跑 - if not data: - time.sleep(0.05) - continue - try: - self.buffer_queue.put_nowait(data) # 非阻塞写入队列 - except queue.Full: - continue # 队列满直接丢掉，保证不卡 - except Exception as e: - print(f"拉流异常: {e}") + """稳定拉取抖音 HLS，保证 buffer_queue 有数据""" + while not self.stop_flag.is_set(): + cmd = [ + "ffmpeg", + "-rw_timeout", "50000000", + "-user_agent", "Mozilla/5.0", + "-headers", "Referer: https://www.douyin.com/", + "-protocol_whitelist", "file,http,https,tcp,tls,crypto", + "-fflags", "+nobuffer+genpts", + "-flags", "low_delay", + "-i", real_url, + "-c:v", "libx264", + "-preset", "veryfast", + "-tune", "zerolatency", + "-c:a", "aac", + "-b:a", f"{self.audio_bitrate}k", + "-f", "mpegts", + "-" + ] + try: + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + print("[拉流] FFmpeg 启动成功") + + # 打印拉流日志 + def log_stderr(): + while proc.poll() is None and not self.stop_flag.is_set(): + line = proc.stderr.readline() + if line: + print(f"[拉流 FFmpeg]: {line.decode(errors='ignore').strip()}") + threading.Thread(target=log_stderr, daemon=True).start() + + while not self.stop_flag.is_set(): + try: + data = proc.stdout.read(188*2000) # 一次读大块 + if not data: + raise EOFError("拉流 EOF") + self.buffer_queue.put(data, timeout=1) + except queue.Full: + time.sleep(0.05) # 队列满，短暂等待 + except EOFError: + print("[拉流] 遇到 EOF，重启 FFmpeg") + break # 跳出内循环，重新启动 FFmpeg + except Exception as e: + print(f"[拉流] 异常: {e}") + finally: + try: + if proc and proc.poll() is None: + proc.kill() + except: + pass + time.sleep(1) # 重连前等待 + + # ----------------- 推流线程 ----------------- def _push_thread(self, youtube_key, room_name): min_buffer = max(1, self.delay_sec // self.segment_time) while not self.stop_flag.is_set(): @@ -97,6 +114,7 @@ class AdaptiveHLS2YouTube: proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE) self.processes[room_name] = proc + # 打印推流日志 def read_stderr(): while proc.poll() is None: line = proc.stderr.readline() @@ -106,17 +124,15 @@ class AdaptiveHLS2YouTube: while not self.stop_flag.is_set(): try: - data = self.buffer_queue.get(timeout=0.5) - try: - proc.stdin.write(data) - proc.stdin.flush() - except (BrokenPipeError, OSError): - break # 管道断开，退出推流循环 + data = self.buffer_queue.get(timeout=1) + proc.stdin.write(data) + proc.stdin.flush() except queue.Empty: continue except Exception as e: print(f"[{room_name}] 推流异常: {e}") + # ----------------- ABR 自动码率 ----------------- def _abr_thread(self): while not self.stop_flag.is_set(): queue_len = self.buffer_queue.qsize() @@ -129,8 +145,17 @@ class AdaptiveHLS2YouTube: if new_bitrate != self.current_bitrate: print(f"[ABR] 调整码率: {self.current_bitrate}k -> {new_bitrate}k") self.current_bitrate = new_bitrate - time.sleep(max(self.check_interval, 10)) # 降低 ABR 更新频率，避免频繁干扰 + time.sleep(self.check_interval) + # ----------------- 启动推流 ----------------- + def start_stream(self, real_url, youtube_key, room_name=None): + room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" + print(f"[{room_name}] 启动推流，目标 YouTube Key: {youtube_key}") + threading.Thread(target=self._pull_thread, args=(real_url,), daemon=True).start() + threading.Thread(target=self._push_thread, args=(youtube_key, room_name), daemon=True).start() + threading.Thread(target=self._abr_thread, daemon=True).start() + + # ----------------- 停止所有 ----------------- def stop_all(self): print("停止所有推流...") self.stop_flag.set() @@ -139,7 +164,7 @@ class AdaptiveHLS2YouTube: proc.terminate() -# ----------------- 抖音解析逻辑 ----------------- +# ----------------- 抖音解析 ----------------- async def get_douyin_real_url(record_url, record_quality="原画"): 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) @@ -149,7 +174,7 @@ async def get_douyin_real_url(record_url, record_quality="原画"): return port_info.get("record_url") -# ----------------- 启动推流 ----------------- +# ----------------- 启动函数 ----------------- def start_douyin_to_youtube(record_url, youtube_key, room_name=None, delay_sec=40): try: real_url = asyncio.run(get_douyin_real_url(record_url)) @@ -165,7 +190,7 @@ def start_douyin_to_youtube(record_url, youtube_key, room_name=None, delay_sec=4 return hls2yt -# ----------------- 控制台监控面板 ----------------- +# ----------------- 监控面板 ----------------- def monitor_panel_thread(hls_instance, interval=2, max_width=50): term_width, _ = shutil.get_terminal_size() bar_width = min(max_width, term_width - 30) @@ -183,9 +208,9 @@ def monitor_panel_thread(hls_instance, interval=2, max_width=50): time.sleep(interval) -# ----------------- 示例 ----------------- +# ----------------- 示例运行 ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/125742860435" + DOUYIN_URL = "https://live.douyin.com/299336549416" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=40)