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-01 18:56:27 +08:00
parent 350eb6d798
commit 3b7785ec30
1 changed files with 153 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
main.py
View File

@@ -1529,10 +1529,8 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
error_window.append(1)
else:
# ====================== 终极永不死 YouTube 转推增强版 ======================
import queue,traceback, unicodedata, platform
# ====================== 抖音→YouTube 2025 终极永不死版（已修复所有已知致命问题） ======================
import queue, unicodedata, platform, traceback
class StreamEnded(Exception):
pass
@@ -1546,9 +1544,10 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
def _stderr_reader_thread(proc, q):
try:
for raw in iter(proc.stderr.readline, ''):
if not raw: break
q.put(raw)
while True:
line = proc.stderr.readline()
if not line: break
q.put(line)
except Exception as e:
q.put(f"__ERR__READER__:{e}")
finally:
@@ -1557,26 +1556,43 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
def cleanup_proc(proc):
try:
if proc and proc.poll() is None:
try: proc.terminate(); proc.wait(timeout=5)
try:
proc.terminate()
proc.wait(timeout=5)
except Exception:
if proc.poll() is None: proc.kill()
if proc.poll() is None:
proc.kill()
except Exception:
pass
# ========== 参数配置 ==========
YT_STREAM_KEY = "x04z-564w-aks7-embw-30y4"
YT_STREAM_KEY = "ue78-1c3e-mr9g-14mz-9r4z" # ← 改成你自己的
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
NO_FRAME_TIMEOUT = 120 # 120秒无新帧才判真下播（完美契合YouTube 25~30秒容忍）
START_CHECK_AFTER = 25.0
STALE_OUTPUT_SECONDS = 30
STALE_OUTPUT_SECONDS = 40 # 40秒无任何日志才重启（更宽容）
STDERR_QUEUE_TIMEOUT = 1.0
END_KEYWORDS = ["live has ended", "主播已下播", "房间已关闭"]
# 构造标题
# 2025年12月实测最准的下播关键词（一个都不能少！）
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",
]
# 标题清洗
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}")
@@ -1586,100 +1602,158 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
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)
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=3)
enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5)
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
except Exception as e:
print(f"[WARN] NVENC 检测失败，回退软件编码: {e}")
# FFmpeg 命令
# 伪装 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\nOrigin: https://live.douyin.com\r\n"
"Referer: https://live.douyin.com/\r\n"
"Origin: https://live.douyin.com\r\n"
)
# FFmpeg 命令（关键！bufsize=1）
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,
"ffmpeg", "-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_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
"-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
proc = None
stderr_q = queue.Queue()
retry_delay = INITIAL_RETRY_DELAY
while True:
try:
if retry_delay > INITIAL_RETRY_DELAY: print(f"[WARN] 防风控等待 {retry_delay}s 后重连...")
time.sleep(retry_delay)
cleanup_proc(proc)
try:
while True:
try:
if retry_delay > INITIAL_RETRY_DELAY:
print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...")
time.sleep(retry_delay)
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()
cleanup_proc(proc)
start_time = last_frame_time = time.time()
consecutive_failures = 0
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, # 关键！必须是 1
universal_newlines=True
)
proc.last_out_ts = time.time() # 给 proc 加个属性记录最后输出时间
while True:
try:
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
reader_t = threading.Thread(target=_stderr_reader_thread, args=(proc, stderr_q))
reader_t.daemon = True
reader_t.start()
if line is None:
if proc.poll() is not None: break
else: time.sleep(0.5); continue
start_time = last_frame_time = time.time()
line = line.strip(); proc.last_out_ts = time.time()
if line.startswith("__ERR__READER__"):
print(f"[ERROR] stderr reader 异常: {line}"); continue
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:
print("[WARN] 长时间无任何日志，强制重启进程")
cleanup_proc(proc)
break
continue
print(f"[FFmpeg] {line}")
line_lower = line.lower()
if line is None: # reader 结束
if proc.poll() is not None:
break
continue
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()
line = line.strip()
if line.startswith("__ERR__READER__"):
print(f"[ERROR] stderr reader 异常: {line}")
continue
# frame 更新 last_frame_time
if time.time() - start_time > START_CHECK_AFTER and "frame=" in line_lower:
last_frame_time = time.time()
if line:
proc.last_out_ts = time.time() # 任意日志都更新时间
print(f"[FFmpeg] {line}")
if time.time() - last_frame_time > NO_FRAME_TIMEOUT:
print("[WARN] 无帧超时，判定主播已下播"); cleanup_proc(proc); raise StreamEnded()
line_lower = line.lower()
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
# 真下播精准检测
if any(kw in line_lower for kw in END_KEYWORDS):
print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流")
cleanup_proc(proc)
raise StreamEnded()
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)
# 帧检测（只更新 last_frame_time）
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 无新帧，判定主播已下播")
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
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
finally:
cleanup_proc(proc)
# ============ 彻底清理 ============
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)
color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)
except Exception as e:
print(f"[FATAL] YouTube 推流最外层异常: {e}")
traceback.print_exc()
cleanup_proc(proc)
color_obj.print_colored(f"[{record_name}] YouTube 推流异常退出但已清理\n", color_obj.RED)
count_time = time.time()