's'
This commit is contained in:
eric
115
live.py
115
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)
|
||||
|
||||
Reference in New Issue
Block a user