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

's'

Browse Source
This commit is contained in:
eric
2025-12-02 09:16:17 +08:00
parent d12a35d249
commit 63bf62df84
2 changed files with 292 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
backup/7777 Normal file
View File

@@ -0,0 +1,203 @@
else:
# ====================== 抖音→YouTube 2025（终极精简版） ======================
import queue
import time
import datetime
import subprocess
import threading
import unicodedata
import platform
import re
import traceback
# --------------------- 配置区（可外部传入） ---------------------
YT_STREAM_KEY = "qxvb-r47b-r5ju-6ud3-6k7z"
YOUTUBE_RTMP = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}"
MAX_RETRY_DELAY = 90
INITIAL_RETRY_DELAY = 3
NO_FRAME_TIMEOUT = 120 # 超过多久无新帧视为下播
START_CHECK_AFTER = 25.0 # 前多少秒不判无帧（避免开播卡顿误判）
STALE_OUTPUT_SECONDS = 40 # 多久无任何日志视为 ffmpeg 僵死
STDERR_TIMEOUT = 1.0
END_KEYWORDS = [
"connection reset by peer", "failed to write trailer", "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 refused"
]
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"
)
# --------------------- 工具函数 ---------------------
def sanitize_title(text: str, max_len: int = 120) -> str:
if not text:
return ""
text = unicodedata.normalize("NFKC", text)
text = ''.join(c for c in text if unicodedata.category(c)[0] != 'C')
text = ''.join(c for c in text if ord(c) <= 0xFFFF)
return text.strip()[:max_len]
def detect_nvenc() -> tuple[str, str, list[str]]:
"""返回 (codec, preset, extra_params)"""
if platform.system().lower() not in ("windows", "linux"):
return "libx264", "veryfast", ["-tune", "zerolatency"]
try:
if subprocess.run(["nvidia-smi"], capture_output=True, timeout=5).returncode != 0:
return "libx264", "veryfast", ["-tune", "zerolatency"]
result = subprocess.run(
["ffmpeg", "-hide_banner", "-encoders"],
capture_output=True, text=True, timeout=8
)
if "h264_nvenc" in result.stdout:
print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速")
return "h264_nvenc", "p5", ["-tune", "ll", "-rc", "vbr", "-cq", "21", "-b:v", "2500k"]
except Exception as e:
print(f"[WARN] NVENC 检测失败，回落软件编码: {e}")
return "libx264", "veryfast", ["-tune", "zerolatency"]
def build_ffmpeg_command(real_url: str):
codec_v, preset_v, tune_params = detect_nvenc()
return [
"ffmpeg", "-y", "-loglevel", "info",
"-rw_timeout", "15000000",
"-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_params,
"-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
]
# --------------------- 标题生成 ---------------------
timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime())
clean_title = sanitize_title(title_in_name)
stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}")
print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}")
# --------------------- 主逻辑（单层循环 + 单一 finally） ---------------------
proc = None
reader_thread = None
should_exit = False # 替代 StreamEnded 异常
retry_delay = INITIAL_RETRY_DELAY
def cleanup_resources():
"""统一清理（只调用一次）"""
nonlocal proc, reader_thread
if proc and proc.poll() is None:
try: proc.terminate(); proc.wait(timeout=5)
except: proc.kill()
if reader_thread and reader_thread.is_alive():
reader_thread.join(timeout=1)
# 全局状态清理（只执行一次）
try:
recording.discard(record_name)
recording_time_list.pop(record_name, None)
if record_url in running_list:
running_list.remove(record_url)
global monitoring
monitoring = max(0, monitoring - 1)
clear_record_info(record_name, record_url)
except: pass
try:
ffmpeg_cmd = build_ffmpeg_command(real_url)
while not should_exit:
if retry_delay > INITIAL_RETRY_DELAY:
print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接...")
time.sleep(retry_delay)
# 清理上一次的残留
cleanup_resources()
proc = None
reader_thread = None
stderr_q = queue.Queue()
print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}")
proc = subprocess.Popen(
ffmpeg_cmd,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
universal_newlines=True
)
reader_thread = threading.Thread(
target=lambda p=proc, q=stderr_q: (
[q.put(l) for l in iter(p.stderr.readline, "")] or q.put(None)
)
)
reader_thread.daemon = True
reader_thread.start()
start_time = last_frame_time = last_log_time = time.time()
while True:
try:
line = stderr_q.get(timeout=STDERR_TIMEOUT)
except queue.Empty:
# 超时检查僵死
if time.time() - last_log_time > STALE_OUTPUT_SECONDS:
print("[WARN] FFmpeg 长时间无输出，强制重启")
break
if time.time() - start_time > START_CHECK_AFTER and time.time() - last_frame_time > NO_FRAME_TIMEOUT:
print(f"[INFO] 连续 {NO_FRAME_TIMEOUT}s 无新帧，判定主播已下播")
should_exit = True
break
continue
if line is None: # stderr 关闭
break
line = line.rstrip()
if line:
last_log_time = time.time()
print(f"[FFmpeg] {line}")
lower = line.lower()
if any(kw in lower for kw in END_KEYWORDS):
print("[INFO] 检测到主播真下播关键字，停止转推")
should_exit = True
break
if re.search(r'frame=\s*\d+', lower) or "fps=" in lower:
last_frame_time = time.time()
# 内层循环退出：判断是否真下播
if should_exit:
break
# 网络波动等情况 → 指数退避重试
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
# --------------------- 正常退出 ---------------------
color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)
except Exception as e:
print(f"[FATAL] YouTube 推流意外崩溃: {e}")
traceback.print_exc()
color_obj.print_colored(f"[{record_name}] YouTube 推流异常退出但已清理\n", color_obj.RED)
finally:
# 无论如何只清理一次
cleanup_resources()
return

135
main.py
View File

@@ -1529,7 +1529,9 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
error_window.append(1)
else:
# ====================== 抖音→YouTube 2025 终极永不死版（已修复所有已知致命问题） ======================
# ====================== 抖音→YouTube 2025（重构：单层重试 + 统一清理） ======================
import queue, unicodedata, platform, traceback
class StreamEnded(Exception):
@@ -1546,12 +1548,19 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
try:
while True:
line = proc.stderr.readline()
if not line: break
if not line:
break
q.put(line)
except Exception as e:
q.put(f"__ERR__READER__:{e}")
try:
q.put(f"__ERR__READER__:{e}")
except Exception:
pass
finally:
q.put(None)
try:
q.put(None)
except Exception:
pass
def cleanup_proc(proc):
try:
@@ -1561,29 +1570,24 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
proc.wait(timeout=5)
except Exception:
if proc.poll() is None:
proc.kill()
try:
proc.kill()
except Exception:
pass
except Exception:
pass
# ========== 参数配置 ==========
# YT_STREAM_KEY = "ue78-1c3e-mr9g-14mz-9r4z" # 没有ypp的美食
YT_STREAM_KEY = "qxvb-r47b-r5ju-6ud3-6k7z" # vlog频道
# ========== 参数配置（保留你原始配置） ==========
YT_STREAM_KEY = "qxvb-r47b-r5ju-6ud3-6k7z"
youtube_rtmp = f"rtmp://a.rtmp.youtube.com/live2/{YT_STREAM_KEY}"
# 内网SRS服务器
# YT_STREAM_KEY = "douyin" # vlog频道
# youtube_rtmp = f"rtmp://192.168.31.184/live/{YT_STREAM_KEY}"
MAX_RETRY_DELAY = 90
INITIAL_RETRY_DELAY = 3
NO_FRAME_TIMEOUT = 120 # 120秒无新帧才判真下播（完美契合YouTube 25~30秒容忍）
NO_FRAME_TIMEOUT = 120
START_CHECK_AFTER = 25.0
STALE_OUTPUT_SECONDS = 40 # 40秒无任何日志才重启（更宽容）
STALE_OUTPUT_SECONDS = 40
STDERR_QUEUE_TIMEOUT = 1.0
# 2025年12月实测最准的下播关键词（一个都不能少！）
END_KEYWORDS = [
"connection reset by peer",
"failed to write trailer",
@@ -1599,13 +1603,12 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
"connection timed out",
]
# 标题清洗
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 检测
# NVENC 检测（原样保留）
codec_v, preset_v, tune_param = "libx264", "veryfast", ["-tune", "zerolatency"]
try:
if platform.system().lower() in ["windows", "linux"]:
@@ -1618,14 +1621,13 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
except Exception as e:
print(f"[WARN] NVENC 检测失败，回退软件编码: {e}")
# 伪装 headers
# headers & ffmpeg command（保持原来参数）
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 命令（关键！bufsize=1）
ffmpeg_command = [
"ffmpeg", "-loglevel", "info",
"-rw_timeout", "15000000",
@@ -1646,8 +1648,10 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
"-f", "flv", youtube_rtmp
]
# ---- 单层重试循环（清晰可控） ----
proc = None
stderr_q = queue.Queue()
stderr_q = None
reader_t = None
retry_delay = INITIAL_RETRY_DELAY
try:
@@ -1657,7 +1661,15 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...")
time.sleep(retry_delay)
cleanup_proc(proc)
# 每次重连前先清理上一次的进程/线程
try:
cleanup_proc(proc)
except Exception:
pass
if stderr_q:
# 垃圾化旧队列以便 GC
stderr_q = None
stderr_q = queue.Queue()
print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}")
proc = subprocess.Popen(
@@ -1666,33 +1678,37 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
bufsize=1, # 关键！必须是 1
bufsize=1,
universal_newlines=True
)
proc.last_out_ts = time.time() # 给 proc 加个属性记录最后输出时间
proc.last_out_ts = time.time()
# 启动 stderr reader
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()
# ---- 监控循环（单一职责：处理 stderr 日志并决定是否结束/重连） ----
while True:
try:
try:
line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT)
except queue.Empty:
# 超时检查僵死
if time.time() - proc.last_out_ts > STALE_OUTPUT_SECONDS:
# 超时检查：如果太久没有任何日志，判为僵死，重启 proc（跳出内层，进行重试）
if time.time() - getattr(proc, "last_out_ts", time.time()) > STALE_OUTPUT_SECONDS:
print("[WARN] 长时间无任何日志，强制重启进程")
cleanup_proc(proc)
break
continue
if line is None: # reader 结束
if line is None:
# reader 线程结束；如果 proc 已退出，跳出去重试
if proc.poll() is not None:
break
continue
else:
continue
line = line.strip()
if line.startswith("__ERR__READER__"):
@@ -1700,22 +1716,21 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
continue
if line:
proc.last_out_ts = time.time() # 任意日志都更新时间
proc.last_out_ts = time.time()
print(f"[FFmpeg] {line}")
line_lower = line.lower()
# 真下播精准检测
# 检测到明确下播关键字 -> 认为主播真下播，立即退出（不再重试）
if any(kw in line_lower for kw in END_KEYWORDS):
print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流")
cleanup_proc(proc)
raise StreamEnded()
# 帧检测（只更新 last_frame_time）
# 帧检测：更新 frame 时间戳
if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower:
last_frame_time = time.time()
# 120秒无新帧 = 真下播（完美契合 YouTube 25~30秒容忍）
# 超时无新帧 => 认为下播
if time.time() - start_time > START_CHECK_AFTER:
if time.time() - last_frame_time > NO_FRAME_TIMEOUT:
print(f"[WARN] 连续 {NO_FRAME_TIMEOUT}s 无新帧，判定主播已下播")
@@ -1723,43 +1738,71 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
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
break # 跳出监控循环 -> 外层 while 会继续并重试
# 如果监控循环走到这里并不是因为 StreamEnded（例如 break），则重试连接（loop continue）
# 将在 while True 的下一次循环中进行等待并重连
continue
except StreamEnded:
# 真实的主播下播，停止并彻底退出本录制线程
print(f"[INFO] 主播已下播，停止 YouTube 转推 → {record_name}")
break
except Exception as e:
# 外层致命异常：打印并退出循环（会在 finally 里做统一清理）
print(f"[ERROR] 外层异常: {e}")
traceback.print_exc()
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
# 等待并尝试重连（除非你希望直接退出，这里保留重试策略）
continue
finally:
cleanup_proc(proc)
# ============ 彻底清理 ============
# ============ 正常或 StreamEnded 后的统一清理（在这里 break 到这里） ============
cleanup_proc(proc)
recording.discard(record_name)
recording_time_list.pop(record_name, None)
if record_url in running_list:
running_list.remove(record_url)
global monitoring
monitoring = max(0, monitoring - 1)
clear_record_info(record_name, record_url)
# 以下清理放在 try/except 里以防止清理过程出错导致漏逻辑
try:
recording.discard(record_name)
recording_time_list.pop(record_name, None)
if record_url in running_list:
running_list.remove(record_url)
global monitoring
monitoring = max(0, monitoring - 1)
clear_record_info(record_name, record_url)
except Exception:
pass
color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)
return
except Exception as e:
# 最外层异常也要统一清理（保证不会留下僵尸状态）
print(f"[FATAL] YouTube 推流最外层异常: {e}")
traceback.print_exc()
cleanup_proc(proc)
try:
cleanup_proc(proc)
except Exception:
pass
try:
recording.discard(record_name)
recording_time_list.pop(record_name, None)
if record_url in running_list:
running_list.remove(record_url)
# global monitoring
monitoring = max(0, monitoring - 1)
clear_record_info(record_name, record_url)
except Exception:
pass
color_obj.print_colored(f"[{record_name}] YouTube 推流异常退出但已清理\n", color_obj.RED)
return
count_time = time.time()