From 6bbf547ab2c9baed489728022c6e11c9d6191199 Mon Sep 17 00:00:00 2001 From: eric Date: Sat, 13 Sep 2025 10:49:03 +0800 Subject: [PATCH] =?UTF-8?q?'5000=E7=A0=81=E7=8E=87'?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- live.py | 57 ++++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/live.py b/live.py index e596af2..ceead1c 100644 --- a/live.py +++ b/live.py @@ -11,10 +11,11 @@ import shutil import spider import stream +# ----------------- 自适应推流器 ----------------- class AdaptiveHLS2YouTube: - def __init__(self, max_bitrate=2500, min_bitrate=1500, audio_bitrate=128, + def __init__(self, max_bitrate=5000, min_bitrate=4000, audio_bitrate=192, delay_sec=30, segment_time=2, check_interval=5): - self.processes = {} + self.processes = {} # room_name -> ffmpeg process self.stop_flag = threading.Event() self.delay_sec = delay_sec self.segment_time = segment_time @@ -25,14 +26,24 @@ class AdaptiveHLS2YouTube: self.audio_bitrate = audio_bitrate self.current_bitrate = max_bitrate - def start_stream(self, real_url, youtube_key, room_name=None): + def start_stream(self, real_urls, youtube_key, room_name=None): + """ + 支持 real_urls 列表，主备流分别推送 + """ 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() + + # 启动拉流线程（可支持多源） + for idx, url in enumerate(real_urls): + threading.Thread(target=self._pull_thread, args=(url, idx), daemon=True).start() + + # 推送线程 threading.Thread(target=self._push_thread, args=(youtube_key, room_name), daemon=True).start() + + # ABR线程 threading.Thread(target=self._abr_thread, daemon=True).start() - def _pull_thread(self, real_url): + def _pull_thread(self, real_url, idx): cmd = [ "ffmpeg", "-rw_timeout", "50000000", @@ -51,7 +62,7 @@ class AdaptiveHLS2YouTube: ] try: proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) - print("[拉流] FFmpeg 启动成功") + print(f"[拉流 {idx}] FFmpeg 启动成功") while not self.stop_flag.is_set(): data = proc.stdout.read(188*50) if not data: @@ -59,7 +70,7 @@ class AdaptiveHLS2YouTube: continue self.buffer_queue.put(data) except Exception as e: - print(f"拉流异常: {e}") + print(f"[拉流 {idx}] 异常: {e}") def _push_thread(self, youtube_key, room_name): min_buffer = max(1, self.delay_sec // self.segment_time) @@ -103,8 +114,11 @@ class AdaptiveHLS2YouTube: while not self.stop_flag.is_set(): if not self.buffer_queue.empty(): data = self.buffer_queue.get() - proc.stdin.write(data) - proc.stdin.flush() + try: + proc.stdin.write(data) + proc.stdin.flush() + except Exception: + break else: time.sleep(0.01) except Exception as e: @@ -114,11 +128,12 @@ class AdaptiveHLS2YouTube: while not self.stop_flag.is_set(): queue_len = self.buffer_queue.qsize() if queue_len > 100: - new_bitrate = max(self.min_bitrate, self.current_bitrate - 300) + new_bitrate = max(self.min_bitrate, self.current_bitrate - 200) elif queue_len < 30: - new_bitrate = min(self.max_bitrate, self.current_bitrate + 300) + new_bitrate = min(self.max_bitrate, self.current_bitrate + 200) else: new_bitrate = self.current_bitrate + if new_bitrate != self.current_bitrate: print(f"[ABR] 调整码率: {self.current_bitrate}k -> {new_bitrate}k") self.current_bitrate = new_bitrate @@ -131,7 +146,6 @@ class AdaptiveHLS2YouTube: if proc.poll() is None: 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: @@ -141,22 +155,24 @@ async def get_douyin_real_url(record_url, record_quality="原画"): port_info = await stream.get_douyin_stream_url(json_data, record_quality, None) return port_info.get("record_url") - # ----------------- 启动推流 ----------------- def start_douyin_to_youtube(record_url, youtube_key, room_name=None, delay_sec=30): try: - real_url = asyncio.run(get_douyin_real_url(record_url)) + # 获取主/备流 + real_url_main = asyncio.run(get_douyin_real_url(record_url)) + real_url_backup = real_url_main.replace("stage", "backup") # 简单替换为备份流 except Exception as e: print(f"获取真实直播流失败: {e}") return - if not real_url: + if not real_url_main: print("获取真实直播流失败") return - print(f"实际直播流: {real_url}") - hls2yt = AdaptiveHLS2YouTube(delay_sec=delay_sec) - hls2yt.start_stream(real_url, youtube_key, room_name) - return hls2yt + print(f"实际直播流主: {real_url_main}") + print(f"实际直播流备: {real_url_backup}") + hls2yt = AdaptiveHLS2YouTube(delay_sec=delay_sec) + hls2yt.start_stream([real_url_main, real_url_backup], youtube_key, room_name) + return hls2yt # ----------------- 控制台监控面板 ----------------- def monitor_panel_thread(hls_instance, interval=2, max_width=50): @@ -175,10 +191,9 @@ def monitor_panel_thread(hls_instance, interval=2, max_width=50): print(f"[监控] 队列: [{queue_bar}] {queue_len:4} | 码率: {bitrate:4}k | 推流活跃: {active_push}", end="") time.sleep(interval) - # ----------------- 示例 ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/848698179845" + DOUYIN_URL = "https://live.douyin.com/364595817295" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=30)