646 lines
26 KiB
Python
646 lines
26 KiB
Python
# douyin_youtube_ffplay.py
|
||
import queue
|
||
import unicodedata
|
||
import platform
|
||
import traceback
|
||
import time
|
||
import datetime
|
||
import subprocess
|
||
import threading
|
||
import re
|
||
import os
|
||
import json
|
||
import configparser
|
||
from pathlib import Path
|
||
|
||
from web2_proxy_config import merge_child_env
|
||
from web2_push_proxy import (
|
||
apply_push_proxy_to_ffmpeg_cmd,
|
||
ffmpeg_env_with_push_proxy,
|
||
normalize_push_proxy_mode,
|
||
redact_proxy_url,
|
||
resolve_youtube_push_proxy,
|
||
)
|
||
from web2_youtube_encoder_cache import get_cached_encoder, store_cached_encoder
|
||
from web2_youtube_runtime import resolve_live_video_encoder
|
||
from web2_youtube_relay_heartbeat import clear_relay_heartbeat, write_relay_heartbeat
|
||
from web2_youtube_relay_slot import release_relay_slot
|
||
|
||
|
||
class StreamEnded(Exception):
|
||
"""主播真实下播的专用异常"""
|
||
pass
|
||
|
||
|
||
class SourceStalled(Exception):
|
||
"""源流卡死/无进度，交回监控循环或切换备用源"""
|
||
pass
|
||
|
||
|
||
def _ffmpeg_speed_value(line: str) -> float | None:
|
||
match = re.search(r"speed\s*=\s*([0-9.]+)x", line, re.IGNORECASE)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return float(match.group(1))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _ffmpeg_progress_values(line: str) -> tuple[int | None, float | None]:
|
||
frame = None
|
||
media_time = None
|
||
frame_match = re.search(r"frame\s*=\s*(\d+)", line, re.IGNORECASE)
|
||
if frame_match:
|
||
frame = int(frame_match.group(1))
|
||
time_match = re.search(r"time\s*=\s*(\d{2}):(\d{2}):(\d{2}(?:\.\d+)?)", line, re.IGNORECASE)
|
||
if time_match:
|
||
h, m, s = time_match.groups()
|
||
media_time = int(h) * 3600 + int(m) * 60 + float(s)
|
||
return frame, media_time
|
||
|
||
|
||
def _looks_like_invalid_ts_drop(line: str) -> bool:
|
||
return "invalid dropping" in line or "non monotonically increasing dts" in line
|
||
|
||
|
||
def sanitize_title(text: str, max_len: int = 120) -> str:
|
||
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)
|
||
return text.strip()[:max_len]
|
||
|
||
|
||
def _stderr_reader_thread(proc, q):
|
||
try:
|
||
while True:
|
||
line = proc.stderr.readline()
|
||
if not line:
|
||
break
|
||
q.put(line)
|
||
except Exception as e:
|
||
try:
|
||
q.put(f"__ERR__READER__:{e}")
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
try:
|
||
q.put(None)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def cleanup_proc(proc):
|
||
try:
|
||
if proc and proc.poll() is None:
|
||
try:
|
||
proc.terminate()
|
||
proc.wait(timeout=5)
|
||
except Exception:
|
||
if proc.poll() is None:
|
||
try:
|
||
proc.kill()
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def start_douyin_youtube_ffplay(
|
||
title: str,
|
||
anchor_name: str,
|
||
real_url: str,
|
||
rec_info: str,
|
||
record_name: str,
|
||
recording: set,
|
||
recording_time_list: dict,
|
||
running_list: list,
|
||
monitoring: list,
|
||
clear_record_info,
|
||
color_obj,
|
||
*,
|
||
record_url: str = "",
|
||
config_file: str = "",
|
||
fallback_urls: list[str] | None = None,
|
||
):
|
||
"""
|
||
抖音 → YouTube 推流主函数（单路推流）
|
||
只从 config/youtube.ini 读取一个 stream key（取第一个有效值）
|
||
"""
|
||
|
||
# ====================== 读取 YouTube 配置 ======================
|
||
config_file = (config_file or os.environ.get("YOUTUBE_INI_PATH") or "config/youtube.ini").strip()
|
||
rtmps_base = "rtmps://a.rtmp.youtube.com/live2/"
|
||
rtmp_base = "rtmp://a.rtmp.youtube.com/live2/"
|
||
use_rtmps = True
|
||
# 720p 竖屏 YouTube 建议 4–6 Mbps；1080p 竖屏适当提高码率
|
||
b_v, maxrate, bufsize = "4000k", "4500k", "8000k"
|
||
stream_key = None
|
||
fast_audio = False # True=轻量音频链（降噪关）
|
||
output_resolution = "720p"
|
||
force_software_video = False
|
||
audio_denoise_on = False
|
||
has_audio_denoise_key = False
|
||
|
||
if os.path.exists(config_file):
|
||
config = configparser.ConfigParser()
|
||
config.read(config_file, encoding="utf-8")
|
||
if config.has_section("youtube"):
|
||
for k, v in config.items("youtube"):
|
||
v = (v or "").strip()
|
||
if k == "rtmps" and v.lower() in ("0", "false", "否"):
|
||
use_rtmps = False
|
||
elif k == "output_resolution" and v:
|
||
output_resolution = v.lower()
|
||
elif k == "use_gpu" and v:
|
||
force_software_video = v.lower() in ("否", "0", "false", "无", "关", "off")
|
||
elif k == "audio_denoise":
|
||
has_audio_denoise_key = True
|
||
audio_denoise_on = v.lower() in ("1", "true", "yes", "是", "开", "on")
|
||
elif k == "fast_audio" and v.lower() in ("1", "true", "yes", "是", "开", "on"):
|
||
fast_audio = True
|
||
elif k == "bitrate" and v:
|
||
try:
|
||
n = int(v)
|
||
b_v, maxrate, bufsize = f"{n}k", f"{min(n * 12 // 10, 8000)}k", f"{n * 2}k"
|
||
except ValueError:
|
||
pass
|
||
elif (k == "key" or k == "stream_key") and v and not v.startswith("#"):
|
||
stream_key = v
|
||
if has_audio_denoise_key:
|
||
fast_audio = not audio_denoise_on
|
||
|
||
if "1080" in output_resolution:
|
||
b_v, maxrate, bufsize = "6500k", "7800k", "13000k"
|
||
if stream_key:
|
||
if stream_key.startswith("rtmp://") or stream_key.startswith("rtmps://"):
|
||
youtube_rtmp = stream_key
|
||
else:
|
||
youtube_rtmp = (rtmps_base if use_rtmps else rtmp_base) + stream_key
|
||
print(f"[INFO] 从 {config_file} 读取 YouTube 推流地址")
|
||
else:
|
||
raise ValueError(f"请在 {config_file} 中配置 key=你的stream_key")
|
||
|
||
# ====================== 常量配置 ======================
|
||
_script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
_relay_pm2 = (
|
||
os.environ.get("LIVE_PM2_PROCESS_NAME") or os.environ.get("PM2_SERVICE_NAME") or "youtube"
|
||
).strip() or "youtube"
|
||
|
||
def _emit_relay_heartbeat(state: str, **extra: object) -> None:
|
||
write_relay_heartbeat(
|
||
_relay_pm2,
|
||
{
|
||
"state": state,
|
||
"stream_title": stream_title,
|
||
"source_url": real_url,
|
||
"codec": codec_v,
|
||
"record_name": record_name,
|
||
"anchor_name": anchor_name,
|
||
**extra,
|
||
},
|
||
base_dir=_script_dir,
|
||
)
|
||
|
||
MAX_RETRY_DELAY = 90
|
||
INITIAL_RETRY_DELAY = 3
|
||
STDERR_QUEUE_TIMEOUT = 1.0
|
||
NO_FRAME_TIMEOUT = 120
|
||
START_CHECK_AFTER = 25
|
||
STALE_OUTPUT_SECONDS = 40
|
||
RECONNECT_DELAY_MAX = 10
|
||
FAST_SPEED_THRESHOLD = 1.08
|
||
FAST_SPEED_STRIKES_NEEDED = 3
|
||
INVALID_DROP_WINDOW_SECONDS = 30
|
||
INVALID_DROP_LIMIT = 80
|
||
|
||
TERMINAL_KEYWORDS = [
|
||
"live has ended",
|
||
"server returned 404",
|
||
"ingestion error",
|
||
"access denied",
|
||
"server returned 403",
|
||
]
|
||
TRANSIENT_KEYWORDS = [
|
||
"connection reset by peer",
|
||
"failed to write trailer",
|
||
"invalid data found",
|
||
"server returned 500",
|
||
"packet too short",
|
||
"no route to host",
|
||
"connection timed out",
|
||
"connection refused",
|
||
"error writing output",
|
||
"broken pipe",
|
||
"i/o error",
|
||
"client disconnected",
|
||
"error opening output file",
|
||
"error opening output files",
|
||
"error opening output",
|
||
]
|
||
|
||
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"
|
||
)
|
||
|
||
# ====================== 标题处理 ======================
|
||
timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime())
|
||
clean_title = sanitize_title(title)
|
||
stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}")
|
||
print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}")
|
||
try:
|
||
print(
|
||
"[RELAY_META] "
|
||
+ json.dumps(
|
||
{
|
||
"ts": int(time.time()),
|
||
"anchor": anchor_name,
|
||
"douyinHint": (real_url or record_name or "")[:400],
|
||
"streamTitle": stream_title,
|
||
"youtubeKeySuffix": (stream_key[-8:] if stream_key and len(stream_key) > 8 else stream_key),
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
flush=True,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
from web2_relay_log import on_relay_push_start, relay_db_path
|
||
|
||
ini_text = ""
|
||
if os.path.exists(config_file):
|
||
with open(config_file, encoding="utf-8-sig") as ini_fp:
|
||
ini_text = ini_fp.read()
|
||
on_relay_push_start(
|
||
process=_relay_pm2,
|
||
anchor_name=anchor_name,
|
||
record_url=record_url or record_name,
|
||
stream_title=stream_title,
|
||
real_url=real_url,
|
||
youtube_ini_text=ini_text,
|
||
db_path=relay_db_path(Path(_script_dir)),
|
||
base_dir=Path(_script_dir),
|
||
)
|
||
except Exception as relay_log_exc:
|
||
print(f"[WARN] relay session log start skipped: {relay_log_exc}")
|
||
|
||
source_candidates: list[str] = []
|
||
_seen_source: set[str] = set()
|
||
for item in [real_url, *(fallback_urls or [])]:
|
||
u = (item or "").strip()
|
||
if not u or u in _seen_source:
|
||
continue
|
||
if not u.startswith("http"):
|
||
continue
|
||
_seen_source.add(u)
|
||
source_candidates.append(u)
|
||
if not source_candidates:
|
||
source_candidates = [real_url]
|
||
source_index = 0
|
||
|
||
push_proxy_mode = normalize_push_proxy_mode()
|
||
push_proxy = resolve_youtube_push_proxy() if push_proxy_mode != "off" else ""
|
||
if push_proxy:
|
||
print(f"[INFO] YouTube push proxy ({push_proxy_mode}): {redact_proxy_url(push_proxy)}")
|
||
|
||
if "1080" in output_resolution:
|
||
_sw, _sh = "1080", "1920"
|
||
else:
|
||
_sw, _sh = "720", "1280"
|
||
scale_vf = (
|
||
f"fps=30,scale={_sw}:{_sh}:force_original_aspect_ratio=decrease:flags=bilinear,"
|
||
f"pad={_sw}:{_sh}:(ow-iw)/2:(oh-ih)/2:black"
|
||
)
|
||
|
||
# ====================== NVENC / 编码器选择 ======================
|
||
codec_v = "libx264"
|
||
video_args = ["-preset", "ultrafast", "-tune", "zerolatency", "-profile:v", "high", "-level", "4.0",
|
||
"-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize]
|
||
_machine = platform.machine().lower()
|
||
_want_enc = (os.environ.get("LIVE_VIDEO_ENCODER") or "").strip()
|
||
try:
|
||
if platform.system().lower() == "linux":
|
||
_cached_enc = get_cached_encoder(_machine, "ffmpeg", base_dir=_script_dir)
|
||
if _cached_enc:
|
||
codec_v = _cached_enc
|
||
if _cached_enc == "h264_nvenc":
|
||
video_args = ["-profile:v", "high", "-level", "4.0",
|
||
"-tune", "ll", "-rc", "cbr", "-b:v", b_v,
|
||
"-maxrate", maxrate, "-bufsize", bufsize]
|
||
print(f"[INFO] 使用编码器缓存: {codec_v}")
|
||
elif not force_software_video:
|
||
enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=6)
|
||
blob = (enc.stdout or "") + (enc.stderr or "")
|
||
selected, note = resolve_live_video_encoder(_want_enc, blob, _machine)
|
||
if note:
|
||
print(f"[INFO] {note}")
|
||
if selected:
|
||
codec_v = selected
|
||
if selected == "h264_nvenc":
|
||
video_args = ["-profile:v", "high", "-level", "4.0",
|
||
"-tune", "ll", "-rc", "cbr", "-b:v", b_v,
|
||
"-maxrate", maxrate, "-bufsize", bufsize]
|
||
store_cached_encoder(_machine, "ffmpeg", codec_v, base_dir=_script_dir)
|
||
elif (
|
||
not force_software_video
|
||
and 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"
|
||
video_args = ["-profile:v", "high", "-level", "4.0",
|
||
"-tune", "ll", "-rc", "cbr", "-b:v", b_v,
|
||
"-maxrate", maxrate, "-bufsize", bufsize]
|
||
print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速")
|
||
elif force_software_video:
|
||
print("[INFO] use_gpu=否/无：强制软件编码 libx264")
|
||
except Exception as e:
|
||
print(f"[WARN] 编码器检测失败，回退软件编码: {e}")
|
||
|
||
# ====================== 开源音频处理：背景音乐弱化 + 防版权 ======================
|
||
_ffmpeg_cwd = None
|
||
if fast_audio:
|
||
print("[INFO] fast_audio=开：轻量音频链（无 arnndn/长 EQ），减轻 CPU 以利 speed≥1.0")
|
||
_af_chain = "afftdn=nf=-26:tn=1,aresample=async=1:first_pts=0"
|
||
else:
|
||
# 优先 RNNoise(arnndn)，无模型则用 afftdn；aphaser 相位扰动可干扰 Content ID 指纹
|
||
_arnndn_paths = [
|
||
os.path.join(_script_dir, "arnndn-models", "cb.rnnn"),
|
||
os.path.join(os.path.dirname(_script_dir), "arnndn-models", "cb.rnnn"),
|
||
"arnndn-models/cb.rnnn",
|
||
]
|
||
_arnndn_model = None
|
||
for _p in _arnndn_paths:
|
||
if os.path.isfile(_p):
|
||
_arnndn_model = os.path.normpath(_p)
|
||
break
|
||
if _arnndn_model:
|
||
print(f"[INFO] 使用 RNNoise 降噪模型: {_arnndn_model}")
|
||
_arnndn_base = os.path.dirname(os.path.dirname(os.path.abspath(_arnndn_model)))
|
||
_af_denoise = "arnndn=m=arnndn-models/cb.rnnn:mix=0.82,"
|
||
_ffmpeg_cwd = _arnndn_base
|
||
else:
|
||
print("[INFO] 未找到 arnndn 模型，使用 afftdn 降噪（可克隆 richardpl/arnndn-models 到 arnndn-models/）")
|
||
_af_denoise = "afftdn=nf=-26:tn=1,"
|
||
|
||
# 强化版：收紧带通 + 频率偏移 + 音高微调 + 多段EQ + 相位扰动（纯 FFmpeg 理论上限 ~70%）
|
||
_af_chain = (
|
||
_af_denoise
|
||
+ "highpass=f=200,lowpass=f=3800,"
|
||
+ "equalizer=f=150:width_type=o:width=2.5:g=-14,"
|
||
+ "equalizer=f=350:width_type=o:width=2:g=-10,"
|
||
+ "equalizer=f=600:width_type=o:width=1.8:g=-8,"
|
||
+ "equalizer=f=1000:width_type=o:width=1.5:g=-6,"
|
||
+ "equalizer=f=2500:width_type=o:width=1.2:g=-4,"
|
||
+ "acompressor=threshold=-26dB:ratio=5:attack=8:release=80:makeup=5,"
|
||
+ "afreqshift=shift=22,"
|
||
+ "asetrate=48000*1.035,aresample=48000,"
|
||
+ "aphaser=type=t:decay=0.4:delay=3:speed=0.7,"
|
||
+ "volume=1.12,"
|
||
+ "aresample=async=1:first_pts=0"
|
||
)
|
||
|
||
# 720p 竖屏，符合 YouTube 推荐：4–6 Mbps、2s 关键帧、AAC 128k
|
||
_video_mid = ["-vsync", "cfr", "-g", "60", "-keyint_min", "60", "-r", "30",
|
||
"-bf", "0", "-sc_threshold", "0"]
|
||
if codec_v == "libx264":
|
||
_video_mid += ["-x264-params", "force-cfr=1:scenecut=0:open_gop=0:sync-lookahead=0"]
|
||
def _build_ffmpeg_command(input_url: str) -> list[str]:
|
||
cmd = [
|
||
"ffmpeg", "-y", "-loglevel", "info", "-nostdin",
|
||
"-stats", "-stats_period", "1",
|
||
"-rw_timeout", "50000000",
|
||
"-reconnect", "1", "-reconnect_at_eof", "1", "-reconnect_streamed", "1",
|
||
"-reconnect_delay_max", str(RECONNECT_DELAY_MAX),
|
||
"-fflags", "+genpts+discardcorrupt",
|
||
"-err_detect", "ignore_err",
|
||
"-max_delay", "500000",
|
||
"-thread_queue_size", "2048",
|
||
"-filter_threads", "4",
|
||
"-headers", douyin_headers,
|
||
"-re",
|
||
"-i", input_url,
|
||
"-vf", scale_vf,
|
||
"-c:v", codec_v,
|
||
*video_args,
|
||
*_video_mid,
|
||
"-pix_fmt", "yuv420p",
|
||
"-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2",
|
||
"-af", _af_chain,
|
||
"-flags", "+global_header",
|
||
"-flvflags", "no_duration_filesize",
|
||
"-f", "flv", youtube_rtmp,
|
||
]
|
||
return apply_push_proxy_to_ffmpeg_cmd(cmd, push_proxy, push_proxy_mode, base_dir=_script_dir)
|
||
|
||
ffmpeg_command, _uses_proxychains = _build_ffmpeg_command(source_candidates[source_index])
|
||
_emit_relay_heartbeat("starting")
|
||
|
||
proc = None
|
||
stderr_q = None
|
||
reader_t = None
|
||
retry_delay = INITIAL_RETRY_DELAY
|
||
fast_speed_strikes = 0
|
||
invalid_drop_count = 0
|
||
invalid_drop_window_start = 0.0
|
||
last_progress_frame = 0
|
||
last_progress_time = 0.0
|
||
|
||
try:
|
||
while True:
|
||
try:
|
||
if retry_delay > INITIAL_RETRY_DELAY:
|
||
print(f"[WARN] 防风控等待 {retry_delay}s 后重新连接抖音源...")
|
||
time.sleep(retry_delay)
|
||
|
||
cleanup_proc(proc)
|
||
stderr_q = queue.Queue()
|
||
|
||
print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 YouTube 推流 → {stream_title}")
|
||
_popen_kw = dict(
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
bufsize=1,
|
||
universal_newlines=True,
|
||
)
|
||
if _ffmpeg_cwd:
|
||
_popen_kw["cwd"] = _ffmpeg_cwd
|
||
base_env = merge_child_env()
|
||
_popen_kw["env"] = ffmpeg_env_with_push_proxy(
|
||
push_proxy,
|
||
push_proxy_mode,
|
||
use_proxychains=_uses_proxychains,
|
||
base_env=base_env,
|
||
)
|
||
proc = subprocess.Popen(ffmpeg_command, **_popen_kw)
|
||
proc.last_out_ts = time.time()
|
||
|
||
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()
|
||
fast_speed_strikes = 0
|
||
|
||
while True:
|
||
try:
|
||
try:
|
||
line = stderr_q.get(timeout=STDERR_QUEUE_TIMEOUT)
|
||
except queue.Empty:
|
||
if time.time() - getattr(proc, "last_out_ts", time.time()) > STALE_OUTPUT_SECONDS:
|
||
print("[WARN] 长时间无任何日志，强制重启进程")
|
||
cleanup_proc(proc)
|
||
raise SourceStalled()
|
||
continue
|
||
|
||
if line is None:
|
||
if proc.poll() is not None:
|
||
break
|
||
continue
|
||
|
||
line = line.strip()
|
||
if line.startswith("__ERR__READER__"):
|
||
print(f"[ERROR] stderr reader 异常: {line}")
|
||
continue
|
||
|
||
if line:
|
||
proc.last_out_ts = time.time()
|
||
print(f"[FFmpeg] {line}")
|
||
line_lower = line.lower()
|
||
is_progress_line = "frame=" in line_lower and "speed=" in line_lower
|
||
|
||
if any(kw in line_lower for kw in TERMINAL_KEYWORDS):
|
||
print("[ERROR] 检测到主播真下播或被明确拒绝，停止推流")
|
||
cleanup_proc(proc)
|
||
raise StreamEnded()
|
||
|
||
if any(kw in line_lower for kw in TRANSIENT_KEYWORDS):
|
||
print("[WARN] 检测到瞬时网络/IO 错误，准备重连")
|
||
cleanup_proc(proc)
|
||
retry_delay = min(max(retry_delay * 2, INITIAL_RETRY_DELAY), MAX_RETRY_DELAY)
|
||
break
|
||
|
||
now_ts = time.time()
|
||
if _looks_like_invalid_ts_drop(line_lower):
|
||
if now_ts - invalid_drop_window_start > INVALID_DROP_WINDOW_SECONDS:
|
||
invalid_drop_window_start = now_ts
|
||
invalid_drop_count = 0
|
||
invalid_drop_count += 1
|
||
if invalid_drop_count >= INVALID_DROP_LIMIT:
|
||
print("[WARN] 源流时间戳异常过多，重启 FFmpeg")
|
||
cleanup_proc(proc)
|
||
raise SourceStalled()
|
||
|
||
if is_progress_line:
|
||
speed = _ffmpeg_speed_value(line_lower)
|
||
frame, media_time = _ffmpeg_progress_values(line_lower)
|
||
_emit_relay_heartbeat(
|
||
"streaming",
|
||
frame=frame,
|
||
media_time=media_time,
|
||
speed=speed,
|
||
source_index=source_index,
|
||
)
|
||
if speed is not None and time.time() - start_time > START_CHECK_AFTER:
|
||
if speed > FAST_SPEED_THRESHOLD:
|
||
fast_speed_strikes += 1
|
||
if fast_speed_strikes >= FAST_SPEED_STRIKES_NEEDED:
|
||
print(f"[WARN] 推流 speed={speed:.2f}x 异常，重启 FFmpeg")
|
||
cleanup_proc(proc)
|
||
raise SourceStalled()
|
||
else:
|
||
fast_speed_strikes = 0
|
||
|
||
frame, media_time = _ffmpeg_progress_values(line_lower)
|
||
if frame is not None or media_time is not None:
|
||
frame_advanced = frame is not None and frame > last_progress_frame
|
||
time_advanced = media_time is not None and media_time > last_progress_time + 0.01
|
||
if frame_advanced or time_advanced:
|
||
last_frame_time = time.time()
|
||
retry_delay = INITIAL_RETRY_DELAY
|
||
if frame is not None:
|
||
last_progress_frame = frame
|
||
if media_time is not None:
|
||
last_progress_time = media_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 无新帧，交回监控循环重新获取直播源")
|
||
cleanup_proc(proc)
|
||
raise SourceStalled()
|
||
|
||
except (StreamEnded, SourceStalled):
|
||
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 SourceStalled:
|
||
if source_index + 1 < len(source_candidates):
|
||
source_index += 1
|
||
ffmpeg_command, _uses_proxychains = _build_ffmpeg_command(source_candidates[source_index])
|
||
print(
|
||
f"[WARN] 切换备用源 URL ({source_index + 1}/{len(source_candidates)}): "
|
||
f"{source_candidates[source_index][:120]}"
|
||
)
|
||
_emit_relay_heartbeat("source_failover", source_index=source_index)
|
||
retry_delay = INITIAL_RETRY_DELAY
|
||
continue
|
||
print(f"[WARN] 当前直播源无进度，返回监控循环刷新源地址 → {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 Exception as e:
|
||
print(f"[FATAL] YouTube 推流最外层异常: {e}")
|
||
traceback.print_exc()
|
||
finally:
|
||
cleanup_proc(proc)
|
||
clear_relay_heartbeat(_relay_pm2, base_dir=_script_dir)
|
||
release_relay_slot(record_name, pm2=_relay_pm2, base_dir=_script_dir)
|
||
try:
|
||
from web2_relay_log import on_relay_push_end, relay_db_path
|
||
|
||
on_relay_push_end(
|
||
_relay_pm2,
|
||
anchor_name=anchor_name,
|
||
reason="push_stopped",
|
||
db_path=relay_db_path(Path(_script_dir)),
|
||
)
|
||
except Exception as relay_log_exc:
|
||
print(f"[WARN] relay session log end skipped: {relay_log_exc}")
|
||
|
||
try:
|
||
recording.discard(record_name)
|
||
recording_time_list.pop(record_name, None)
|
||
if real_url in running_list:
|
||
running_list.remove(real_url)
|
||
monitoring[0] = max(0, monitoring[0] - 1)
|
||
clear_record_info(record_name, real_url)
|
||
except Exception:
|
||
pass
|
||
|
||
color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN) |