diff --git a/live.py b/live.py new file mode 100644 index 0000000..171f964 --- /dev/null +++ b/live.py @@ -0,0 +1,140 @@ +import os +import subprocess +import threading +import asyncio +import uuid +import queue +import time +import signal +import sys +import spider +import stream + +class HLS2YouTube: + def __init__(self, delay_sec=30): + self.processes = {} + self.stop_flag = threading.Event() + self.delay_sec = delay_sec + self.segment_time = 2 # 内部缓冲每个小片段2秒 + self.buffer_queue = queue.Queue() + + 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() + + 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) + 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}") + + 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}") + + 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=30): + 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 = HLS2YouTube(delay_sec=delay_sec) + hls2yt.start_stream(real_url, youtube_key, room_name) + return hls2yt + +# ----------------- 示例 ----------------- +if __name__ == "__main__": + DOUYIN_URL = "https://live.douyin.com/104914538985" + YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" + + hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY, delay_sec=30) + + 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) + + while True: + time.sleep(60) diff --git a/liveback b/liveback index 74c13a3..182a582 100644 --- a/liveback +++ b/liveback @@ -1,99 +1,73 @@ import os import subprocess -import threading -import time -import glob import asyncio import uuid +import signal +import sys import spider import stream -# import JS_SCRIPT_PATH -# import utils +import threading +import time -# ----------------- HLS2YouTube 推流器 ----------------- class HLS2YouTube: - def __init__(self, cache_dir="cache", segment_time=10, check_interval=1): - self.cache_dir = cache_dir - self.segment_time = segment_time - self.check_interval = check_interval - self.recording_threads = {} - os.makedirs(cache_dir, exist_ok=True) + def __init__(self): + self.processes = {} + self.stop_flag = threading.Event() - def start_record_and_push(self, real_url, youtube_key, room_name="room"): - tmp_segment_prefix = os.path.join(self.cache_dir, f"{room_name}_out") + 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._stream_thread, args=(real_url, youtube_key, room_name), daemon=True).start() - # 清空旧分段 - for f in glob.glob(f"{tmp_segment_prefix}*.ts"): - os.remove(f) + def _stream_thread(self, real_url, youtube_key, room_name): + while not self.stop_flag.is_set(): + 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: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) + self.processes[room_name] = proc + while proc.poll() is None and not self.stop_flag.is_set(): + time.sleep(1) + if proc.poll() is None: + proc.terminate() + print(f"[{room_name}] 推流中断，5秒后重连") + time.sleep(5) + except Exception as e: + print(f"[{room_name}] 推流异常: {e}, 5秒后重试") + time.sleep(5) - # ----------------- 录制线程 ----------------- - def record_thread(): - while True: - record_cmd = [ - "ffmpeg", - "-rw_timeout", "50000000", - "-user_agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", - "-headers", "Referer: https://www.douyin.com/", - "-fflags", "+nobuffer", - "-flags", "low_delay", - "-reconnect", "1", - "-reconnect_at_eof", "1", - "-reconnect_streamed", "1", - "-reconnect_delay_max", "10", - "-i", real_url, - "-c", "copy", - "-f", "segment", - "-segment_time", str(self.segment_time), - "-reset_timestamps", "1", - f"{tmp_segment_prefix}%03d.ts" - ] - try: - subprocess.run(record_cmd) - except Exception as e: - print(f"[{room_name}] 录制异常: {e}, 5秒后重试") - time.sleep(5) - - # ----------------- 推流线程 ----------------- - def push_thread(): - pushed_files = set() - while True: - try: - segments = sorted(glob.glob(f"{tmp_segment_prefix}*.ts")) - new_segments = [s for s in segments if s not in pushed_files] - for seg in new_segments: - push_cmd = [ - "ffmpeg", - "-re", - "-i", seg, - "-c:v", "libx264", "-preset", "veryfast", - "-b:v", "2500k", "-maxrate", "2500k", "-bufsize", "5000k", - "-r", "30", "-g", "60", - "-c:a", "aac", "-b:a", "128k", - "-f", "flv", f"rtmp://a.rtmp.youtube.com/live2/{youtube_key}" - ] - subprocess.run(push_cmd) - pushed_files.add(seg) - time.sleep(self.check_interval) - except Exception as e: - print(f"[{room_name}] 推流异常: {e}, 5秒后重试") - time.sleep(5) - - t1 = threading.Thread(target=record_thread, daemon=True) - t2 = threading.Thread(target=push_thread, daemon=True) - t1.start() - t2.start() - - self.recording_threads[room_name] = (t1, t2) - print(f"[{room_name}] 录制+推流线程已启动") - -_hls2yt_instance = HLS2YouTube() + def stop_all(self): + print("停止所有推流...") + self.stop_flag.set() + for room_name, proc in self.processes.items(): + if proc.poll() is None: + proc.terminate() # ----------------- 抖音解析逻辑 ----------------- async def get_douyin_real_url(record_url, record_quality="原画"): - # import spider # 你的抖音爬虫模块 - # import stream # 你的流解析模块 - - # 根据 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) else: @@ -103,20 +77,33 @@ async def get_douyin_real_url(record_url, record_quality="原画"): # ----------------- 启动推流 ----------------- def start_douyin_to_youtube(record_url, youtube_key, room_name=None): - if not room_name: - room_name = f"抖音_{str(uuid.uuid4())[:8]}" - real_url = asyncio.run(get_douyin_real_url(record_url)) + 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"[{room_name}] 实际直播流: {real_url}") - _hls2yt_instance.start_record_and_push(real_url, youtube_key, room_name) + print(f"实际直播流: {real_url}") + hls2yt = HLS2YouTube() + hls2yt.start_stream(real_url, youtube_key, room_name) + return hls2yt # ----------------- 示例 ----------------- if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/210443559964" # 填入抖音直播 URL - YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" # 填入直播 Key - start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) - # 保持主线程 + DOUYIN_URL = "https://live.douyin.com/104914538985" # 抖音直播 URL + YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" # YouTube 直播 Key + + hls_instance = start_douyin_to_youtube(DOUYIN_URL, YOUTUBE_KEY) + + 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) + while True: time.sleep(60) diff --git a/livee.py b/livee.py deleted file mode 100644 index d9cbe17..0000000 --- a/livee.py +++ /dev/null @@ -1,102 +0,0 @@ -import os -import subprocess -import threading -import time -import glob -import asyncio -import uuid -import spider -import stream - -class Douyin2YouTube: - def __init__(self, cache_dir="cache", segment_time=10): - self.cache_dir = cache_dir - self.segment_time = segment_time - os.makedirs(cache_dir, exist_ok=True) - - async def get_real_url(self, 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(self, record_url, youtube_key, room_name=None): - room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" - real_url = asyncio.run(self.get_real_url(record_url)) - if not real_url: - print(f"[{room_name}] 获取真实直播流失败") - return - print(f"[{room_name}] 实际直播流: {real_url}") - - tmp_prefix = os.path.join(self.cache_dir, f"{room_name}_") - for f in glob.glob(f"{tmp_prefix}*.ts"): - os.remove(f) - - # ----------------- 拉流线程 ----------------- - def record_thread(): - while True: - cmd = [ - "ffmpeg", - "-user_agent", "Mozilla/5.0", - "-headers", "Referer: https://www.douyin.com/", - "-rw_timeout", "15000000", - "-fflags", "+genpts+igndts+nobuffer", - "-flags", "low_delay", - "-reconnect", "1", - "-reconnect_streamed", "1", - "-reconnect_delay_max", "10", - "-i", real_url, - "-c", "copy", - "-f", "segment", - "-segment_time", str(self.segment_time), - "-reset_timestamps", "1", - f"{tmp_prefix}%03d.ts" - ] - try: - subprocess.run(cmd) - except Exception as e: - print(f"[{room_name}] 拉流异常: {e}, 10秒后重试") - time.sleep(10) - - # ----------------- 推流线程 ----------------- - def push_thread(): - pushed_files = set() - while True: - segments = sorted(glob.glob(f"{tmp_prefix}*.ts")) - new_segments = [s for s in segments if s not in pushed_files] - for seg in new_segments: - cmd = [ - "ffmpeg", - "-re", - "-i", seg, - "-c:v", "libx264", "-preset", "veryfast", - "-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: - subprocess.run(cmd) - pushed_files.add(seg) - except Exception as e: - print(f"[{room_name}] 推流异常: {e}, 10秒后重试") - time.sleep(10) - time.sleep(1) - - threading.Thread(target=record_thread, daemon=True).start() - threading.Thread(target=push_thread, daemon=True).start() - print(f"[{room_name}] 转播线程已启动，YouTube 推流保证连续") - -# ----------------- 示例 ----------------- -if __name__ == "__main__": - DOUYIN_URL = "https://live.douyin.com/210443559964" # 填抖音直播 URL - YOUTUBE_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" # 填 YouTube 直播 Key - broadcaster = Douyin2YouTube() - broadcaster.start(DOUYIN_URL, YOUTUBE_KEY) - - while True: - time.sleep(60)