eric/gitlab-instance-0a899031_douyinyoutube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

''

Browse Source
This commit is contained in:
eric
2026-01-27 03:21:13 -06:00
parent 7ade49e46d
commit 187f7a2ce8
1 changed files with 146 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
douyin_youtube_ffplay.py
View File

@@ -11,6 +11,7 @@ import re
import os import os
import configparser import configparser
class StreamEnded(Exception): class StreamEnded(Exception):
"""主播真实下播的专用异常""" """主播真实下播的专用异常"""
pass pass
@@ -33,40 +34,31 @@ def _stderr_reader_thread(proc, q):
break break
q.put(line) q.put(line)
except Exception as e: except Exception as e:
q.put(f"__ERR__READER__:{e}")
finally:
q.put(None)
def _cleanup_proc(proc):
if proc and proc.poll() is None:
try: try:
proc.terminate() q.put(f"__ERR__READER__:{e}")
proc.wait(timeout=5) except Exception:
except: pass
try: finally:
proc.kill() try:
except: q.put(None)
pass except Exception:
pass
def _detect_nvenc(): def cleanup_proc(proc):
try: try:
if platform.system().lower() not in ["windows", "linux"]: if proc and proc.poll() is None:
return "libx264", "veryfast", ["-tune", "zerolatency"] try:
proc.terminate()
check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=5) proc.wait(timeout=5)
if check.returncode != 0: except Exception:
return "libx264", "veryfast", ["-tune", "zerolatency"] if proc.poll() is None:
try:
enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5) proc.kill()
if "h264_nvenc" in (enc.stdout or ""): except Exception:
print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") pass
return "h264_nvenc", "p5", ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "5200k"] except Exception:
return "libx264", "veryfast", ["-tune", "zerolatency"] pass
except Exception as e:
print(f"[WARN] NVENC 检测失败，回退软件编码: {e}")
return "libx264", "veryfast", ["-tune", "zerolatency"]
def start_douyin_youtube_ffplay( def start_douyin_youtube_ffplay(
@@ -78,18 +70,20 @@ def start_douyin_youtube_ffplay(
recording: set, recording: set,
recording_time_list: dict, recording_time_list: dict,
running_list: list, running_list: list,
monitoring: list, # 可变列表，如 [count]，用于并发计数 monitoring: list,
clear_record_info, clear_record_info,
color_obj, color_obj,
): ):
""" """
抖音 → YouTube 多路推流主函数（支持 config/youtube.ini 中只配置 stream key） 抖音 → YouTube 推流主函数（单路推流）
只从 config/youtube.ini 读取一个 stream key（取第一个有效值）
""" """
# ====================== 读取 YouTube stream key 配置 ====================== # ====================== 读取 YouTube stream key 配置 ======================
config_file = "config/youtube.ini" config_file = "config/youtube.ini"
rtmp_base = "rtmp://a.rtmp.youtube.com/live2/" rtmp_base = "rtmp://a.rtmp.youtube.com/live2/"
default_key = "qxvb-r47b-r5ju-6ud3-6k7z" default_key = "qxvb-r47b-r5ju-6ud3-6k7z" # 备用默认 key
rtmp_targets = [] youtube_rtmp = rtmp_base + default_key # 默认值
if os.path.exists(config_file): if os.path.exists(config_file):
config = configparser.ConfigParser() config = configparser.ConfigParser()
@@ -97,21 +91,13 @@ def start_douyin_youtube_ffplay(
if config.has_section("youtube"): if config.has_section("youtube"):
for key_name, value in config.items("youtube"): for key_name, value in config.items("youtube"):
value = value.strip() value = value.strip()
if not value: if value:
continue if value.startswith("rtmp://"):
if value.startswith("rtmp://"): youtube_rtmp = value
rtmp_targets.append(value) else:
else: youtube_rtmp = rtmp_base + value
rtmp_targets.append(rtmp_base + value) print(f"[INFO] 从 {config_file} 读取 YouTube stream key: {youtube_rtmp}")
break # 只取第一个有效 key，保持单路推流（不加多路逻辑）
if not rtmp_targets:
rtmp_targets = [rtmp_base + default_key]
print(f"[WARN] 未在 {config_file} 中找到有效配置，使用默认 stream key 推流")
num_targets = len(rtmp_targets)
print(f"[INFO] 加载了 {num_targets} 个 YouTube 推流目标")
tee_outputs = "|".join(f"[f=flv:flvflags=no_duration_filesize]{url}" for url in rtmp_targets)
# ====================== 常量配置 ====================== # ====================== 常量配置 ======================
MAX_RETRY_DELAY = 90 MAX_RETRY_DELAY = 90
@@ -122,9 +108,17 @@ def start_douyin_youtube_ffplay(
STALE_OUTPUT_SECONDS = 40 STALE_OUTPUT_SECONDS = 40
END_KEYWORDS = [ END_KEYWORDS = [
"connection reset by peer", "failed to write trailer", "invalid data found", "connection reset by peer",
"server returned 403", "server returned 404", "ingestion error", "access denied", "failed to write trailer",
"packet too short", "client disconnected", "live has ended", "no route to host", "invalid data found",
"server returned 403",
"server returned 404",
"ingestion error",
"access denied",
"packet too short",
"client disconnected",
"live has ended",
"no route to host",
"connection timed out", "connection timed out",
] ]
@@ -138,12 +132,26 @@ def start_douyin_youtube_ffplay(
timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime())
clean_title = sanitize_title(title) clean_title = sanitize_title(title)
stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}") stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}")
print(f"[INFO] {rec_info} 开始推流到 YouTube ({num_targets} 路): {stream_title}") print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}")
# ====================== 编码器检测 ====================== # ====================== NVENC 检测 ======================
codec_v, preset_v, extra_tune_params = _detect_nvenc() 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=5)
if check.returncode == 0:
enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5)
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("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速")
except Exception as e:
print(f"[WARN] NVENC 检测失败，回退软件编码: {e}")
# ====================== FFmpeg 命令（1080p + tee 多路） ====================== # ====================== FFmpeg 命令（1080p 单路推流，原样保留） ======================
ffmpeg_command = [ ffmpeg_command = [
"ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-re", "ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-re",
"-stats", "-stats_period", "1", "-stats", "-stats_period", "1",
@@ -164,7 +172,7 @@ def start_douyin_youtube_ffplay(
"-c:v", codec_v, "-c:v", codec_v,
"-preset", preset_v, "-preset", preset_v,
*extra_tune_params, *tune_param,
"-profile:v", "high", "-profile:v", "high",
"-b:v", "5200k", "-maxrate", "5800k", "-bufsize", "10400k", "-b:v", "5200k", "-maxrate", "5800k", "-bufsize", "10400k",
@@ -180,109 +188,119 @@ def start_douyin_youtube_ffplay(
"-af", "aresample=async=1:first_pts=0", "-af", "aresample=async=1:first_pts=0",
"-flags", "+global_header", "-flags", "+global_header",
"-f", "tee", tee_outputs "-flvflags", "no_duration_filesize",
"-f", "flv", youtube_rtmp
] ]
# ====================== 主循环 ====================== # ====================== 主重试循环（原样保留） ======================
proc = None proc = None
stderr_q = queue.Queue() stderr_q = None
reader_t = None reader_t = None
retry_delay = INITIAL_RETRY_DELAY retry_delay = INITIAL_RETRY_DELAY
try: try:
while True: while True:
if retry_delay > INITIAL_RETRY_DELAY: try:
print(f"[WARN] 防风控等待 {retry_delay}s 后重连...") if retry_delay > INITIAL_RETRY_DELAY:
time.sleep(retry_delay) print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...")
time.sleep(retry_delay)
_cleanup_proc(proc) cleanup_proc(proc)
stderr_q = queue.Queue() stderr_q = queue.Queue()
print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}") print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}")
proc = subprocess.Popen( proc = subprocess.Popen(
ffmpeg_command, ffmpeg_command,
stdin=subprocess.DEVNULL, stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, text=True,
bufsize=1, bufsize=1,
universal_newlines=True universal_newlines=True
) )
proc.last_out_ts = time.time() proc.last_out_ts = time.time()
reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q)) reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q))
reader_t.daemon = True reader_t.daemon = True
reader_t.start() reader_t.start()
start_time = last_frame_time = time.time() start_time = last_frame_time = time.time()
while True: while True:
try:
try: try:
line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT) try:
except queue.Empty: line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT)
if time.time() - proc.last_out_ts > STALE_OUTPUT_SECONDS: except queue.Empty:
print("[WARN] 长时间无日志，强制重启 ffmpeg 进程") if time.time() - getattr(proc, "last_out_ts", time.time()) > STALE_OUTPUT_SECONDS:
_cleanup_proc(proc) print("[WARN] 长时间无任何日志，强制重启进程")
break cleanup_proc(proc)
continue break
continue
if line is None: if line is None:
if proc.poll() is not None: if proc.poll() is not None:
break break
continue continue
line = line.strip() line = line.strip()
if line.startswith("__ERR__READER__"): if line.startswith("__ERR__READER__"):
print(f"[ERROR] stderr reader 异常: {line}") print(f"[ERROR] stderr reader 异常: {line}")
continue continue
if line: if line:
proc.last_out_ts = time.time() proc.last_out_ts = time.time()
print(f"[FFmpeg] {line}") print(f"[FFmpeg] {line}")
line_lower = line.lower() line_lower = line.lower()
if any(kw in line_lower for kw in END_KEYWORDS): if any(kw in line_lower for kw in END_KEYWORDS):
print("[ERROR] 检测到主播真实下播或被拒，停止推流") print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流")
raise StreamEnded() cleanup_proc(proc)
if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower:
last_frame_time = time.time()
if time.time() - start_time > START_CHECK_AFTER:
if time.time() - last_frame_time > NO_FRAME_TIMEOUT:
print(f"[WARN] 超过 {NO_FRAME_TIMEOUT}s 无新帧，判定主播下播")
raise StreamEnded() raise StreamEnded()
except StreamEnded: if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower:
raise last_frame_time = time.time()
except Exception as e:
print(f"[ERROR] 监控循环异常: {e}")
traceback.print_exc()
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
break
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) if time.time() - start_time > START_CHECK_AFTER:
continue if time.time() - last_frame_time > NO_FRAME_TIMEOUT:
print(f"[WARN] 连续 {NO_FRAME_TIMEOUT}s 无新帧，判定主播已下播")
cleanup_proc(proc)
raise StreamEnded()
except StreamEnded:
raise
except Exception as e:
print(f"[ERROR] 内层异常: {e}")
traceback.print_exc()
cleanup_proc(proc)
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
break
continue
except StreamEnded:
print(f"[INFO] 主播已下播，停止 YouTube 转推 → {record_name}")
break
except Exception as e:
print(f"[ERROR] 外层异常: {e}")
traceback.print_exc()
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
continue
except StreamEnded:
print(f"[INFO] 主播已下播，停止 YouTube 转推 → {record_name}")
except KeyboardInterrupt:
print("[INFO] 用户中断，停止推流")
except Exception as e: except Exception as e:
print(f"[FATAL] 推流异常: {e}") print(f"[FATAL] YouTube 推流最外层异常: {e}")
traceback.print_exc() traceback.print_exc()
finally: finally:
_cleanup_proc(proc) cleanup_proc(proc)
try: try:
recording.discard(record_name) recording.discard(record_name)
recording_time_list.pop(record_name, None) recording_time_list.pop(record_name, None)
if real_url in running_list: if real_url in running_list:
running_list.remove(real_url) running_list.remove(real_url)
monitoring[0] = max(0, monitoring[0] - 1) # 与 douyin_srs_ffplay.py 保持一致 monitoring[0] = max(0, monitoring[0] - 1)
clear_record_info(record_name, real_url) clear_record_info(record_name, real_url)
except Exception: except Exception:
pass pass
color_obj.print_colored(f"[{record_name}] YouTube 推流已停止并清理完毕\n", color_obj.GREEN) color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)