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-09-27 17:16:13 +08:00
parent 34990a29de
commit 0067152f69
2 changed files with 106 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config/youtube.ini
View File

@@ -7,5 +7,5 @@ audio_bitrate = 128
fps = 30
gop = 60
copy_when_safe = true
resolution =
resolution = 1080x1920
single_mode = true

185
main.py
View File

@@ -28,7 +28,7 @@ from src.utils import logger, Color, check_md5
from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
# --------------------- 基础环境 ---------------------
version = "relay-1.0.0"
version = "relay-1.0.1"
os.environ['PATH'] = ffmpeg_path + os.pathsep + current_env_path
script_path = os.path.split(os.path.realpath(sys.argv[0]))[0]
config_file = f'{script_path}/config/config.ini'
@@ -186,7 +186,6 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
else:
logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。")
port_info = {"anchor_name": "", "is_live": False}
elif (".m3u8" in url) or (".flv" in url):
# 自定义直链
port_info = {
@@ -214,7 +213,7 @@ def build_youtube_output() -> Tuple[str, list, list]:
secure = _yt_get("youtube", "secure", fallback=False, cast=bool)
ingest = _yt_get("youtube", "ingest", fallback="a", cast=str)
res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip()
res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() # 例如 "1080x1920"
bv = _safe_int(_yt_get("youtube", "bitrate", fallback=5000, cast=int), 300, 20000, 5000)
ba = _safe_int(_yt_get("youtube", "audio_bitrate", fallback=128, cast=int), 64, 384, 128)
@@ -223,10 +222,11 @@ def build_youtube_output() -> Tuple[str, list, list]:
out_url = _build_youtube_rtmp_url(key, secure, ingest)
# —— 默认强制转码（最稳），如果想省 CPU 可把 -c:v libx264 改为 copy —— #
# —— 默认强制转码（稳） —— #
v_args = [
"-c:v", "libx264",
"-preset", "veryfast",
"-tune", "zerolatency",
"-profile:v", "high",
"-level:v", "4.2",
"-pix_fmt", "yuv420p",
@@ -243,34 +243,46 @@ def build_youtube_output() -> Tuple[str, list, list]:
"-ac", "2",
]
# 可选缩放/留黑边，保持 YT 吃得更稳
# 可选缩放/留黑边 + 避免 1088x1920（加 crop 强制精确尺寸）
if res_str:
try:
w, h = map(int, res_str.lower().split("x"))
vf = f"scale={w}:{h}:force_original_aspect_ratio=decrease,pad={w}:{h}:(ow-iw)/2:(oh-ih)/2,format=yuv420p"
vf = (
f"scale={w}:{h}:force_original_aspect_ratio=decrease,"
f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2,"
f"crop={w}:{h},"
f"format=yuv420p"
)
v_args = ["-vf", vf] + v_args
except Exception:
logger.warning(f"resolution '{res_str}' 非法，忽略")
return out_url, v_args, a_args
def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = None) -> subprocess.Popen:
def launch_ffmpeg(input_url: str,
out_url: str,
extra_headers: Optional[str] = None,
v_args: Optional[list] = None,
a_args: Optional[list] = None) -> subprocess.Popen:
"""
启动一个 ffmpeg 进程：输入 input_url，输出 RTMP 到 out_url。
启动一个 ffmpeg 进程：输入 input_url，输出 RTMP/RTMPS 到 out_url。
- v_args / a_args 由 build_youtube_output() 生成并传入
- 启动 stdout reader 线程以防缓冲阻塞
- 设置代理环境变量以兼容 https/hls
"""
import threading
user_agent = ("Mozilla/5.0 (Linux; Android 11; SM-G973U) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/87.0.4280.141 Mobile Safari/537.36")
# —— 输入侧容错/重连/时间戳修复 ——
rw_timeout = "15000000"
analyzeduration = "20000000"
probesize = "10000000"
# 输入侧容错/重连/探测
base = [
"ffmpeg", "-nostdin", "-hide_banner",
"-stats", # 输出统计信息
"-stats_period", "60", # 每60秒一次
"-progress", "pipe:1", # 状态写到stdout，便于reader捕获
"-loglevel", "info",
# 输入容错 & 自动重连
"-rw_timeout", "15000000",
"-timeout", "10000000",
@@ -279,68 +291,69 @@ def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = N
"-reconnect_at_eof", "1",
"-reconnect_delay_max", "30",
# User-Agent 和协议
# UA/协议白名单
"-user_agent", user_agent,
"-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy",
# 缓冲 & 探测优化
# 缓冲 & 探测
"-thread_queue_size", "4096",
"-analyzeduration", "10000000",
"-probesize", "10000000",
]
# 输入流
"-re", "-i", input_url,
# 容错标志：防止丢包、断音直接崩溃
# 平台 headers（referer/origin）要在 -i 之前
if extra_headers:
base += ["-headers", extra_headers]
# 代理：http_proxy/https_proxy 环境变量（同时保留 -http_proxy 兼容）
env = os.environ.copy()
if proxy_addr:
env["http_proxy"] = proxy_addr
env["https_proxy"] = proxy_addr
base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:]
# 输入
base += ["-re", "-i", input_url]
# 通用容错
base += [
"-fflags", "+genpts+discardcorrupt+nobuffer",
"-use_wallclock_as_timestamps", "1",
# 视频编码
"-c:v", "libx264",
"-preset", "veryfast",
"-tune", "zerolatency",
"-profile:v", "high",
"-level:v", "4.1",
"-pix_fmt", "yuv420p",
"-r", "30",
"-g", "60",
"-b:v", "5000k",
"-maxrate", "5000k",
"-bufsize", "10000k",
"-vf", "scale=1080:1920:force_original_aspect_ratio=decrease,pad=1080:1920:(ow-iw)/2:(oh-ih)/2",
# 音频编码
"-c:a", "aac",
"-b:a", "128k",
"-ar", "44100",
"-ac", "2",
# 输出到 YouTube
"-f", "flv", out_url,
"-fflags", "+genpts+discardcorrupt+nobuffer+flush_packets",
"-avioflags", "direct",
"-rtbufsize", "100M",
]
# 视频/音频编码参数（使用 build_youtube_output 的结果）
v_args = v_args or []
a_args = a_args or []
base += v_args + a_args
# 平台需要自定义 header（如 referer/origin）
if extra_headers:
# 插到 -user_agent 后、-thread_queue_size 前
i1 = base.index("-user_agent") + 2
i2 = base.index("-thread_queue_size")
base[i1:i1] = ["-headers", extra_headers]
# 输出（flv 到 YT）
base += ["-f", "flv", out_url]
# 代理
if proxy_addr:
base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:]
# 打印命令便于排障（截断）
cmd_str = " ".join(base)
print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...")
# 打印命令便于排障
print("ffmpeg cmd:", " ".join(base if len(" ".join(base)) < 500 else base[:30] + ["..."]))
logger.info(f"启动推流：{input_url} -> {out_url}")
logger.info(f"启动推流：{input_url} -> YouTube")
p = subprocess.Popen(base, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1)
creationflags = 0
if os.name == "nt":
# 防止子进程挂住控制台（可选）
creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
# —— 启动 watchdog：若 20s 没有任何输出，判定卡死，主循环会重启它 ——
p = subprocess.Popen(
base,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=0, # 避免行缓冲告警 & 尽快刷新
env=env,
creationflags=creationflags
)
# 最近输出时间戳，用于 watchdog
p.last_out_ts = time.time()
def _reader():
@@ -348,12 +361,19 @@ def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = N
for line in iter(p.stdout.readline, b''):
if not line:
break
p.last_out_ts = time.time() # 任何输出都视为“仍在工作”
text = line.decode(errors="ignore").strip()
p.last_out_ts = time.time()
# 常见错误提示
if "Server disconnected" in text:
logger.error(f"[YouTube输出异常] {text}")
elif "not supported" in text or "Invalid" in text or "invalid" in text:
logger.error(f"[转码异常] {text}")
except Exception:
pass
t = threading.Thread(target=_reader, daemon=True)
t.start()
return p
def stop_ffmpeg(p: Optional[subprocess.Popen]):
@@ -367,9 +387,10 @@ def stop_ffmpeg(p: Optional[subprocess.Popen]):
p.stdin.flush()
except Exception:
pass
p.wait(timeout=6)
else:
p.send_signal(signal.SIGINT)
p.wait(timeout=10)
p.wait(timeout=10)
except Exception:
try:
p.kill()
@@ -439,6 +460,10 @@ def main():
scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒
last_md5 = "" # URL 文件变化时热更新
# watchdog 参数
watchdog_grace = 60 # 推流开始 60s 内不做静默判定
watchdog_silence = 300 # 5 分钟完全无输出才重启
try:
while True:
# 动态热更新 URL 列表（URL_config.ini 有变更则重载）
@@ -457,33 +482,22 @@ def main():
except Exception as e:
logger.warning(f"URL 列表热更新失败：{e}")
if current_proc and current_url:
# 已在推流：检测是否仍直播 & 进程状态
if current_proc.poll() is not None:
color.print_colored(f"[{now_str()}] ffmpeg 进程退出，重启推流…", color.YELLOW)
retcode = current_proc.poll()
if retcode is not None:
color.print_colored(f"[{now_str()}] ffmpeg 已退出 (code={retcode})，切源…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = launch_ffmpeg(real_url, out_url, headers)
current_proc.last_out_ts = time.time()
current_proc = None
current_url = None
continue
# 检测 ffmpeg 是否卡住（10 秒无输出）
no_out_for = time.time() - getattr(current_proc, "last_out_ts", time.time())
if no_out_for > 20: # ✅ 改成 20 秒
color.print_colored(f"\n[{now_str()}] ffmpeg 20s 无输出，尝试重启当前源…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = launch_ffmpeg(real_url, out_url, headers)
current_proc.last_out_ts = time.time()
continue
# 轻量检测：仍直播？
# 检查是否停播（依赖 spider）
try:
port = asyncio.run(fetch_port_info(current_url, record_quality_code))
still_live = bool(port.get("is_live"))
except Exception as e:
logger.error(f"检测当前源失败：{e}")
still_live = True # ⚠️ 出错时默认仍在播，避免误切
still_live = True # 容错：无法判断时先认为还活着
if still_live:
elapsed = int(time.time() - start_time)
@@ -491,10 +505,21 @@ def main():
m, s = divmod(m, 60)
dur_str = f"{h:02d}:{m:02d}:{s:02d}"
print(f"\r[{now_str()}] 正在转播({dur_str})：{current_platform} | {current_anchor} | {current_url}", end="")
# ✅ watchdog（推流开始 > 60s 且 5 分钟完全无输出）
silence = time.time() - getattr(current_proc, "last_out_ts", 0)
if elapsed >= watchdog_grace and silence > watchdog_silence:
color.print_colored(f"\n[{now_str()}] 超过5分钟未见 ffmpeg 输出，判定异常，重启…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
time.sleep(1)
continue
time.sleep(check_interval_live)
continue
else:
color.print_colored(f"\n[{now_str()}] 主播停播，才切换下一个源…", color.YELLOW)
color.print_colored(f"\n[{now_str()}] 主播停播，切换下一个源…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
@@ -532,8 +557,8 @@ def main():
headers = headers_for_platform(current_platform, url)
# 启动 ffmpeg
out_url, _, _ = build_youtube_output()
current_proc = launch_ffmpeg(real_url, out_url, headers)
out_url, v_args, a_args = build_youtube_output()
current_proc = launch_ffmpeg(real_url, out_url, headers, v_args=v_args, a_args=a_args)
current_url = url
start_time = time.time()
color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN)