's''

2025-09-13 10:15:00 +08:00
parent e49b8b6865
commit aa26e03ca1
4 changed files with 576 additions and 51 deletions
--- a/live.py
+++ b/live.py
@@ -7,25 +7,32 @@ import queue
 import time
 import signal
 import sys
+import shutil
 import spider
 import stream

-class HLS2YouTube:
-    def __init__(self, delay_sec=30):
+class AdaptiveHLS2YouTube:
+    def __init__(self, max_bitrate=6000, min_bitrate=3000, audio_bitrate=192,
+                 delay_sec=30, segment_time=2, check_interval=5):
        self.processes = {}
        self.stop_flag = threading.Event()
        self.delay_sec = delay_sec
-        self.segment_time = 2  # 内部缓冲每个小片段2秒
+        self.segment_time = segment_time
+        self.check_interval = check_interval
        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",
@@ -42,53 +49,108 @@ class HLS2YouTube:
            "-f", "mpegts",
            "-"
        ]
-        try:
-            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
-            while not self.stop_flag.is_set():
-                data = proc.stdout.read(188*50)  # 每次读小片段
-                if not data:
-                    time.sleep(0.1)
-                    continue
-                self.buffer_queue.put(data)
-        except Exception as e:
-            print(f"拉流异常: {e}")
+
+        while not self.stop_flag.is_set():
+            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*20)
+                    if data:
+                        self.buffer_queue.put(data)
+                    else:
+                        time.sleep(0.01)
+
+                if proc.poll() is not None:
+                    print("[拉流] FFmpeg 退出，5 秒后重启...")
+                    time.sleep(5)
+
+            except Exception as e:
+                print(f"[拉流] 异常: {e}, 5 秒后重试")
+                time.sleep(5)

    def _push_thread(self, youtube_key, room_name):
-        min_buffer = self.delay_sec // self.segment_time
-        cmd = [
-            "ffmpeg",
-            "-re",
-            "-f", "mpegts",
-            "-i", "-",
-            "-c:v", "libx264",
-            "-preset", "veryfast",
-            "-tune", "zerolatency",
-            "-b:v", "3000k",
-            "-maxrate", "3000k",
-            "-bufsize", "6000k",
-            "-r", "30",
-            "-g", "60",
-            "-c:a", "aac",
-            "-b:a", "128k",
-            "-f", "flv",
-            f"rtmp://a.rtmp.youtube.com/live2/{youtube_key}"
-        ]
-        try:
-            proc = subprocess.Popen(cmd, stdin=subprocess.PIPE)
-            self.processes[room_name] = proc
-            # 等待队列蓄水
-            while self.buffer_queue.qsize() < min_buffer and not self.stop_flag.is_set():
-                time.sleep(0.5)
-            # 持续推送
-            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()
-                else:
-                    time.sleep(0.05)
-        except Exception as e:
-            print(f"[{room_name}] 推流异常: {e}")
+        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",
+                "-b:v", f"{self.current_bitrate}k",
+                "-maxrate", f"{self.current_bitrate}k",
+                "-bufsize", f"{self.current_bitrate*4}k",
+                "-r", "30",
+                "-g", "60",
+                "-c:a", "aac",
+                "-b:a", f"{self.audio_bitrate}k",
+                "-ar", "44100",
+                "-ac", "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() and proc.poll() is None:
+                    if not self.buffer_queue.empty():
+                        data = self.buffer_queue.get()
+                        try:
+                            proc.stdin.write(data)
+                            proc.stdin.flush()
+                        except BrokenPipeError:
+                            print(f"[{room_name}] FFmpeg stdin 已断开，重启推流")
+                            break
+                    else:
+                        time.sleep(0.01)
+
+                if proc.poll() is not None:
+                    print(f"[{room_name}] 推流 FFmpeg 退出，5 秒后重启...")
+                    time.sleep(5)
+
+            except Exception as e:
+                print(f"[{room_name}] 推流异常: {e}, 5 秒后重试")
+                time.sleep(5)
+
+    def _abr_thread(self):
+        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 - 500)
+            elif queue_len < 30:
+                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(self.check_interval)

    def stop_all(self):
        print("停止所有推流...")
@@ -97,6 +159,7 @@ class HLS2YouTube:
            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:
@@ -106,6 +169,7 @@ 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:
@@ -117,16 +181,40 @@ def start_douyin_to_youtube(record_url, youtube_key, room_name=None, delay_sec=3
        print("获取真实直播流失败")
        return
    print(f"实际直播流: {real_url}")
-    hls2yt = HLS2YouTube(delay_sec=delay_sec)
+    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 = any(proc and proc.poll() is None for proc in hls_instance.processes.values())
+
+        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/104914538985"
+    DOUYIN_URL = "https://live.douyin.com/848698179845"
    YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z"

    hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=30)
+    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:
@@ -136,5 +224,6 @@ if __name__ == "__main__":
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

+    print("程序已启动，按 Ctrl+C 停止...")
    while True:
        time.sleep(60)