diff --git a/back b/backup/back similarity index 100% rename from back rename to backup/back diff --git a/back2 b/backup/back2 similarity index 100% rename from back2 rename to backup/back2 diff --git a/back3 b/backup/back3 similarity index 100% rename from back3 rename to backup/back3 diff --git a/back4 b/backup/back4 similarity index 100% rename from back4 rename to backup/back4 diff --git a/backup/back5 b/backup/back5 new file mode 100644 index 0000000..ef4db7c --- /dev/null +++ b/backup/back5 @@ -0,0 +1,220 @@ +import os +import subprocess +import threading +import asyncio +import uuid +import queue +import time +import signal +import sys +import shutil +import spider +import stream + + +class AdaptiveHLS2YouTube: + def __init__(self, max_bitrate=5000, min_bitrate=3000, audio_bitrate=192, + delay_sec=40, segment_time=2, check_interval=5): + self.processes = {} + self.stop_flag = threading.Event() + self.delay_sec = delay_sec + self.segment_time = segment_time + self.check_interval = check_interval + self.buffer_queue = queue.Queue(maxsize=800) # 队列加大 + 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.PIPE) + print("[拉流] FFmpeg 启动成功") + + def read_stderr(): + while proc.poll() is None: + line = proc.stderr.readline() + if line: + print(f"[拉流FFmpeg]: {line.decode(errors='ignore').strip()}") + threading.Thread(target=read_stderr, daemon=True).start() + + while not self.stop_flag.is_set(): + data = proc.stdout.read(188 * 5000) # 提升吞吐 + 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}") + + 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(): + if self.buffer_queue.qsize() < min_buffer: + time.sleep(0.05) + continue + + cmd = [ + "ffmpeg", + "-re", + "-f", "mpegts", + "-i", "-", + "-c:v", "libx264", + "-preset", "veryfast", + "-tune", "zerolatency", + "-x264-params", "keyint=60:scenecut=0", + "-b:v", f"{self.current_bitrate}k", + "-maxrate", f"{self.current_bitrate}k", + "-bufsize", f"{self.current_bitrate*4}k", + "-r", "30", + "-g", "60", + "-s", "720x1280", + "-c:a", "aac", + "-b:a", f"{self.audio_bitrate}k", + "-ar", "44100", + "-ac", "2", + "-strict", "-2", + "-f", "flv", + "-rtmp_live", "live", + f"rtmp://a.rtmp.youtube.com/live2/{youtube_key}" + ] + try: + 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() + if line: + print(f"[{room_name} FFmpeg]: {line.decode(errors='ignore').strip()}") + threading.Thread(target=read_stderr, daemon=True).start() + + 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): + print(f"[{room_name}] 推流管道断开，准备重启") + break + except queue.Empty: + continue + except Exception as e: + print(f"[{room_name}] 推流异常: {e}") + + def _abr_thread(self): + while not self.stop_flag.is_set(): + queue_len = self.buffer_queue.qsize() + if queue_len > 200: + new_bitrate = max(self.min_bitrate, self.current_bitrate - 500) + elif queue_len < 20: + new_bitrate = min(self.max_bitrate, self.current_bitrate + 500) + 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 + time.sleep(max(self.check_interval, 10)) + + def stop_all(self): + print("停止所有推流...") + self.stop_flag.set() + for proc in self.processes.values(): + 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: + json_data = await spider.get_douyin_stream_data(url=record_url) + else: + json_data = await spider.get_douyin_app_stream_data(url=record_url) + 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=40): + try: + real_url = asyncio.run(get_douyin_real_url(record_url)) + except Exception as e: + print(f"获取真实直播流失败: {e}") + return + if not real_url: + print("获取真实直播流失败") + return + print(f"实际直播流: {real_url}") + hls2yt = AdaptiveHLS2YouTube(delay_sec=delay_sec) + hls2yt.start_stream(real_url, youtube_key, room_name) + 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) + + while not hls_instance.stop_flag.is_set(): + queue_len = hls_instance.buffer_queue.qsize() + bitrate = hls_instance.current_bitrate + active_push = sum(1 for proc in hls_instance.processes.values() if proc and proc.poll() is None) + + queue_bar_len = min(bar_width, queue_len) + queue_bar = "#" * queue_bar_len + "-" * (bar_width - queue_bar_len) + + print("\r", end="") + 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/592179817312" + YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" + + hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=40) + if not hls_instance: + print("推流实例创建失败，程序退出") + sys.exit(1) + + threading.Thread(target=monitor_panel_thread, args=(hls_instance,), daemon=True).start() + + def signal_handler(sig, frame): + if hls_instance: + hls_instance.stop_all() + sys.exit(0) + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + print("程序已启动，按 Ctrl+C 停止...") + while True: + time.sleep(60) diff --git a/list b/list index 9a99421..384d35b 100644 --- a/list +++ b/list @@ -1,3 +1,4 @@ https://live.douyin.com/525514386431 https://live.douyin.com/8687122573 -https://live.douyin.com/642534242822 \ No newline at end of file +https://live.douyin.com/642534242822 +https://live.douyin.com/589037028237 \ No newline at end of file diff --git a/live.py b/live.py index 511680b..8ae0740 100644 --- a/live.py +++ b/live.py @@ -20,7 +20,7 @@ class AdaptiveHLS2YouTube: self.delay_sec = delay_sec self.segment_time = segment_time self.check_interval = check_interval - self.buffer_queue = queue.Queue(maxsize=800) # 队列加大 + self.buffer_queue = queue.Queue(maxsize=800) self.max_bitrate = max_bitrate self.min_bitrate = min_bitrate self.audio_bitrate = audio_bitrate @@ -45,33 +45,41 @@ class AdaptiveHLS2YouTube: "-reconnect_at_eof", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "10", + "-f", "hls", # 强制 HLS "-i", real_url, "-c", "copy", "-f", "mpegts", "-" ] try: - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=10**8) print("[拉流] FFmpeg 启动成功") def read_stderr(): while proc.poll() is None: - line = proc.stderr.readline() - if line: + try: + line = proc.stderr.readline() + if not line: + continue print(f"[拉流FFmpeg]: {line.decode(errors='ignore').strip()}") + except Exception: + continue threading.Thread(target=read_stderr, daemon=True).start() while not self.stop_flag.is_set(): - data = proc.stdout.read(188 * 5000) # 提升吞吐 - if not data: - time.sleep(0.05) - continue try: + data = proc.stdout.read(188 * 5000) + if not data: + time.sleep(0.05) + continue self.buffer_queue.put_nowait(data) except queue.Full: continue + except Exception as e: + print(f"[拉流异常] {e}") + time.sleep(1) except Exception as e: - print(f"拉流异常: {e}") + print(f"拉流进程异常: {e}") def _push_thread(self, youtube_key, room_name): min_buffer = max(1, self.delay_sec // self.segment_time) @@ -105,19 +113,25 @@ class AdaptiveHLS2YouTube: f"rtmp://a.rtmp.youtube.com/live2/{youtube_key}" ] try: - proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE) + proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=10**8) self.processes[room_name] = proc def read_stderr(): while proc.poll() is None: - line = proc.stderr.readline() - if line: + try: + line = proc.stderr.readline() + if not line: + continue print(f"[{room_name} FFmpeg]: {line.decode(errors='ignore').strip()}") + except Exception: + continue threading.Thread(target=read_stderr, daemon=True).start() while not self.stop_flag.is_set(): try: data = self.buffer_queue.get(timeout=0.5) + if not data: + continue try: proc.stdin.write(data) proc.stdin.flush() @@ -126,42 +140,54 @@ class AdaptiveHLS2YouTube: break except queue.Empty: continue + except Exception as e: + print(f"[{room_name}] 推流异常: {e}") + break except Exception as e: - print(f"[{room_name}] 推流异常: {e}") + print(f"[{room_name}] 推流进程启动失败: {e}") + time.sleep(5) def _abr_thread(self): while not self.stop_flag.is_set(): - queue_len = self.buffer_queue.qsize() - if queue_len > 200: - new_bitrate = max(self.min_bitrate, self.current_bitrate - 500) - elif queue_len < 20: - new_bitrate = min(self.max_bitrate, self.current_bitrate + 500) - 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 + try: + queue_len = self.buffer_queue.qsize() + if queue_len > 200: + new_bitrate = max(self.min_bitrate, self.current_bitrate - 500) + elif queue_len < 20: + new_bitrate = min(self.max_bitrate, self.current_bitrate + 500) + 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 + except Exception: + pass time.sleep(max(self.check_interval, 10)) def stop_all(self): print("停止所有推流...") self.stop_flag.set() for proc in self.processes.values(): - if proc.poll() is None: - proc.terminate() + try: + if proc and proc.poll() is None: + proc.terminate() + except Exception: + continue -# ----------------- 抖音解析逻辑 ----------------- 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) - else: - json_data = await spider.get_douyin_app_stream_data(url=record_url) - port_info = await stream.get_douyin_stream_url(json_data, record_quality, None) - return port_info.get("record_url") + 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) + else: + json_data = await spider.get_douyin_app_stream_data(url=record_url) + port_info = await stream.get_douyin_stream_url(json_data, record_quality, None) + return port_info.get("record_url") + except Exception as e: + print(f"获取抖音真实地址失败: {e}") + return None -# ----------------- 启动推流 ----------------- 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)) @@ -177,27 +203,28 @@ 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() + term_width, _ = shutil.get_terminal_size(fallback=(80, 20)) bar_width = min(max_width, term_width - 30) while not hls_instance.stop_flag.is_set(): - queue_len = hls_instance.buffer_queue.qsize() - bitrate = hls_instance.current_bitrate - active_push = sum(1 for proc in hls_instance.processes.values() if proc and proc.poll() is None) + try: + queue_len = hls_instance.buffer_queue.qsize() + bitrate = hls_instance.current_bitrate + active_push = sum(1 for proc in hls_instance.processes.values() if proc and proc.poll() is None) - queue_bar_len = min(bar_width, queue_len) - queue_bar = "#" * queue_bar_len + "-" * (bar_width - queue_bar_len) + queue_bar_len = min(bar_width, queue_len) + queue_bar = "#" * queue_bar_len + "-" * (bar_width - queue_bar_len) - print("\r", end="") - print(f"[监控] 队列: [{queue_bar}] {queue_len:4} | 码率: {bitrate:4}k | 推流进程: {active_push}", end="") + print("\r", end="") + print(f"[监控] 队列: [{queue_bar}] {queue_len:4} | 码率: {bitrate:4}k | 推流进程: {active_push}", end="") + except Exception: + pass time.sleep(interval) -# ----------------- 示例 ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/89581874559" + DOUYIN_URL = "https://live.douyin.com/97282309368" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=40)