472 lines
18 KiB
Python
472 lines
18 KiB
Python
# douyin_youtube_ffplay.py
|
||
# Linux：仓库内推流实现，编码由 config/hardware.env（LIVE_VIDEO_ENCODER）与可选 NVENC 探测决定。
|
||
import queue
|
||
import unicodedata
|
||
import platform
|
||
import traceback
|
||
import time
|
||
import datetime
|
||
import subprocess
|
||
import threading
|
||
import re
|
||
import os
|
||
import configparser
|
||
|
||
|
||
class StreamEnded(Exception):
|
||
"""主播真实下播的专用异常"""
|
||
pass
|
||
|
||
|
||
_FFMPEG_STATS_PERIOD_OK: bool | None = None
|
||
|
||
|
||
def _require_ffmpeg_stats_period() -> None:
|
||
"""推流依赖 -stats_period（ffmpeg 约 4.4+）；Debian 11 默认 4.3 需升级。"""
|
||
global _FFMPEG_STATS_PERIOD_OK
|
||
if _FFMPEG_STATS_PERIOD_OK is True:
|
||
return
|
||
if _FFMPEG_STATS_PERIOD_OK is False:
|
||
raise RuntimeError(
|
||
"ffmpeg 不支持 -stats_period。板端执行: sudo bash scripts/linux/upgrade_ffmpeg_newer.sh"
|
||
)
|
||
try:
|
||
r = subprocess.run(
|
||
["ffmpeg", "-hide_banner", "-h", "full"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30,
|
||
)
|
||
blob = (r.stdout or "") + (r.stderr or "")
|
||
if "-stats_period" in blob:
|
||
_FFMPEG_STATS_PERIOD_OK = True
|
||
return
|
||
except FileNotFoundError:
|
||
_FFMPEG_STATS_PERIOD_OK = False
|
||
raise RuntimeError("未找到 ffmpeg，请先安装。") from None
|
||
except Exception as e:
|
||
print(f"[WARN] 无法检测 ffmpeg 是否支持 -stats_period: {e}")
|
||
return
|
||
_FFMPEG_STATS_PERIOD_OK = False
|
||
raise RuntimeError(
|
||
"ffmpeg 不支持 -stats_period（需约 4.4+）。板端执行: sudo bash scripts/linux/upgrade_ffmpeg_newer.sh"
|
||
)
|
||
|
||
|
||
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,
|
||
clear_record_info,
|
||
color_obj,
|
||
):
|
||
"""
|
||
抖音 → YouTube 推流主函数（单路推流）
|
||
只从 config/youtube.ini 读取一个 stream key（取第一个有效值）
|
||
"""
|
||
|
||
# ====================== 读取 YouTube 配置 ======================
|
||
config_file = "config/youtube.ini"
|
||
rtmps_base = "rtmps://a.rtmp.youtube.com/live2/"
|
||
rtmp_base = "rtmp://a.rtmp.youtube.com/live2/"
|
||
use_rtmps = True
|
||
# 720p 竖屏 YouTube 建议 4–6 Mbps，偏低易触发「数据不足/卡顿」提示
|
||
b_v, maxrate, bufsize = "4000k", "4500k", "8000k"
|
||
stream_key = None
|
||
fast_audio = False # config: fast_audio=1 可跳过 arnndn+长 EQ，显著减轻 CPU
|
||
|
||
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 == "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 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")
|
||
|
||
# ====================== 常量配置 ======================
|
||
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
|
||
|
||
# 明确结束 / 平台拒绝：停止整条转推（与瞬时网络错误区分）
|
||
TERMINAL_KEYWORDS = [
|
||
"live has ended",
|
||
"server returned 404",
|
||
"ingestion error",
|
||
"access denied",
|
||
"server returned 403",
|
||
"client disconnected",
|
||
]
|
||
# 瞬时错误：结束当前 ffmpeg 进程后由外层循环退避重连
|
||
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",
|
||
]
|
||
|
||
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}")
|
||
|
||
# ====================== 视频编码：hardware.env（ARM）→ NVENC → libx264 ======================
|
||
def _load_simple_env(path: str) -> dict[str, str]:
|
||
out: dict[str, str] = {}
|
||
if not os.path.isfile(path):
|
||
return out
|
||
with open(path, "r", encoding="utf-8", errors="ignore") as fh:
|
||
for line in fh:
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
key = k.strip()
|
||
if key.startswith("#"):
|
||
continue
|
||
out[key] = v.strip().strip('"').strip("'")
|
||
return out
|
||
|
||
codec_v = "libx264"
|
||
video_args = [
|
||
"-preset", "ultrafast", "-tune", "zerolatency", "-profile:v", "high", "-level", "4.0",
|
||
"-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize,
|
||
]
|
||
|
||
_config_dir = os.path.dirname(os.path.abspath(config_file))
|
||
_hw_env = _load_simple_env(os.path.join(_config_dir, "hardware.env"))
|
||
_want_enc = (_hw_env.get("LIVE_VIDEO_ENCODER") or "").strip()
|
||
_hw_enc_active = False
|
||
if _want_enc and platform.system().lower() == "linux":
|
||
try:
|
||
enc = subprocess.run(
|
||
["ffmpeg", "-hide_banner", "-encoders"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=6,
|
||
)
|
||
blob = (enc.stdout or "") + (enc.stderr or "")
|
||
if _want_enc in blob and _want_enc.startswith("h264_"):
|
||
codec_v = _want_enc
|
||
video_args = ["-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize]
|
||
_hw_enc_active = True
|
||
print(f"[INFO] 使用 config/hardware.env LIVE_VIDEO_ENCODER={codec_v}（板载硬件编码）")
|
||
except Exception as e:
|
||
print(f"[WARN] LIVE_VIDEO_ENCODER 校验失败: {e}")
|
||
|
||
if not _hw_enc_active:
|
||
try:
|
||
if platform.system().lower() == "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 硬件加速")
|
||
except Exception as e:
|
||
print(f"[WARN] NVENC 检测失败，回退软件编码: {e}")
|
||
|
||
_cpu_n = os.cpu_count() or 4
|
||
_filter_threads = str(min(4, max(1, _cpu_n // 2)))
|
||
|
||
# ====================== 开源音频处理：背景音乐弱化 + 防版权 ======================
|
||
_script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
_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"]
|
||
# -re：按源时间戳节奏拉流，避免突发过快；不用 realtime/arealtime（输入抖动时易让 YouTube 判为数据不足）
|
||
ffmpeg_command = [
|
||
"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", _filter_threads,
|
||
"-headers", douyin_headers,
|
||
"-re",
|
||
"-i", real_url,
|
||
# bilinear 比 lanczos 轻很多，利于 speed 稳定在 ≥1.0x
|
||
"-vf", "fps=30,scale=720:1280:force_original_aspect_ratio=decrease:flags=bilinear,pad=720:1280:(ow-iw)/2:(oh-ih)/2:black",
|
||
"-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
|
||
]
|
||
|
||
_require_ffmpeg_stats_period()
|
||
|
||
proc = None
|
||
stderr_q = None
|
||
reader_t = None
|
||
retry_delay = INITIAL_RETRY_DELAY
|
||
|
||
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
|
||
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()
|
||
|
||
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)
|
||
break
|
||
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()
|
||
|
||
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(f"[WARN] 瞬时错误，将退避重试: {line[:160]}")
|
||
cleanup_proc(proc)
|
||
retry_delay = min(max(retry_delay, INITIAL_RETRY_DELAY) * 2, MAX_RETRY_DELAY)
|
||
break
|
||
|
||
if re.search(r'frame=\s*\d+', line_lower) or "fps=" in line_lower:
|
||
last_frame_time = time.time()
|
||
retry_delay = INITIAL_RETRY_DELAY
|
||
|
||
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
|
||
|
||
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 Exception as e:
|
||
print(f"[FATAL] YouTube 推流最外层异常: {e}")
|
||
traceback.print_exc()
|
||
finally:
|
||
cleanup_proc(proc)
|
||
|
||
try:
|
||
recording_time_list.pop(record_name, None)
|
||
# 统一由 clear_record_info 维护 running_list / monitoring / url_comments
|
||
clear_record_info(record_name, real_url)
|
||
except Exception:
|
||
pass
|
||
|
||
color_obj.print_colored(f"[{record_name}] YouTube 推流已彻底停止并清理完毕\n", color_obj.GREEN)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 兼容旧 PM2/脚本名：实际业务入口为 youtube.py（监控 URL_config + 调用本模块推流）
|
||
import runpy
|
||
from pathlib import Path
|
||
|
||
_root = Path(__file__).resolve().parent
|
||
runpy.run_path(str(_root / "youtube.py"), run_name="__main__") |