@@ -56,7 +56,10 @@ def _yt_get(section: str, key: str, fallback=None, cast=str):
return _config . get ( section , key , fallback = fallback )
except Exception :
return fallback
# ===【单进程推流协调模式】全局控制 ===
yt_single_mode = _yt_get ( " youtube " , " single_mode " , fallback = True , cast = bool ) # 建议默认 True
yt_thread_started = False
yt_coordinator_lock = threading . Lock ( )
def _safe_int ( v , lo , hi , default ) :
try :
v = int ( v )
@@ -65,6 +68,196 @@ def _safe_int(v, lo, hi, default):
return v
except Exception :
return default
def _yt_build_output_args ( port_info , res_str , bv , ba , fps , gop , allow_copy ) :
# 输出端（向 YouTube）
v_args = [
" -c:v " , " libx264 " ,
" -preset " , " veryfast " ,
" -profile:v " , " high " ,
" -level:v " , " 4.2 " ,
" -pix_fmt " , " yuv420p " ,
" -r " , str ( fps ) ,
" -g " , str ( gop ) ,
" -b:v " , f " { bv } k " ,
" -maxrate " , f " { bv } k " ,
" -bufsize " , f " { max ( 2 * bv , 4000 ) } k " ,
]
a_args = [
" -c:a " , " aac " ,
" -b:a " , f " { ba } k " ,
" -ar " , " 44100 " ,
" -ac " , " 2 " ,
]
vf_chain = [ ]
if res_str :
try :
w , h = map ( int , res_str . lower ( ) . split ( " x " ) )
vf_chain + = [
f " scale= { w } : { h } :force_original_aspect_ratio=decrease " ,
f " pad= { w } : { h } :(ow-iw)/2:(oh-ih)/2 " ,
" format=yuv420p " ,
]
except Exception :
logger . warning ( f " 警告: resolution ' { res_str } ' 非法，已忽略 " )
# 源若已是 h264+aac 且未强制缩放，尝试零转码 copy
if allow_copy and not vf_chain and _can_copy_h264_aac ( port_info ) :
v_args = [ " -c:v " , " copy " ]
a_args = [ " -c:a " , " copy " ]
base = [ " -map " , " 0 " ]
if vf_chain :
base + = [ " -vf " , " , " . join ( vf_chain ) ]
base + = v_args + a_args + [
" -flvflags " , " no_duration_filesize " ,
" -rtmp_live " , " live " ,
]
return base
def _yt_make_input_chain ( input_url , proxy_address , headers ,
rw_timeout = " 15000000 " , analyzeduration = " 20000000 " ,
probesize = " 10000000 " , bufsize = " 8000k " ,
max_muxing_queue_size = " 1024 " ) :
# 输入端（从抖音等源拉流）
user_agent = ( " Mozilla/5.0 (Linux; Android 11; SAMSUNG SM-G973U) AppleWebKit/537.36 "
" (KHTML, like Gecko) SamsungBrowser/14.2 Chrome/87.0.4280.141 Mobile Safari/537.36 " )
cmd = [
" ffmpeg " , " -y " ,
" -v " , " verbose " ,
" -rw_timeout " , rw_timeout ,
" -loglevel " , " error " ,
" -hide_banner " ,
" -user_agent " , user_agent ,
" -protocol_whitelist " , " rtmp,crypto,file,http,https,tcp,tls,udp,rtp,httpproxy " ,
" -thread_queue_size " , " 1024 " ,
" -analyzeduration " , analyzeduration ,
" -probesize " , probesize ,
" -fflags " , " +discardcorrupt " ,
" -re " , " -i " , input_url ,
" -bufsize " , bufsize ,
" -sn " , " -dn " ,
" -reconnect_delay_max " , " 60 " ,
" -reconnect_streamed " , " -reconnect_at_eof " ,
" -max_muxing_queue_size " , max_muxing_queue_size ,
" -correct_ts_overflow " , " 1 " ,
" -avoid_negative_ts " , " 1 " ,
]
if headers :
cmd . insert ( 11 , " -headers " )
cmd . insert ( 12 , headers )
if proxy_address :
cmd . insert ( 1 , " -http_proxy " )
cmd . insert ( 2 , proxy_address )
return cmd
def _try_open_rtmp_out ( key , secure , ingest ) :
# 优先 rtMPS；
nodes = [ ingest ] if ingest and len ( ingest ) == 1 else [ " a " ]
nodes + = [ c for c in " bcdef " if c not in nodes ]
schemes = [ " rtmps " , " rtmp " ] if secure else [ " rtmp " , " rtmps " ]
for sch in schemes :
for node in nodes :
yield _build_youtube_rtmp_url ( key , sch == " rtmps " , node )
def yt_single_coordinator ( record_quality , proxy_address ) :
"""
单进程 YouTube 转播协调器：
- 轮询 URL_config.ini， ；
- 找到就启动 ffmpeg 推到同一条 YouTube live；
- 源断/下播 -> 立刻切到下一路在播；
- 没有任何在播源 -> 不在面板里显示录制项（保持空）。
"""
key = ( _yt_get ( " youtube " , " key " , fallback = " " , cast = str ) or " " ) . strip ( )
if not key :
logger . error ( " 错误: youtube.ini[youtube].key 未配置或为空 " )
return
secure = _yt_get ( " youtube " , " secure " , fallback = True , cast = bool )
ingest = ( _yt_get ( " youtube " , " ingest " , fallback = " a " , cast = str ) or " a " ) . strip ( ) . lower ( )
res_str = ( _yt_get ( " youtube " , " resolution " , fallback = " " , cast = str ) or " " ) . strip ( )
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 )
fps = _safe_int ( _yt_get ( " youtube " , " fps " , fallback = 30 , cast = int ) , 10 , 60 , 30 )
gop = _safe_int ( _yt_get ( " youtube " , " gop " , fallback = 2 * fps , cast = int ) , fps , 4 * fps , 2 * fps )
allow_copy = _yt_get ( " youtube " , " copy_when_safe " , fallback = True , cast = bool )
out_url_candidates = list ( _try_open_rtmp_out ( key , secure , ingest ) )
out_url_idx = 0
# 输入端时序阈值（海外可加大）
rw_timeout = " 15000000 " ; analyzeduration = " 20000000 " ; probesize = " 10000000 "
bufsize = " 8000k " ; max_muxing_queue_size = " 1024 "
current_display_name = None # 面板上当前显示的条目名（带主播）
current_anchor = None
while True :
# 1) 找到“第一条在播”的源
real_url , port_info = find_next_live_source ( None , proxy_address , record_quality )
if not real_url :
# 没有任何在播源 -> 清理面板显示
if current_display_name :
recording . discard ( current_display_name )
recording_time_list . pop ( current_display_name , None )
current_display_name = None
current_anchor = None
time . sleep ( 3 )
continue
# 2) 更新面板显示名称为“转播·主播名”
anchor = clean_name ( port_info . get ( " anchor_name " ) or " 直播 " )
display_name = f " 转播· { anchor } "
if display_name != current_display_name :
if current_display_name :
recording . discard ( current_display_name )
recording_time_list . pop ( current_display_name , None )
recording . add ( display_name )
recording_time_list [ display_name ] = [ datetime . datetime . now ( ) , video_record_quality ]
current_display_name = display_name
current_anchor = anchor
headers = None # 如需平台定制 headers 可在此设置
in_cmd = _yt_make_input_chain ( real_url , proxy_address , headers ,
rw_timeout , analyzeduration , probesize ,
bufsize , max_muxing_queue_size )
out_args = _yt_build_output_args ( port_info , res_str , bv , ba , fps , gop , allow_copy )
# 3) 逐个尝试 YouTube 出口（节点/协议）
tried = 0
while tried < len ( out_url_candidates ) :
out_url = out_url_candidates [ out_url_idx % len ( out_url_candidates ) ]
out_url_idx + = 1
tried + = 1
run_cmd = list ( in_cmd ) + out_args + [ " -f " , " flv " , out_url ]
logger . info ( f " [YT协调] { anchor } -> { out_url } " )
proc = subprocess . Popen (
run_cmd , stdin = subprocess . PIPE , stderr = subprocess . STDOUT ,
startupinfo = get_startup_info ( os_type )
)
# 等待 ffmpeg：
while proc . poll ( ) is None :
# 如需全局停止/注释判断可在此处加判断后优雅退出
time . sleep ( 1 )
rc = proc . returncode
if rc == 0 :
# 正常退出（极少）-> 直接跳出去重新找源
break
else :
logger . warning ( f " [YT协调] 推流退出 rc= { rc } ，更换 YouTube 节点/协议重试… " )
# 换节点继续 while tried…
# 4) 当前源结束/异常：回到循环顶部去找下一路在播；若没有，则上面的分支会清空面板
time . sleep ( 1 )
def read_all_urls_from_list ( url_file : str ) - > list [ tuple [ str , str ] ] :
"""
读取 URL_config.ini，
@@ -1975,27 +2168,43 @@ while True:
new_word = replace_words [ 1 ]
update_file ( url_config_file , old_str = replace_words [ 0 ] , new_str = new_word , start_str = start_with )
text_no_repeat_url = list ( set ( url_tuples_list ) )
# 保序去重（文件顺序 = 备源优先级）
text_no_repeat_url = list ( dict . fromkeys ( url_tuples_list ) )
if len ( text_no_repeat_url ) > 0 :
for url_tuple in text_no_repeat_url :
monitoring = len ( running_list )
# 监测数量显示为“URL 条数”（哪怕不启动录制线程）
monitoring = len ( text_no_repeat_url )
if url_tuple [ 1 ] in not_record_list :
continue
if yt_single_mode and video_save_type . upper ( ) == " MP4 " :
# ===== 单进程协调模式：只启 1 个协调器，不再为每个 URL 启动 start_record =====
with yt_coordinator_lock :
# global yt_thread_started
if not yt_thread_started :
print ( " \n 启用【单进程 YouTube 转播协调模式】（只保留 1 个 ffmpeg 推流进程） " )
record_quality = get_quality_code ( video_record_quality )
args = ( record_quality , proxy_addr if use_proxy else None )
t_yt = threading . Thread ( target = yt_single_coordinator , args = args , daemon = True )
t_yt . start ( )
yt_thread_started = True
else :
# ===== 传统模式：按原先逻辑为每个 URL 启动录制（本地落地 / 其他保存格式）=====
if len ( text_no_repeat_url ) > 0 :
for url_tuple in text_no_repeat_url :
if url_tuple [ 1 ] in not_record_list :
continue
if url_tuple [ 1 ] not in running_list :
print ( f " \r { ' 新增 ' if not first_start else ' 传入 ' } 地址: { url_tuple [ 1 ] } " )
idx = len ( running_list ) + 1
args = [ url_tuple , idx ]
create_var [ f ' thread_ { idx } ' ] = threading . Thread ( target = start_record , args = args )
create_var [ f ' thread_ { idx } ' ] . daemon = True
create_var [ f ' thread_ { idx } ' ] . start ( )
running_list . append ( url_tuple [ 1 ] )
time . sleep ( local_delay_default )
if url_tuple [ 1 ] not in running_list :
print ( f " \r { ' 新增 ' if not first_start else ' 传入 ' } 地址: { url_tuple [ 1 ] } " )
monitoring + = 1
args = [ url_tuple , monitoring ]
create_var [ f ' thread_ { monitoring } ' ] = threading . Thread ( target = start_record , args = args )
create_var [ f ' thread_ { monitoring } ' ] . daemon = True
create_var [ f ' thread_ { monitoring } ' ] . start ( )
running_list . append ( url_tuple [ 1 ] )
time . sleep ( local_delay_default )
url_tuples_list = [ ]
first_start = False
except Exception as err :
logger . error ( f " 错误信息: { err } 发生错误的行数: { err . __traceback__ . tb_lineno } " )
eric