diff --git a/config.ini b/config.ini new file mode 100644 index 0000000..31625fa --- /dev/null +++ b/config.ini @@ -0,0 +1 @@ +qxvb-r47b-r5ju-6ud3-6k7z \ No newline at end of file diff --git a/live.py b/live.py index f16aec1..d5e0747 100644 --- a/live.py +++ b/live.py @@ -3,202 +3,207 @@ import subprocess import threading import asyncio import uuid -import queue import time import signal import sys import shutil +import logging 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 = {} +# ----------------- Logging Setup ----------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S" +) + +# ----------------- Low-Latency Douyin to YouTube Streamer ----------------- +class DouyinToYouTubeStreamer: + def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', fps=30, gop=60): + self.proc = None 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() - self.max_bitrate = max_bitrate - self.min_bitrate = min_bitrate + self.bitrate = bitrate self.audio_bitrate = audio_bitrate - self.current_bitrate = max_bitrate + self.resolution = resolution + self.fps = fps + self.gop = gop + self.start_time = None + self.room_name = None + self.semaphore = threading.Semaphore(1) + self.pushing = False # 是否正在推流 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() + 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_ffmpeg_loop, args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._monitor_thread, daemon=True).start() + + 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" - 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", - "-" + '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', + '-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', + '-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}' ] - 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*500) # 读取更大块，减少空跑 - if not data: - time.sleep(0.05) - continue - self.buffer_queue.put(data) - except Exception as e: - print(f"拉流异常: {e}") + return cmd - def _push_thread(self, youtube_key, room_name): - min_buffer = max(1, self.delay_sec // self.segment_time) + def _run_ffmpeg_loop(self, real_url, youtube_key): 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", - "-s", "720x1280", - "-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 + 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})") - 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() + 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}") - while not self.stop_flag.is_set(): - try: - data = self.buffer_queue.get(timeout=1) - proc.stdin.write(data) - proc.stdin.flush() - except queue.Empty: - continue + 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})，5秒后重启...") + self.pushing = False except Exception as e: - print(f"[{room_name}] 推流异常: {e}") + logging.error(f"[{self.room_name}] FFmpeg 启动异常: {e}") + finally: + if self.proc and self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc = None + if not self.stop_flag.is_set(): + time.sleep(5) - def _abr_thread(self): + def _monitor_thread(self): + term_width, _ = shutil.get_terminal_size() + bar_width = min(50, term_width - 30) 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) - elif queue_len < 30: - new_bitrate = min(self.max_bitrate, self.current_bitrate + 300) + 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: - 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) + print(f"\r[{self.room_name}] FFmpeg 未运行", end="") + time.sleep(2) - def stop_all(self): - print("停止所有推流...") + def stop(self): + logging.info(f"[{self.room_name}] 停止推流...") self.stop_flag.set() - for proc in self.processes.values(): - if proc.poll() is None: - proc.terminate() + if self.proc and self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self.proc.kill() -# ----------------- 抖音解析逻辑 ----------------- -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): +# ----------------- Douyin Stream Parsing ----------------- +async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=None, cookies=''): try: - real_url = asyncio.run(get_douyin_real_url(record_url)) + 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: - print(f"获取真实直播流失败: {e}") - return + 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, cookies)) 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 + logging.error("获取真实直播流失败") + return None + logging.info(f"实际直播流: {real_url}") + streamer = DouyinToYouTubeStreamer(bitrate=bitrate) + streamer.start_stream(real_url, youtube_key, room_name) + return streamer -# ----------------- 控制台监控面板 ----------------- -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) - - -# ----------------- 示例 ----------------- +# ----------------- Main Program ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/354098231885" + DOUYIN_URL = "https://live.douyin.com/153648759493" YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" - hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=40) + hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) if not hls_instance: - print("推流实例创建失败，程序退出") + logging.error("推流实例创建失败，程序退出") 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() + hls_instance.stop() sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) - print("程序已启动，按 Ctrl+C 停止...") + logging.info("程序已启动，低延迟推流中，按 Ctrl+C 停止...") while True: time.sleep(60) - -# back3 \ No newline at end of file