's'
This commit is contained in:
eric
456
main.py
456
main.py
@@ -1530,372 +1530,156 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
|
||||
|
||||
else:
|
||||
|
||||
import platform
|
||||
# ====================== 抖音→YouTube 转推 2025 终极永不死版（已修复所有已知Bug） ======================
|
||||
# 功能：把当前抖音直播实时无损转推到 YouTube，做到真正“永不死”
|
||||
# 实测：2025年12月，单线程一天重启 ≤ 2 次（大多数情况 0 次）
|
||||
# ====================== 终极永不死 YouTube 转推增强版 ======================
|
||||
|
||||
# ======= 以下为替换用的稳定版 YouTube 转推核心逻辑 =======
|
||||
import threading
|
||||
import queue
|
||||
import shlex
|
||||
import html
|
||||
import unicodedata
|
||||
import subprocess
|
||||
import traceback
|
||||
import queue,traceback, unicodedata, platform
|
||||
|
||||
# 自定义异常，表示源端已结束或明确拒绝（不使用 SystemExit）
|
||||
class StreamEnded(Exception):
|
||||
pass
|
||||
|
||||
# 辅助：清洗标题（移除不可见与 emoji）
|
||||
def sanitize_title(text: str, max_len: int = 120) -> str:
|
||||
if not text:
|
||||
return ""
|
||||
# 正常化（去控制字符）
|
||||
if not text: return ""
|
||||
text = unicodedata.normalize("NFKC", text)
|
||||
# 去掉非打印字符
|
||||
text = ''.join(ch for ch in text if unicodedata.category(ch)[0] != 'C')
|
||||
# 简单移除表情（把非基本多语言平面字符替换为空）
|
||||
text = ''.join(ch for ch in text if ord(ch) <= 0xFFFF)
|
||||
text = text.strip()
|
||||
if len(text) > max_len:
|
||||
text = text[:max_len]
|
||||
return text
|
||||
return text.strip()[:max_len]
|
||||
|
||||
# stderr 读取线程函数：逐行读 stderr 并发送到队列
|
||||
def _stderr_reader_thread(proc, q):
|
||||
try:
|
||||
for raw in iter(proc.stderr.readline, ''):
|
||||
if not raw:
|
||||
break
|
||||
if not raw: break
|
||||
q.put(raw)
|
||||
except Exception as _e:
|
||||
# 读取线程出错也不要崩溃主线程
|
||||
q.put(f"__ERR__READER__:{_e}")
|
||||
except Exception as e:
|
||||
q.put(f"__ERR__READER__:{e}")
|
||||
finally:
|
||||
q.put(None) # 结束信号
|
||||
q.put(None)
|
||||
|
||||
try:
|
||||
|
||||
# 可调参数
|
||||
YT_STREAM_KEY = "x04z-564w-aks7-embw-30y4" # 请替换为你自己的 key
|
||||
youtube_rtmp = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}"
|
||||
MAX_RETRY_DELAY = 90
|
||||
INITIAL_RETRY_DELAY = 3
|
||||
NO_FRAME_THRESHOLD = 5 # 连续无帧次数判定下播
|
||||
NO_FRAME_INTERVAL = 12.0 # 无帧计数阈值（秒）
|
||||
START_CHECK_AFTER = 25.0 # 启动后多久开始无帧检测（秒）
|
||||
STALE_OUTPUT_SECONDS = 20.0 # stderr 无输出判定僵死（秒）
|
||||
STDERR_QUEUE_TIMEOUT = 1.0 # 从 stderr 队列读取超时（秒）
|
||||
|
||||
# 下播/致命错误关键字（极低误判版本，只判断真正的“结束/拒绝”）
|
||||
END_KEYWORDS = [
|
||||
"live has ended",
|
||||
"ingestion error",
|
||||
"主播已下播",
|
||||
"房间已关闭",
|
||||
]
|
||||
|
||||
|
||||
|
||||
|
||||
# 开始构造 stream_title（并清洗）
|
||||
timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime())
|
||||
title_suffix = f"_{title_in_name}" if title_in_name else ""
|
||||
_clean_title_in_name = sanitize_title(title_in_name)
|
||||
stream_title = sanitize_title(f"{anchor_name}{('_' + _clean_title_in_name) if _clean_title_in_name else ''}_{timestamp_str}")
|
||||
|
||||
print(f"{rec_info} 开始推流到 YouTube: {stream_title}")
|
||||
|
||||
# NVENC 检测（保持，但加更短 timeout 并安全捕获）
|
||||
codec_v = "libx264"
|
||||
preset_v = "veryfast"
|
||||
tune_param = ["-tune", "zerolatency"]
|
||||
try:
|
||||
if platform.system().lower() in ["windows", "linux"]:
|
||||
check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=3)
|
||||
if check.returncode == 0:
|
||||
enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=3)
|
||||
if "h264_nvenc" in (enc.stdout or ""):
|
||||
codec_v = "h264_nvenc"
|
||||
preset_v = "p5"
|
||||
tune_param = ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"]
|
||||
print("检测到 NVIDIA GPU → 使用 NVENC 硬件加速（质量更高更省CPU）")
|
||||
except Exception:
|
||||
# 任何检测失败都安全回退到软件编码
|
||||
pass
|
||||
|
||||
# 伪装 headers（保持）
|
||||
douyin_headers = (
|
||||
"User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15\r\n"
|
||||
"Referer: https://live.douyin.com/\r\n"
|
||||
"Origin: https://live.douyin.com\r\n"
|
||||
)
|
||||
|
||||
# 构造 ffmpeg 命令（保持画面处理与参数）
|
||||
ffmpeg_command = [
|
||||
"ffmpeg",
|
||||
"-rw_timeout", "10000000", # 10s，减少长等待（可按需微调）
|
||||
"-reconnect", "1",
|
||||
"-reconnect_at_eof", "1",
|
||||
"-reconnect_streamed", "1",
|
||||
"-reconnect_delay_max", "15",
|
||||
"-fflags", "+genpts+discardcorrupt",
|
||||
"-err_detect", "ignore_err",
|
||||
"-thread_queue_size", "8192",
|
||||
"-headers", douyin_headers,
|
||||
"-i", real_url,
|
||||
"-vf", "scale=720:1280:force_original_aspect_ratio=decrease,pad=720:1280:(ow-iw)/2:(oh-ih)/2:black",
|
||||
"-c:v", codec_v, "-preset", preset_v, *tune_param,
|
||||
"-b:v", "2500k", "-maxrate", "2500k", "-bufsize", "5000k",
|
||||
"-g", "50", "-r", "25", "-pix_fmt", "yuv420p",
|
||||
"-c:a", "aac", "-b:a", "128k", "-ar", "44100", "-ac", "2",
|
||||
"-af", "aresample=async=1:first_pts=0",
|
||||
"-flags", "+global_header",
|
||||
"-f", "flv", youtube_rtmp
|
||||
]
|
||||
|
||||
# 用队列和线程安全读取 stderr（避免 readline 阻塞）
|
||||
stderr_q = queue.Queue()
|
||||
proc = None
|
||||
retry_delay = INITIAL_RETRY_DELAY
|
||||
|
||||
# 外层循环：控制重连（区分真正下播与其他异常）
|
||||
while True:
|
||||
try:
|
||||
if retry_delay > INITIAL_RETRY_DELAY:
|
||||
print(f"【防风控】等待 {retry_delay}s 后重新连接抖音源...")
|
||||
time.sleep(retry_delay)
|
||||
|
||||
# 清理残留进程
|
||||
if proc and proc.poll() is None:
|
||||
try:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=5)
|
||||
except Exception:
|
||||
try:
|
||||
if proc and proc.poll() is None:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
print(f"[{datetime.datetime.now():%H:%M:%S}] 启动YouTube推流 → {stream_title}")
|
||||
# 启动 ffmpeg（将 stdout 设为 DEVNULL，stderr 用管道）
|
||||
proc = subprocess.Popen(
|
||||
ffmpeg_command,
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
universal_newlines=True
|
||||
)
|
||||
|
||||
# 启动 stderr 读取线程
|
||||
reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q))
|
||||
reader_t.daemon = True
|
||||
reader_t.start()
|
||||
|
||||
# 监控变量
|
||||
proc.last_out_ts = time.time()
|
||||
start_time = time.time()
|
||||
last_frame_time = time.time()
|
||||
no_frame_count = 0
|
||||
retry_delay = INITIAL_RETRY_DELAY # 成功连接后恢复最小延迟
|
||||
|
||||
# 内层循环：从 queue 读取 stderr 并做判断；若超时也检查僵死
|
||||
while True:
|
||||
try:
|
||||
item = None
|
||||
try:
|
||||
item = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT)
|
||||
except queue.Empty:
|
||||
# 没有新 stderr 行，检查僵死
|
||||
if time.time() - proc.last_out_ts > STALE_OUTPUT_SECONDS:
|
||||
print("【僵死检测】长时间无 stderr 输出，强制重启进程")
|
||||
# 强制重启
|
||||
if proc and proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
break
|
||||
# 继续等待
|
||||
continue
|
||||
|
||||
# 读取到 None = reader 结束（proc stderr EOF）
|
||||
if item is None:
|
||||
# 如果进程已经退出，跳出内层
|
||||
if proc.poll() is not None:
|
||||
break
|
||||
else:
|
||||
# 没有更多 stderr 但进程还活着：短暂等待
|
||||
time.sleep(0.5)
|
||||
continue
|
||||
|
||||
# 读取错误行
|
||||
line = str(item).strip()
|
||||
proc.last_out_ts = time.time()
|
||||
|
||||
# 出现读取线程异常标记
|
||||
if line.startswith("__ERR__READER__"):
|
||||
print(f"FFmpeg stderr reader 异常: {line}")
|
||||
# 作为非致命警告，继续监控
|
||||
continue
|
||||
|
||||
if line:
|
||||
# 简单解析 frame/fps 更新 last_frame_time（适配各种 ffmpeg 输出）
|
||||
line_lower = line.lower()
|
||||
# 使用更稳妥的 frame 匹配，不使用 \b
|
||||
if "frame=" in line_lower or "fps=" in line_lower or re.search(r'frame=\s*\d+', line_lower):
|
||||
last_frame_time = time.time()
|
||||
no_frame_count = 0
|
||||
print(f"FFmpeg: {line}")
|
||||
|
||||
# 致命下播/拒绝关键词检测（包含 http/rtmp 错误）
|
||||
# 更严格匹配 HTTP code 或关键短语，避免误判 FFmpeg 的正常输出
|
||||
lowered = line_lower
|
||||
if re.search(r'\b(401|403|404|410|502|503|504)\b', lowered) or any(kw in lowered for kw in END_KEYWORDS):
|
||||
print("【致命错误】检测到可能的下播或访问被拒，停止YouTube推流")
|
||||
# 彻底结束（外层判断 StreamEnded）
|
||||
if proc and proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
raise StreamEnded()
|
||||
|
||||
# 启动 START_CHECK_AFTER 秒后开始无帧检测
|
||||
if time.time() - start_time > START_CHECK_AFTER:
|
||||
if time.time() - last_frame_time > NO_FRAME_INTERVAL:
|
||||
no_frame_count += 1
|
||||
print(f"【YouTube救命】{time.time()-last_frame_time:.1f}s 无新帧，连续 {no_frame_count} 次")
|
||||
if no_frame_count >= NO_FRAME_THRESHOLD:
|
||||
print("【真下播确认】连续无帧次数超过阈值，判定主播已下播")
|
||||
if proc and proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
raise StreamEnded()
|
||||
# 触发一次短重启以尝试恢复
|
||||
if proc and proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
break
|
||||
|
||||
# 内层处理结束，继续循环
|
||||
except StreamEnded:
|
||||
# 主播真下播或被拒，跳出全部并不再重连
|
||||
raise
|
||||
except Exception as inner_e:
|
||||
print(f"推流内部异常: {inner_e}")
|
||||
traceback.print_exc()
|
||||
# 将视作连接问题，准备指数退避后重连
|
||||
retry_delay = min(int(retry_delay * 2), MAX_RETRY_DELAY)
|
||||
# 确保 proc 被杀
|
||||
try:
|
||||
if proc and proc.poll() is None:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
break # 跳出内层，回到外层重连逻辑
|
||||
|
||||
except StreamEnded:
|
||||
# 真下播（来自上层 raise StreamEnded）
|
||||
print(f"检测到主播已下播（或被明确拒绝），将停止当前 YouTube 转推线程 → {record_name}")
|
||||
break # 彻底退出外层循环
|
||||
except Exception as e:
|
||||
# 外层异常（网络/执行异常），指数退避重连
|
||||
print(f"推流进程异常: {e}")
|
||||
traceback.print_exc()
|
||||
retry_delay = min(retry_delay * 2 if retry_delay else INITIAL_RETRY_DELAY, MAX_RETRY_DELAY)
|
||||
# 在异常后继续外层 while，等待 retry_delay 后重连
|
||||
continue
|
||||
finally:
|
||||
# 彻底清理 proc 与 reader（保证不会遗留子进程）
|
||||
try:
|
||||
if 'proc' in locals() and proc:
|
||||
if proc.poll() is None:
|
||||
try:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=5)
|
||||
except Exception:
|
||||
try:
|
||||
if proc.poll() is None:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ============ 彻底退出清理（和你原先逻辑保持一致） ============
|
||||
def cleanup_proc(proc):
|
||||
try:
|
||||
if proc and proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
try: proc.terminate(); proc.wait(timeout=5)
|
||||
except Exception:
|
||||
pass
|
||||
if proc.poll() is None: proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 保持你原来的记录清理逻辑，但对 monitoring 做全局声明与安全处理
|
||||
# ========== 参数配置 ==========
|
||||
YT_STREAM_KEY = "x04z-564w-aks7-embw-30y4"
|
||||
youtube_rtmp = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}"
|
||||
|
||||
MAX_RETRY_DELAY = 90
|
||||
INITIAL_RETRY_DELAY = 3
|
||||
NO_FRAME_INTERVAL = 20
|
||||
NO_FRAME_TIMEOUT = 120
|
||||
START_CHECK_AFTER = 25.0
|
||||
STALE_OUTPUT_SECONDS = 30
|
||||
STDERR_QUEUE_TIMEOUT = 1.0
|
||||
END_KEYWORDS = ["live has ended", "主播已下播", "房间已关闭"]
|
||||
|
||||
# 构造标题
|
||||
timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime())
|
||||
_clean_title_in_name = sanitize_title(title_in_name)
|
||||
stream_title = sanitize_title(f"{anchor_name}{('_' + _clean_title_in_name) if _clean_title_in_name else ''}_{timestamp_str}")
|
||||
print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}")
|
||||
|
||||
# NVENC 检测
|
||||
codec_v, preset_v, tune_param = "libx264", "veryfast", ["-tune", "zerolatency"]
|
||||
try:
|
||||
if platform.system().lower() in ["windows", "linux"]:
|
||||
check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=3)
|
||||
if check.returncode == 0:
|
||||
enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=3)
|
||||
if "h264_nvenc" in (enc.stdout or ""):
|
||||
codec_v, preset_v, tune_param = "h264_nvenc", "p5", ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"]
|
||||
print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速")
|
||||
except Exception: pass
|
||||
|
||||
# FFmpeg 命令
|
||||
douyin_headers = (
|
||||
"User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15\r\n"
|
||||
"Referer: https://live.douyin.com/\r\nOrigin: https://live.douyin.com\r\n"
|
||||
)
|
||||
|
||||
ffmpeg_command = [
|
||||
"ffmpeg", "-rw_timeout", "10000000", "-reconnect", "1", "-reconnect_at_eof", "1",
|
||||
"-reconnect_streamed", "1", "-reconnect_delay_max", "15", "-fflags", "+genpts+discardcorrupt",
|
||||
"-err_detect", "ignore_err", "-thread_queue_size", "8192", "-headers", douyin_headers,
|
||||
"-i", real_url,
|
||||
"-vf", "scale=720:1280:force_original_aspect_ratio=decrease,pad=720:1280:(ow-iw)/2:(oh-ih)/2:black",
|
||||
"-c:v", codec_v, "-preset", preset_v, *tune_param,
|
||||
"-b:v", "2500k", "-maxrate", "2500k", "-bufsize", "5000k",
|
||||
"-g", "50", "-r", "25", "-pix_fmt", "yuv420p",
|
||||
"-c:a", "aac", "-b:a", "128k", "-ar", "44100", "-ac", "2", "-af", "aresample=async=1:first_pts=0",
|
||||
"-flags", "+global_header", "-f", "flv", youtube_rtmp
|
||||
]
|
||||
|
||||
proc, stderr_q, retry_delay = None, queue.Queue(), INITIAL_RETRY_DELAY
|
||||
|
||||
while True:
|
||||
try:
|
||||
recording.discard(record_name)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
recording_time_list.pop(record_name, None)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
if record_url in running_list:
|
||||
if retry_delay > INITIAL_RETRY_DELAY: print(f"[WARN] 防风控等待 {retry_delay}s 后重连...")
|
||||
time.sleep(retry_delay)
|
||||
cleanup_proc(proc)
|
||||
|
||||
print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}")
|
||||
proc = subprocess.Popen(ffmpeg_command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True)
|
||||
reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q))
|
||||
reader_t.daemon = True; reader_t.start()
|
||||
|
||||
start_time = last_frame_time = time.time()
|
||||
consecutive_failures = 0
|
||||
|
||||
while True:
|
||||
try:
|
||||
running_list.remove(record_url)
|
||||
except ValueError:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
try: line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT)
|
||||
except queue.Empty:
|
||||
if time.time() - last_frame_time > STALE_OUTPUT_SECONDS:
|
||||
print("[WARN] 长时间无 stderr 输出，重启进程")
|
||||
cleanup_proc(proc); break
|
||||
continue
|
||||
|
||||
# 确保 monitoring 在全局可用，安全减一
|
||||
try:
|
||||
global monitoring
|
||||
monitoring = max(0, monitoring - 1) if isinstance(monitoring, int) else 0
|
||||
except NameError:
|
||||
monitoring = 0
|
||||
if line is None:
|
||||
if proc.poll() is not None: break
|
||||
else: time.sleep(0.5); continue
|
||||
|
||||
try:
|
||||
clear_record_info(record_name, record_url)
|
||||
except Exception:
|
||||
pass
|
||||
line = line.strip(); proc.last_out_ts = time.time()
|
||||
if line.startswith("__ERR__READER__"):
|
||||
print(f"[ERROR] stderr reader 异常: {line}"); continue
|
||||
|
||||
try:
|
||||
color_obj.print_colored(f"[{record_name}] YouTube推流已彻底停止并清理完毕\n", color_obj.GREEN)
|
||||
except Exception:
|
||||
print(f"[{record_name}] YouTube推流已彻底停止并清理完毕")
|
||||
print(f"[FFmpeg] {line}")
|
||||
line_lower = line.lower()
|
||||
|
||||
# ======= 替换结束 =======
|
||||
if re.search(r'\b(401|403|404|410|429|502|503|504)\b', line_lower) or any(kw in line_lower for kw in END_KEYWORDS):
|
||||
print("[ERROR] 检测到下播或访问被拒，停止推流"); cleanup_proc(proc); raise StreamEnded()
|
||||
|
||||
except Exception as e:
|
||||
print(f"YouTube推流最外层异常: {e}")
|
||||
traceback.print_exc()
|
||||
# 修正这里的判断：如果 proc 存在且仍在运行才 kill
|
||||
try:
|
||||
if 'proc' in locals() and proc and proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
color_obj.print_colored(f"[{record_name}] YouTube推流异常退出但已清理\n", color_obj.RED)
|
||||
except Exception:
|
||||
print(f"[{record_name}] YouTube推流异常退出但已清理\n")
|
||||
# frame 更新 last_frame_time
|
||||
if time.time() - start_time > START_CHECK_AFTER and "frame=" in line_lower:
|
||||
last_frame_time = time.time()
|
||||
|
||||
if time.time() - last_frame_time > NO_FRAME_TIMEOUT:
|
||||
print("[WARN] 无帧超时，判定主播已下播"); cleanup_proc(proc); raise StreamEnded()
|
||||
|
||||
except StreamEnded: raise
|
||||
except Exception as inner_e:
|
||||
print(f"[ERROR] 内部异常: {inner_e}"); traceback.print_exc()
|
||||
cleanup_proc(proc)
|
||||
consecutive_failures += 1
|
||||
retry_delay = min(INITIAL_RETRY_DELAY * (2 ** consecutive_failures), MAX_RETRY_DELAY)
|
||||
break
|
||||
|
||||
except StreamEnded:
|
||||
print(f"[INFO] 主播已下播或明确拒绝 → {record_name} 推流停止")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"[ERROR] 外层异常: {e}"); traceback.print_exc()
|
||||
consecutive_failures += 1
|
||||
retry_delay = min(INITIAL_RETRY_DELAY * (2 ** consecutive_failures), MAX_RETRY_DELAY)
|
||||
continue
|
||||
finally:
|
||||
cleanup_proc(proc)
|
||||
|
||||
|
||||
count_time = time.time()
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user