eric/gitlab-instance-0a899031_douyinyoutube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'-m'

Browse Source
This commit is contained in:
eric
2025-09-27 19:04:22 +08:00
parent 617fdcb158
commit 4c6b040887
5 changed files with 1704 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

617
back/back4 Normal file
View File

@@ -0,0 +1,617 @@
# -*- encoding: utf-8 -*-
"""
单进程 ffmpeg 推流：从 URL_config.ini 顺序检测直播源，
一旦当前源停播/异常，自动切到下一个正在直播的源并推流到 YouTube。
仅做“检测+转推”，不做录制、分段、通知等。
"""
import asyncio
import os
import sys
import signal
import time
import datetime
import re
import uuid
import subprocess
from pathlib import Path
from typing import Any, List, Tuple, Optional
import configparser
import urllib.request
from urllib.error import URLError, HTTPError
import httpx
from src import spider, stream
from src.utils import logger, Color, check_md5
from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
# --------------------- 基础环境 ---------------------
version = "relay-1.1.0"
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'
url_config_file = f'{script_path}/config/URL_config.ini'
text_encoding = 'utf-8-sig'
color = Color()
# ---------- YouTube 推流配置（读取 youtube.ini） ----------
_yt_cfg = configparser.ConfigParser()
with open(f"{script_path}/config/youtube.ini", "r", encoding="utf-8-sig") as f:
_yt_cfg.read_file(f)
def _yt_get(section: str, key: str, fallback=None, cast=str):
try:
if cast is bool:
return _yt_cfg.getboolean(section, key, fallback=fallback)
if cast is int:
return _yt_cfg.getint(section, key, fallback=fallback)
return _yt_cfg.get(section, key, fallback=fallback)
except Exception:
return fallback
def _safe_int(v, lo, hi, default):
try:
v = int(v)
if v < lo or v > hi:
return default
return v
except Exception:
return default
def _round_res(w: int, h: int) -> Tuple[int, int]:
# 强制偶数，避免 1088/1918 这类非标准尺寸
return (w & ~1, h & ~1)
def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None):
node = (ingest or "a").strip().lower()
node = node if node.isalpha() and len(node) == 1 else "a"
scheme = "rtmps" if secure else "rtmp"
host = f"{node}.rtmps.youtube.com" if secure else f"{node}.rtmp.youtube.com"
return f"{scheme}://{host}/live2/{key}"
# --------------------- 简化配置读取 ---------------------
def read_config_value(cfg: configparser.RawConfigParser, section: str, option: str, default_value: Any) -> Any:
try:
cfg.read(config_file, encoding=text_encoding)
if section not in cfg.sections():
cfg.add_section(section)
return cfg.get(section, option)
except (configparser.NoSectionError, configparser.NoOptionError):
cfg.set(section, option, str(default_value))
with open(config_file, 'w', encoding=text_encoding) as f:
cfg.write(f)
return default_value
_CN_BOOL = {"是": True, "否": False}
cfg = configparser.RawConfigParser()
use_proxy = _CN_BOOL.get(read_config_value(cfg, '录制设置', '是否使用代理ip(是/否)', "否"), False)
proxy_addr = read_config_value(cfg, '录制设置', '代理地址', "")
proxy_addr = proxy_addr if use_proxy and proxy_addr.strip() else None
# Cookie（按需）
dy_cookie = read_config_value(cfg, 'Cookie', '抖音cookie', '')
tiktok_cookie = read_config_value(cfg, 'Cookie', 'tiktok_cookie', '')
# 可选：启动时检测系统代理（仅提示）
try:
if not use_proxy:
urllib.request.urlopen("https://www.google.com/", timeout=5)
print("提示：你的网络似乎直连可用（无需代理）。")
except Exception:
if not use_proxy:
color.print_colored("提示：无法直连海外站，如需 TikTok 请在 config.ini 开启代理并填写代理地址。", color.YELLOW)
# --------------------- URL 列表读取 ---------------------
def parse_urls_from_file(path: str) -> List[str]:
"""
读取 URL_config.ini，返回去重且保持相对顺序的 URL 列表。
支持行内格式：
- 纯 URL
- quality,url
- url,主播: xxx
- # 注释行 忽略
"""
urls, seen = [], set()
if not os.path.isfile(path):
open(path, 'w', encoding=text_encoding).close()
print("URL_config.ini 为空，请写入直播间地址后再运行。")
return urls
with open(path, "r", encoding=text_encoding, errors='ignore') as f:
for line in f:
s = line.strip()
if not s or s.startswith("#"):
continue
# 拆分（最多3段，取包含 https 或域名的那段）
parts = re.split(r"[，,]\s*", s)
cand = None
for p in parts:
if '://' in p or re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", p):
cand = p
break
if not cand:
continue
url = cand if '://' in cand else f"https://{cand}"
if url not in seen:
seen.add(url)
urls.append(url)
return urls
# --------------------- 直播解析与选源 ---------------------
def is_flv_preferred_platform(link: str) -> bool:
# 抖音/TikTok 优先走 FLV；但 H265 不走 FLV
return any(i in link for i in ["douyin", "tiktok"])
def select_source_url(link: str, stream_info: dict) -> Optional[str]:
if not stream_info:
return None
if is_flv_preferred_platform(link):
flv = stream_info.get("flv_url")
if flv:
# 若 codec=h265 则不走 flv
try:
from urllib.parse import urlparse, parse_qs
qs = parse_qs(urlparse(flv).query or "")
if (qs.get("codec") or [""])[0].lower() == "h265":
logger.warning("FLV 不支持 h265，改用 HLS。")
else:
return flv
except Exception:
return flv
return stream_info.get("record_url") or stream_info.get("m3u8_url") or stream_info.get("flv_url")
async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
"""
调用项目现有 spider/stream，返回：
{
"anchor_name": str,
"is_live": bool,
"title": str?,
"record_url": "...",
"flv_url": "...",
"m3u8_url": "..."
}
"""
proxy = proxy_addr
port_info = {}
try:
if "douyin.com/" in url:
if 'v.douyin.com' not in url and '/user/' not in url:
j = await spider.get_douyin_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie)
else:
j = await spider.get_douyin_app_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie)
port_info = await stream.get_douyin_stream_url(j, record_quality_code, proxy)
elif url.startswith("https://www.tiktok.com/"):
if proxy:
j = await spider.get_tiktok_stream_data(url=url, proxy_addr=proxy, cookies=tiktok_cookie)
port_info = await stream.get_tiktok_stream_url(j, record_quality_code, proxy)
else:
logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。")
port_info = {"anchor_name": "", "is_live": False}
elif (".m3u8" in url) or (".flv" in url):
# 自定义直链
port_info = {
"anchor_name": "自定义_" + str(uuid.uuid4())[:8],
"is_live": True,
}
if url.endswith(".flv"):
port_info["flv_url"] = url
else:
port_info["m3u8_url"] = url
else:
# 其它站点可按需扩展
port_info = {"anchor_name": "", "is_live": False}
except Exception as e:
logger.error(f"解析失败: {e}")
port_info = {"anchor_name": "", "is_live": False}
return port_info
# --------------------- ffmpeg 单进程推流 ---------------------
def build_youtube_output() -> Tuple[str, list, list]:
key = _yt_get("youtube", "key", fallback="", cast=str)
if not key:
logger.error("错误: config/youtube.ini 未配置 key")
sys.exit(1)
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() # 例如 "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)
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)
out_url = _build_youtube_rtmp_url(key, secure, ingest)
# —— 默认强制转码（稳） —— #
v_args = [
"-c:v", "libx264",
"-preset", "veryfast",
"-tune", "zerolatency",
"-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",
]
# 可选缩放/留黑边 + 避免 1088x1920（crop 强制精确尺寸）+ 偶数对齐
if res_str:
try:
w, h = map(int, res_str.lower().split("x"))
w, h = _round_res(w, h)
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,
v_args: Optional[list] = None,
a_args: Optional[list] = None) -> subprocess.Popen:
"""
启动一个 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")
# 输入侧容错/重连/探测
base = [
"ffmpeg", "-nostdin", "-hide_banner",
"-stats",
"-stats_period", "60",
"-progress", "pipe:1",
"-loglevel", "info",
# 输入容错 & 自动重连
"-rw_timeout", "15000000",
"-timeout", "10000000",
"-reconnect", "1",
"-reconnect_streamed", "1",
"-reconnect_at_eof", "1",
"-reconnect_delay_max", "30",
# 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",
]
# 平台 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+igndts+discardcorrupt+nobuffer",
"-xerror",
"-use_wallclock_as_timestamps", "1",
"-avioflags", "direct",
"-rtbufsize", "100M",
]
# 视频/音频编码参数（使用 build_youtube_output 的结果）
v_args = v_args or []
a_args = a_args or []
base += v_args + a_args
# 输出（flv 到 YT）
base += ["-f", "flv", out_url]
# 打印命令便于排障（截断）
cmd_str = " ".join(base)
print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...")
logger.info(f"启动推流：{input_url} -> {out_url}")
creationflags = 0
if os.name == "nt":
creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
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():
try:
for line in iter(p.stdout.readline, b''):
if not line:
break
text = line.decode(errors="ignore").strip()
p.last_out_ts = time.time()
# 输出关键统计，便于观察码率/帧率
if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text):
logger.info(f"[ffmpeg统计] {text}")
# 常见错误提示
if "Server disconnected" in text:
logger.error(f"[YouTube/RTMP输出异常] {text}")
elif "not supported" in text or "Invalid" in text or "invalid" in text:
logger.error(f"[转码异常] {text}")
except Exception:
pass
import threading
t = threading.Thread(target=_reader, daemon=True)
t.start()
return p
def stop_ffmpeg(p: Optional[subprocess.Popen]):
if not p:
return
try:
if os.name == "nt":
if p.stdin:
try:
p.stdin.write(b"q")
p.stdin.flush()
except Exception:
pass
p.wait(timeout=6)
else:
p.send_signal(signal.SIGINT)
p.wait(timeout=10)
except Exception:
try:
p.kill()
except Exception:
pass
# --------------------- 主循环（单进程转推+切源） ---------------------
def check_ffmpeg_existence() -> bool:
try:
result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True)
if result.returncode == 0:
print(result.stdout.splitlines()[0])
except Exception:
pass
finally:
return bool(check_ffmpeg())
def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def headers_for_platform(platform_name: str, live_url: str) -> Optional[str]:
live_domain = '/'.join(live_url.split('/')[0:3])
record_headers = {
'PandaTV': 'origin:https://www.pandalive.co.kr',
'WinkTV': 'origin:https://www.winktv.co.kr',
'PopkonTV': 'origin:https://www.popkontv.com',
'FlexTV': 'origin:https://www.flextv.co.kr',
'千度热播': 'referer:https://qiandurebo.com',
'17Live': 'referer:https://17.live/en/live/6302408',
'浪Live': 'referer:https://www.lang.live',
'shopee': f'origin:{live_domain}',
'Blued直播': 'referer:https://app.blued.cn'
}
return record_headers.get(platform_name)
def detect_platform(url: str) -> str:
if "douyin.com" in url: return "抖音直播"
if "tiktok.com" in url: return "TikTok直播"
if url.endswith(".flv") or url.endswith(".m3u8"): return "自定义录制直播"
return "未知平台"
def main():
print("-----------------------------------------------------")
print("| DouyinLiveRecorder - 单进程转推版 |")
print("-----------------------------------------------------")
print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder")
print("模式：只检测 + 单进程推流（停播自动切下一个 URL）")
print(".....................................................")
if not check_ffmpeg_existence():
logger.error("缺少 ffmpeg，程序退出")
sys.exit(1)
urls = parse_urls_from_file(url_config_file)
if not urls:
logger.error("URL_config.ini 中没有可用 URL。")
sys.exit(1)
# 顺序轮询索引
idx = 0
current_url: Optional[str] = None
current_real_url: Optional[str] = None # 实际播放地址（flv/m3u8）
current_proc: Optional[subprocess.Popen] = None
current_platform = ""
current_anchor = ""
record_quality_code = "OD" # 固定“原画”；需要也可改为从 config.ini 读取
check_interval_live = 8 # 推流中检测当前源是否仍在直播的间隔秒
scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒
last_md5 = "" # URL 文件变化时热更新
# watchdog 参数（更稳）
watchdog_grace = 120 # 推流开始 120s 内不做静默判定
watchdog_silence = 600 # 10 分钟完全无输出才重启
refresh_real_url_every = 300 # 每 5 分钟刷新一次真实播放地址，避免 token 过期
try:
while True:
# 动态热更新 URL 列表（URL_config.ini 有变更则重载）
try:
md5 = check_md5(url_config_file)
if md5 != last_md5:
urls = parse_urls_from_file(url_config_file) or urls
last_md5 = md5
print(f"[{now_str()}] 已重载 URL 列表，数量：{len(urls)}")
# 若当前 URL 已不在列表，强制切换
if current_url and current_url not in urls:
print(f"[{now_str()}] 当前URL已被移除，停止并重新选择。")
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
current_real_url = None
except Exception as e:
logger.warning(f"URL 列表热更新失败：{e}")
if current_proc and current_url:
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 = None
current_url = None
current_real_url = None
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 # 容错：无法判断时先认为还活着
if still_live:
elapsed = int(time.time() - start_time)
h, m = divmod(elapsed, 3600)
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="")
# 定期刷新真实播放地址，避免 token 过期
try:
if elapsed >= refresh_real_url_every and (elapsed % refresh_real_url_every) < check_interval_live:
new_port = asyncio.run(fetch_port_info(current_url, record_quality_code))
new_real_url = select_source_url(current_url, new_port)
if new_real_url and new_real_url != current_real_url:
color.print_colored(f"\n[{now_str()}] 刷新播放地址（token 变化），重启 ffmpeg…", color.YELLOW)
stop_ffmpeg(current_proc)
out_url, v_args, a_args = build_youtube_output()
headers = headers_for_platform(current_platform, current_url)
current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args=v_args, a_args=a_args)
current_real_url = new_real_url
start_time = time.time()
time.sleep(check_interval_live)
continue
except Exception as e:
logger.warning(f"刷新播放地址失败：{e}")
# ✅ watchdog（推流开始 > 120s 且 10 分钟完全无输出）
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()}] 超过10分钟未见 ffmpeg 输出，判定异常，重启…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
current_real_url = None
time.sleep(1)
continue
time.sleep(check_interval_live)
continue
else:
color.print_colored(f"\n[{now_str()}] 主播停播，切换下一个源…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
current_real_url = None
else:
# 未在推流：从 urls 顺序找第一个 is_live 的
if not urls:
time.sleep(scan_interval_when_idle)
continue
tried = 0
found = False
start_idx = idx
while tried < len(urls):
url = urls[idx]
idx = (idx + 1) % len(urls)
tried += 1
port = asyncio.run(fetch_port_info(url, record_quality_code))
if not port.get("anchor_name"):
print(f"[{now_str()}] 获取失败，跳过：{url}")
continue
if not port.get("is_live"):
print(f"[{now_str()}] 未开播：{port.get('anchor_name','')} | {url}")
continue
# 选取真实播放地址
real_url = select_source_url(url, port)
if not real_url:
print(f"[{now_str()}] 未得到播放地址，跳过：{url}")
continue
current_platform = detect_platform(url)
current_anchor = port.get("anchor_name", "")
headers = headers_for_platform(current_platform, url)
# 启动 ffmpeg
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
current_real_url = real_url
start_time = time.time()
color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN)
found = True
break
if not found:
# 没有任何源在播，过会儿再试
print(f"[{now_str()}] 当前无可用直播源，等待中…")
time.sleep(scan_interval_when_idle)
continue
except KeyboardInterrupt:
print("\n收到中断，正在退出…")
except Exception as e:
logger.error(f"主循环异常：{e}")
finally:
stop_ffmpeg(current_proc)
print("已退出。")
if __name__ == "__main__":
main()

636
back/mult1 Normal file
View File

@@ -0,0 +1,636 @@
# -*- encoding: utf-8 -*-
"""
多路并行版：
- 同时启动 N 路（N = youtube.ini 的 keys 数量）
- 每路独立 ffmpeg 进程、watchdog、定期刷新抖音真实播放地址
- 共享 URL 池，避免重复占用同一个主播
"""
import asyncio
import os
import sys
import signal
import time
import datetime
import re
import uuid
import threading
import subprocess
from typing import Any, List, Tuple, Optional, Dict, Set
import configparser
import urllib.request
from src import spider, stream
from src.utils import logger, Color, check_md5
from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
# --------------------- 基础环境 ---------------------
version = "relay-parallel-2.0.0"
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'
url_config_file = f'{script_path}/config/URL_config.ini'
yt_ini_file = f"{script_path}/config/youtube.ini"
text_encoding = 'utf-8-sig'
color = Color()
# ---------- YouTube 推流配置（读取 youtube.ini） ----------
_yt_cfg = configparser.ConfigParser()
with open(yt_ini_file, "r", encoding="utf-8-sig") as f:
_yt_cfg.read_file(f)
def _yt_get(section: str, key: str, fallback=None, cast=str):
try:
if cast is bool:
return _yt_cfg.getboolean(section, key, fallback=fallback)
if cast is int:
return _yt_cfg.getint(section, key, fallback=fallback)
return _yt_cfg.get(section, key, fallback=fallback)
except Exception:
return fallback
def _yt_get_list(section: str, key: str) -> List[str]:
raw = _yt_get(section, key, fallback="", cast=str) or ""
# 支持逗号、空白、换行分隔
parts = re.split(r"[,;\s]+", raw.strip())
return [p for p in parts if p]
def _safe_int(v, lo, hi, default):
try:
v = int(v)
if v < lo or v > hi:
return default
return v
except Exception:
return default
def _round_res(w: int, h: int) -> Tuple[int, int]:
# 强制偶数，避免 1088/1918 这类尺寸（H.264 对齐问题）
return (w & ~1, h & ~1)
def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None):
node = (ingest or "a").strip().lower()
node = node if node.isalpha() and len(node) == 1 else "a"
scheme = "rtmps" if secure else "rtmp"
host = f"{node}.rtmps.youtube.com" if secure else f"{node}.rtmp.youtube.com"
return f"{scheme}://{host}/live2/{key}"
# --------------------- 简化配置读取 ---------------------
def read_config_value(cfg: configparser.RawConfigParser, section: str, option: str, default_value: Any) -> Any:
try:
cfg.read(config_file, encoding=text_encoding)
if section not in cfg.sections():
cfg.add_section(section)
return cfg.get(section, option)
except (configparser.NoSectionError, configparser.NoOptionError):
cfg.set(section, option, str(default_value))
with open(config_file, 'w', encoding=text_encoding) as f:
cfg.write(f)
return default_value
_CN_BOOL = {"是": True, "否": False}
cfg = configparser.RawConfigParser()
use_proxy = _CN_BOOL.get(read_config_value(cfg, '录制设置', '是否使用代理ip(是/否)', "否"), False)
proxy_addr = read_config_value(cfg, '录制设置', '代理地址', "")
proxy_addr = proxy_addr if use_proxy and proxy_addr.strip() else None
# Cookie（按需）
dy_cookie = read_config_value(cfg, 'Cookie', '抖音cookie', '')
tiktok_cookie = read_config_value(cfg, 'Cookie', 'tiktok_cookie', '')
# 可选：启动时检测系统代理（仅提示）
try:
if not use_proxy:
urllib.request.urlopen("https://www.google.com/", timeout=5)
print("提示：你的网络似乎直连可用（无需代理）。")
except Exception:
if not use_proxy:
color.print_colored("提示：无法直连海外站，如需 TikTok 请在 config.ini 开启代理并填写代理地址。", color.YELLOW)
# --------------------- URL 列表读取 ---------------------
def parse_urls_from_file(path: str) -> List[str]:
urls, seen = [], set()
if not os.path.isfile(path):
open(path, 'w', encoding=text_encoding).close()
print("URL_config.ini 为空，请写入直播间地址后再运行。")
return urls
with open(path, "r", encoding=text_encoding, errors='ignore') as f:
for line in f:
s = line.strip()
if not s or s.startswith("#"):
continue
# 拆分（最多3段，取包含 https 或域名的那段）
parts = re.split(r"[，,]\s*", s)
cand = None
for p in parts:
if '://' in p or re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", p):
cand = p
break
if not cand:
continue
url = cand if '://' in cand else f"https://{cand}"
if url not in seen:
seen.add(url)
urls.append(url)
return urls
# --------------------- 直播解析与选源 ---------------------
def is_flv_preferred_platform(link: str) -> bool:
return any(i in link for i in ["douyin", "tiktok"])
def select_source_url(link: str, stream_info: dict) -> Optional[str]:
if not stream_info:
return None
if is_flv_preferred_platform(link):
flv = stream_info.get("flv_url")
if flv:
try:
from urllib.parse import urlparse, parse_qs
qs = parse_qs(urlparse(flv).query or "")
if (qs.get("codec") or [""])[0].lower() == "h265":
logger.warning("FLV 不支持 h265，改用 HLS。")
else:
return flv
except Exception:
return flv
return stream_info.get("record_url") or stream_info.get("m3u8_url") or stream_info.get("flv_url")
async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
proxy = proxy_addr
port_info = {}
try:
if "douyin.com/" in url:
if 'v.douyin.com' not in url and '/user/' not in url:
j = await spider.get_douyin_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie)
else:
j = await spider.get_douyin_app_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie)
port_info = await stream.get_douyin_stream_url(j, record_quality_code, proxy)
elif url.startswith("https://www.tiktok.com/"):
if proxy:
j = await spider.get_tiktok_stream_data(url=url, proxy_addr=proxy, cookies=tiktok_cookie)
port_info = await stream.get_tiktok_stream_url(j, record_quality_code, proxy)
else:
logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。")
port_info = {"anchor_name": "", "is_live": False}
elif (".m3u8" in url) or (".flv" in url):
port_info = {
"anchor_name": "自定义_" + str(uuid.uuid4())[:8],
"is_live": True,
}
if url.endswith(".flv"):
port_info["flv_url"] = url
else:
port_info["m3u8_url"] = url
else:
port_info = {"anchor_name": "", "is_live": False}
except Exception as e:
logger.error(f"解析失败: {e}")
port_info = {"anchor_name": "", "is_live": False}
return port_info
# --------------------- ffmpeg 输出参数生成（可复用） ---------------------
def build_youtube_output_for_key(key: str, resolution: Optional[str]) -> Tuple[str, list, list]:
if not key:
logger.error("错误: youtube.ini 未配置 key/keys")
sys.exit(1)
secure = _yt_get("youtube", "secure", fallback=False, cast=bool)
ingest = _yt_get("youtube", "ingest", fallback="a", cast=str)
res_str = (resolution or _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)
out_url = _build_youtube_rtmp_url(key, secure, ingest)
v_args = [
"-c:v", "libx264",
"-preset", "veryfast",
"-tune", "zerolatency",
"-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",
]
if res_str:
try:
w, h = map(int, res_str.lower().split("x"))
w, h = _round_res(w, h)
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
# --------------------- 启动 ffmpeg ---------------------
def launch_ffmpeg(input_url: str,
out_url: str,
extra_headers: Optional[str],
v_args: list,
a_args: list) -> subprocess.Popen:
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")
base = [
"ffmpeg", "-nostdin", "-hide_banner",
"-stats", "-stats_period", "60",
"-progress", "pipe:1",
"-loglevel", "info",
# 输入容错 & 自动重连
"-rw_timeout", "15000000",
"-timeout", "10000000",
"-reconnect", "1",
"-reconnect_streamed", "1",
"-reconnect_at_eof", "1",
"-reconnect_delay_max", "30",
"-user_agent", user_agent,
"-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy",
"-thread_queue_size", "4096",
"-analyzeduration", "10000000",
"-probesize", "10000000",
]
if extra_headers:
base += ["-headers", extra_headers]
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+igndts+discardcorrupt+nobuffer",
"-xerror",
"-use_wallclock_as_timestamps", "1",
"-avioflags", "direct",
"-rtbufsize", "100M",
]
base += v_args + a_args
base += ["-f", "flv", out_url]
cmd_str = " ".join(base)
print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...")
logger.info(f"启动推流：{input_url} -> {out_url}")
creationflags = 0
if os.name == "nt":
creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
p = subprocess.Popen(
base,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=0,
env=env,
creationflags=creationflags
)
p.last_out_ts = time.time()
def _reader():
try:
for line in iter(p.stdout.readline, b''):
if not line:
break
text = line.decode(errors="ignore").strip()
p.last_out_ts = time.time()
if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text):
logger.info(f"[ffmpeg统计] {text}")
if "Server disconnected" in text:
logger.error(f"[输出异常] {text}")
elif "not supported" in text or "Invalid" in text or "invalid" in text:
logger.error(f"[转码异常] {text}")
except Exception:
pass
threading.Thread(target=_reader, daemon=True).start()
return p
def stop_ffmpeg(p: Optional[subprocess.Popen]):
if not p:
return
try:
if os.name == "nt":
if p.stdin:
try: p.stdin.write(b"q"); p.stdin.flush()
except Exception: pass
p.wait(timeout=6)
else:
p.send_signal(signal.SIGINT)
p.wait(timeout=10)
except Exception:
try: p.kill()
except Exception: pass
# --------------------- 平台头部 ---------------------
def headers_for_platform(platform_name: str, live_url: str) -> Optional[str]:
live_domain = '/'.join(live_url.split('/')[0:3])
record_headers = {
'PandaTV': 'origin:https://www.pandalive.co.kr',
'WinkTV': 'origin:https://www.winktv.co.kr',
'PopkonTV': 'origin:https://www.popkontv.com',
'FlexTV': 'origin:https://www.flextv.co.kr',
'千度热播': 'referer:https://qiandurebo.com',
'17Live': 'referer:https://17.live/en/live/6302408',
'浪Live': 'referer:https://www.lang.live',
'shopee': f'origin:{live_domain}',
'Blued直播': 'referer:https://app.blued.cn'
}
return record_headers.get(platform_name)
def detect_platform(url: str) -> str:
if "douyin.com" in url: return "抖音直播"
if "tiktok.com" in url: return "TikTok直播"
if url.endswith(".flv") or url.endswith(".m3u8"): return "自定义录制直播"
return "未知平台"
# --------------------- 并行协调器 ---------------------
class UrlCoordinator:
""" 管理 URL 池、热更新、分配给不同 worker，避免重复占用 """
def __init__(self, path: str):
self.path = path
self.urls: List[str] = parse_urls_from_file(path)
self.idx: int = 0
self.in_use: Set[str] = set()
self.lock = threading.Lock()
self.last_md5 = check_md5(path)
def refresh_if_changed(self):
try:
md5 = check_md5(self.path)
if md5 != self.last_md5:
self.urls = parse_urls_from_file(self.path) or self.urls
self.last_md5 = md5
print(f"[{now_str()}] 已重载 URL 列表，数量：{len(self.urls)}")
except Exception as e:
logger.warning(f"URL 列表热更新失败：{e}")
def acquire_next_live(self, quality_code: str = "OD") -> Optional[Tuple[str, dict]]:
""" 顺序扫描，跳过 in_use，找到正在直播的返回并占用 """
with self.lock:
if not self.urls:
return None
start = self.idx
tried = 0
while tried < len(self.urls):
url = self.urls[self.idx]
self.idx = (self.idx + 1) % len(self.urls)
tried += 1
if url in self.in_use:
continue
# 释放锁去解析，避免长时间占用
pass_url = url
break
else:
return None
# 锁外解析
port = asyncio.run(fetch_port_info(pass_url, quality_code))
if not port.get("anchor_name") or not port.get("is_live"):
return None
with self.lock:
# 二次检查，确保没被别人抢占
if pass_url in self.in_use:
return None
self.in_use.add(pass_url)
return pass_url, port
def release(self, url: Optional[str]):
if not url:
return
with self.lock:
self.in_use.discard(url)
def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# --------------------- Worker ---------------------
class StreamWorker(threading.Thread):
def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator):
super().__init__(daemon=True)
self.worker_id = worker_id
self.key = key
self.resolution = resolution
self.coordinator = coordinator
self.current_url: Optional[str] = None
self.current_real_url: Optional[str] = None
self.current_proc: Optional[subprocess.Popen] = None
self.current_platform = ""
self.current_anchor = ""
self.record_quality_code = "OD"
self.watchdog_grace = 120
self.watchdog_silence = 600
self.refresh_real_url_every = 300
self.check_interval_live = 8
self.scan_interval_when_idle = 2
self._stop_event = threading.Event()
def stop(self):
self._stop_event.set()
def run(self):
color.print_colored(f"[Worker-{self.worker_id}] 启动", color.GREEN)
try:
while not self._stop_event.is_set():
# 刷新 URL 文件
self.coordinator.refresh_if_changed()
if self.current_proc and self.current_url:
retcode = self.current_proc.poll()
if retcode is not None:
color.print_colored(f"[{now_str()}][Worker-{self.worker_id}] ffmpeg 退出(code={retcode})，释放并切源…", color.YELLOW)
stop_ffmpeg(self.current_proc)
self.coordinator.release(self.current_url)
self._clear_state()
continue
# 仍在推流：检查主播是否仍在直播
still_live = True
try:
port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code))
still_live = bool(port.get("is_live"))
except Exception as e:
logger.error(f"[Worker-{self.worker_id}] 检测当前源失败：{e}")
still_live = True
if still_live:
elapsed = int(time.time() - self.start_time)
h, m = divmod(elapsed, 3600)
m, s = divmod(m, 60)
dur_str = f"{h:02d}:{m:02d}:{s:02d}"
print(f"\r[{now_str()}][Worker-{self.worker_id}] 正在转播({dur_str})：{self.current_platform} | {self.current_anchor} | {self.current_url}", end="")
# 定期刷新真实播放地址
try:
if elapsed >= self.refresh_real_url_every and (elapsed % self.refresh_real_url_every) < self.check_interval_live:
new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code))
new_real_url = select_source_url(self.current_url, new_port)
if new_real_url and new_real_url != self.current_real_url:
color.print_colored(f"\n[{now_str()}][Worker-{self.worker_id}] 刷新播放地址（token 变化），重启 ffmpeg…", color.YELLOW)
stop_ffmpeg(self.current_proc)
out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution)
headers = headers_for_platform(self.current_platform, self.current_url)
self.current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args, a_args)
self.current_real_url = new_real_url
self.start_time = time.time()
time.sleep(self.check_interval_live)
continue
except Exception as e:
logger.warning(f"[Worker-{self.worker_id}] 刷新播放地址失败：{e}")
# watchdog：心跳超时才重启
silence = time.time() - getattr(self.current_proc, "last_out_ts", 0)
if elapsed >= self.watchdog_grace and silence > self.watchdog_silence:
color.print_colored(f"\n[{now_str()}][Worker-{self.worker_id}] 超过10分钟未见输出，重启…", color.YELLOW)
stop_ffmpeg(self.current_proc)
self._clear_state(release=True)
time.sleep(1)
continue
time.sleep(self.check_interval_live)
continue
else:
color.print_colored(f"\n[{now_str()}][Worker-{self.worker_id}] 主播停播，释放并切源…", color.YELLOW)
stop_ffmpeg(self.current_proc)
self._clear_state(release=True)
continue
# 未在推流：尝试分配新的直播 URL
got = self.coordinator.acquire_next_live(self.record_quality_code)
if not got:
print(f"[{now_str()}][Worker-{self.worker_id}] 无可用直播源，稍后再试…")
time.sleep(self.scan_interval_when_idle)
continue
url, port = got
real_url = select_source_url(url, port)
if not real_url:
print(f"[{now_str()}][Worker-{self.worker_id}] 未得到播放地址，跳过：{url}")
self.coordinator.release(url)
time.sleep(1)
continue
self.current_platform = detect_platform(url)
self.current_anchor = port.get("anchor_name", "")
headers = headers_for_platform(self.current_platform, url)
out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution)
self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args)
self.current_url = url
self.current_real_url = real_url
self.start_time = time.time()
color.print_colored(f"[{now_str()}][Worker-{self.worker_id}] 开始转播：{self.current_platform} | {self.current_anchor} | {url}", color.GREEN)
except Exception as e:
logger.error(f"[Worker-{self.worker_id}] 主循环异常：{e}")
finally:
stop_ffmpeg(self.current_proc)
self.coordinator.release(self.current_url)
print(f"[Worker-{self.worker_id}] 已退出。")
def _clear_state(self, release: bool = False):
if release and self.current_url:
self.coordinator.release(self.current_url)
self.current_url = None
self.current_real_url = None
self.current_proc = None
self.current_platform = ""
self.current_anchor = ""
# --------------------- 启动入口 ---------------------
def check_ffmpeg_existence() -> bool:
try:
result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True)
if result.returncode == 0:
print(result.stdout.splitlines()[0])
except Exception:
pass
finally:
return bool(check_ffmpeg())
def main():
print("-----------------------------------------------------")
print("| DouyinLiveRecorder - 多路并行转推版 |")
print("-----------------------------------------------------")
print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder")
print("模式：并行检测 + 并行推流（每路独立 watchdog）")
print(".....................................................")
if not check_ffmpeg_existence():
logger.error("缺少 ffmpeg，程序退出")
sys.exit(1)
urls = parse_urls_from_file(url_config_file)
if not urls:
logger.error("URL_config.ini 中没有可用 URL。")
sys.exit(1)
keys = _yt_get_list("youtube", "keys")
if not keys:
# 兼容单 key 配置
single_key = _yt_get("youtube", "key", fallback="", cast=str)
if single_key:
keys = [single_key]
if not keys:
logger.error("youtube.ini 没有配置 [youtube].keys 或 key")
sys.exit(1)
# 可选：为每个 key 指定不同分辨率（与 keys 顺序对应），否则使用全局 resolution
per_res = _yt_get_list("youtube", "resolutions") # 例如：1920x1080,1080x1920,1280x720
while per_res and len(per_res) < len(keys):
per_res.append(per_res[-1]) # 不足时沿用最后一个
# 协调器（共享 URL 池）
coordinator = UrlCoordinator(url_config_file)
# 启动多路 worker
workers: List[StreamWorker] = []
for i, key in enumerate(keys, start=1):
res_for_this = per_res[i-1] if per_res else None
w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator)
w.start()
workers.append(w)
# 主线程仅负责捕获 Ctrl+C 并优雅退出
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\n收到中断，正在退出…")
finally:
for w in workers:
w.stop()
for w in workers:
w.join(timeout=10)
print("全部 worker 已退出。")
if __name__ == "__main__":
main()

34
config/URL_config.ini
View File

@@ -1,6 +1,8 @@
https://live.douyin.com/60379224194
https://live.douyin.com/72212955083
https://live.douyin.com/621118100231
https://live.douyin.com/347310028364
https://live.douyin.com/280422307519
https://live.douyin.com/651293616593
https://live.douyin.com/114111936890
https://live.douyin.com/43665279648
https://live.douyin.com/122383435921
@@ -8,7 +10,10 @@ https://live.douyin.com/92876979328
https://live.douyin.com/136996589868
https://live.douyin.com/517715534931
https://live.douyin.com/460525712926
https://live.douyin.com/456576000931
https://live.douyin.com/745606325880
https://live.douyin.com/749687541944
https://live.douyin.com/675343045974
https://live.douyin.com/673565298571
https://live.douyin.com/835571459859
https://live.douyin.com/152358755212
@@ -18,4 +23,29 @@ https://live.douyin.com/967381081829
https://live.douyin.com/15652099787
https://live.douyin.com/78012762575
https://live.douyin.com/80248655914
https://live.douyin.com/163813589919
https://live.douyin.com/689189188452
https://live.douyin.com/498368994814
https://live.douyin.com/617848099204
https://live.douyin.com/821525628944
https://live.douyin.com/993102287144
https://live.douyin.com/701547125568
https://live.douyin.com/816130992220
https://live.douyin.com/876468215361
https://live.douyin.com/291403201249
https://live.douyin.com/511335278313
https://live.douyin.com/483160615952
https://live.douyin.com/673565298571
https://live.douyin.com/525514386431
https://live.douyin.com/8687122573
https://live.douyin.com/642534242822
https://live.douyin.com/743565594721
https://live.douyin.com/249578288248
https://live.douyin.com/472140253414
https://live.douyin.com/700846653732
https://live.douyin.com/81849868631
https://live.douyin.com/690114366322
https://live.douyin.com/642534242822
https://live.douyin.com/469980190666
https://live.douyin.com/388066418744
https://live.douyin.com/122383435921

3
config/youtube.ini
View File

@@ -1,5 +1,6 @@
[youtube]
key = x04z-564w-aks7-embw-30y4
keys = x04z-564w-aks7-embw-30y4,qxvb-r47b-r5ju-6ud3-6k7z
secure = true
ingest = a
bitrate = 5000
@@ -7,5 +8,5 @@ audio_bitrate = 128
fps = 30
gop = 60
copy_when_safe = true
resolution = 1080x1920
resolution = 720x1280
single_mode = true

687
main.py
View File

@@ -1,9 +1,12 @@
# -*- encoding: utf-8 -*-
"""
单进程 ffmpeg 推流：从 URL_config.ini 顺序检测直播源，
一旦当前源停播/异常，自动切到下一个正在直播的源并推流到 YouTube。
仅做“检测+转推”，不做录制、分段、通知等。
并行转推（日志优化版，保持原业务逻辑）
- 保留 Worker-1/2 等前缀
- 日志状态变化才打印，避免刷屏
- 禁止同一个 URL 推到同一个 YouTube key（(url,key) 级别防重复）
- 仍然避免同一个 URL 被多路并行占用（url 级别独占）
- 新增：Ctrl+C/TERM 时优雅停止所有 Worker 与 ffmpeg 子进程（仅最小增改）
"""
import asyncio
@@ -14,31 +17,36 @@ import time
import datetime
import re
import uuid
import threading
import subprocess
from pathlib import Path
from typing import Any, List, Tuple, Optional
from typing import Any, List, Tuple, Optional, Dict, Set
import configparser
import urllib.request
from urllib.error import URLError, HTTPError
import httpx
from src import spider, stream
from src.utils import logger, Color, check_md5
from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
# --------------------- 基础环境 ---------------------
version = "relay-1.1.0"
# =============== 基础环境 ===============
version = "relay-parallel-2.0.1-logonly"
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'
url_config_file = f'{script_path}/config/URL_config.ini'
yt_ini_file = f"{script_path}/config/youtube.ini"
text_encoding = 'utf-8-sig'
color = Color()
# ---------- YouTube 推流配置（读取 youtube.ini） ----------
# ---- 新增：全局 Worker & 进程注册（用于 Ctrl+C 干净退出） ----
ALL_WORKERS: List["StreamWorker"] = []
ALL_PROCS: Set[subprocess.Popen] = set()
SHUTDOWN_EVENT = threading.Event()
ALL_LOCK = threading.Lock()
# =============== 读取 YouTube 配置 ===============
_yt_cfg = configparser.ConfigParser()
with open(f"{script_path}/config/youtube.ini", "r", encoding="utf-8-sig") as f:
with open(yt_ini_file, "r", encoding="utf-8-sig") as f:
_yt_cfg.read_file(f)
def _yt_get(section: str, key: str, fallback=None, cast=str):
@@ -51,17 +59,22 @@ def _yt_get(section: str, key: str, fallback=None, cast=str):
except Exception:
return fallback
def _safe_int(v, lo, hi, default):
def _yt_get_list(section: str, key: str) -> List[str]:
raw = _yt_get(section, key, fallback="", cast=str) or ""
parts = re.split(r"[,;\s]+", raw.strip())
return [p for p in parts if p]
def _safe_int(v, lo, hi, defv):
try:
v = int(v)
if v < lo or v > hi:
return default
if v < lo or v > hi: return defv
return v
except Exception:
return default
return defv
# =============== 工具函数 ===============
def _round_res(w: int, h: int) -> Tuple[int, int]:
# 强制偶数，避免 1088/1918 这类非标准尺寸
# 偶数对齐，避免 1088x1920 类问题
return (w & ~1, h & ~1)
def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None):
@@ -71,7 +84,10 @@ def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None):
host = f"{node}.rtmps.youtube.com" if secure else f"{node}.rtmp.youtube.com"
return f"{scheme}://{host}/live2/{key}"
# --------------------- 简化配置读取 ---------------------
def now_str():
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# =============== 读取运行配置 ===============
def read_config_value(cfg: configparser.RawConfigParser, section: str, option: str, default_value: Any) -> Any:
try:
cfg.read(config_file, encoding=text_encoding)
@@ -85,16 +101,14 @@ def read_config_value(cfg: configparser.RawConfigParser, section: str, option: s
return default_value
_CN_BOOL = {"是": True, "否": False}
cfg = configparser.RawConfigParser()
use_proxy = _CN_BOOL.get(read_config_value(cfg, '录制设置', '是否使用代理ip(是/否)', "否"), False)
proxy_addr = read_config_value(cfg, '录制设置', '代理地址', "")
proxy_addr = proxy_addr if use_proxy and proxy_addr.strip() else None
# Cookie（按需）
dy_cookie = read_config_value(cfg, 'Cookie', '抖音cookie', '')
tiktok_cookie = read_config_value(cfg, 'Cookie', 'tiktok_cookie', '')
# 可选：启动时检测系统代理（仅提示）
# 启动时连通性提示
try:
if not use_proxy:
urllib.request.urlopen("https://www.google.com/", timeout=5)
@@ -103,28 +117,18 @@ except Exception:
if not use_proxy:
color.print_colored("提示：无法直连海外站，如需 TikTok 请在 config.ini 开启代理并填写代理地址。", color.YELLOW)
# --------------------- URL 列表读取 ---------------------
# =============== URL 列表 ===============
def parse_urls_from_file(path: str) -> List[str]:
"""
读取 URL_config.ini，返回去重且保持相对顺序的 URL 列表。
支持行内格式：
- 纯 URL
- quality,url
- url,主播: xxx
- # 注释行 忽略
"""
urls, seen = [], set()
if not os.path.isfile(path):
open(path, 'w', encoding=text_encoding).close()
print("URL_config.ini 为空，请写入直播间地址后再运行。")
return urls
with open(path, "r", encoding=text_encoding, errors='ignore') as f:
for line in f:
s = line.strip()
if not s or s.startswith("#"):
if not s or s.startswith("#"):
continue
# 拆分（最多3段，取包含 https 或域名的那段）
parts = re.split(r"[，,]\s*", s)
cand = None
for p in parts:
@@ -139,9 +143,8 @@ def parse_urls_from_file(path: str) -> List[str]:
urls.append(url)
return urls
# --------------------- 直播解析与选源 ---------------------
# =============== 解析与选源 ===============
def is_flv_preferred_platform(link: str) -> bool:
# 抖音/TikTok 优先走 FLV；但 H265 不走 FLV
return any(i in link for i in ["douyin", "tiktok"])
def select_source_url(link: str, stream_info: dict) -> Optional[str]:
@@ -150,7 +153,6 @@ def select_source_url(link: str, stream_info: dict) -> Optional[str]:
if is_flv_preferred_platform(link):
flv = stream_info.get("flv_url")
if flv:
# 若 codec=h265 则不走 flv
try:
from urllib.parse import urlparse, parse_qs
qs = parse_qs(urlparse(flv).query or "")
@@ -163,17 +165,6 @@ def select_source_url(link: str, stream_info: dict) -> Optional[str]:
return stream_info.get("record_url") or stream_info.get("m3u8_url") or stream_info.get("flv_url")
async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
"""
调用项目现有 spider/stream，返回：
{
"anchor_name": str,
"is_live": bool,
"title": str?,
"record_url": "...",
"flv_url": "...",
"m3u8_url": "..."
}
"""
proxy = proxy_addr
port_info = {}
try:
@@ -191,7 +182,6 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。")
port_info = {"anchor_name": "", "is_live": False}
elif (".m3u8" in url) or (".flv" in url):
# 自定义直链
port_info = {
"anchor_name": "自定义_" + str(uuid.uuid4())[:8],
"is_live": True,
@@ -201,23 +191,21 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
else:
port_info["m3u8_url"] = url
else:
# 其它站点可按需扩展
port_info = {"anchor_name": "", "is_live": False}
except Exception as e:
logger.error(f"解析失败: {e}")
port_info = {"anchor_name": "", "is_live": False}
return port_info
# --------------------- ffmpeg 单进程推流 ---------------------
def build_youtube_output() -> Tuple[str, list, list]:
key = _yt_get("youtube", "key", fallback="", cast=str)
# =============== 构建输出参数（每个 key） ===============
def build_youtube_output_for_key(key: str, resolution: Optional[str]) -> Tuple[str, list, list]:
if not key:
logger.error("错误: config/youtube.ini 未配置 key")
logger.error("错误: youtube.ini 未配置 key/keys")
sys.exit(1)
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() # 例如 "1080x1920"
res_str = (resolution or _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)
@@ -226,7 +214,6 @@ def build_youtube_output() -> Tuple[str, list, list]:
out_url = _build_youtube_rtmp_url(key, secure, ingest)
# —— 默认强制转码（稳） —— #
v_args = [
"-c:v", "libx264",
"-preset", "veryfast",
@@ -247,7 +234,6 @@ def build_youtube_output() -> Tuple[str, list, list]:
"-ac", "2",
]
# 可选缩放/留黑边 + 避免 1088x1920（crop 强制精确尺寸）+ 偶数对齐
if res_str:
try:
w, h = map(int, res_str.lower().split("x"))
@@ -264,27 +250,19 @@ def build_youtube_output() -> Tuple[str, list, list]:
return out_url, v_args, a_args
# =============== 启动/停止 ffmpeg ===============
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/RTMPS 到 out_url。
- v_args / a_args 由 build_youtube_output() 生成并传入
- 启动 stdout reader 线程以防缓冲阻塞
- 设置代理环境变量以兼容 https/hls
"""
import threading
extra_headers: Optional[str],
v_args: list,
a_args: list,
worker_name: str) -> subprocess.Popen:
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")
# 输入侧容错/重连/探测
base = [
"ffmpeg", "-nostdin", "-hide_banner",
"-stats",
"-stats_period", "60",
"-stats", "-stats_period", "60",
"-progress", "pipe:1",
"-loglevel", "info",
@@ -296,31 +274,24 @@ def launch_ffmpeg(input_url: str,
"-reconnect_at_eof", "1",
"-reconnect_delay_max", "30",
# 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",
]
# 平台 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+igndts+discardcorrupt+nobuffer",
"-xerror",
@@ -329,19 +300,12 @@ def launch_ffmpeg(input_url: str,
"-rtbufsize", "100M",
]
# 视频/音频编码参数（使用 build_youtube_output 的结果）
v_args = v_args or []
a_args = a_args or []
base += v_args + a_args
# 输出（flv 到 YT）
base += ["-f", "flv", out_url]
# 打印命令便于排障（截断）
cmd_str = " ".join(base)
print("ffmpeg cmd:", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...")
logger.info(f"启动推流：{input_url} -> {out_url}")
print(f"ffmpeg cmd ({worker_name}):", cmd_str if len(cmd_str) < 500 else cmd_str[:500] + " ...")
logger.info(f"{worker_name} 启动推流：{input_url} -> {out_url}")
creationflags = 0
if os.name == "nt":
@@ -352,14 +316,16 @@ def launch_ffmpeg(input_url: str,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=0, # 立即刷新，避免行缓冲告警
bufsize=0,
env=env,
creationflags=creationflags
)
# 最近输出时间戳，用于 watchdog
p.last_out_ts = time.time()
# ---- 新增：注册进程，便于 Ctrl+C 统一清理 ----
with ALL_LOCK:
ALL_PROCS.add(p)
def _reader():
try:
for line in iter(p.stdout.readline, b''):
@@ -368,33 +334,26 @@ def launch_ffmpeg(input_url: str,
text = line.decode(errors="ignore").strip()
p.last_out_ts = time.time()
# 输出关键统计，便于观察码率/帧率
if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text):
logger.info(f"[ffmpeg统计] {text}")
# 常见错误提示
# 只保留关键统计/错误（前缀带 Worker）
if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text) or ("libx264" in text):
logger.info(f"{worker_name} [ffmpeg统计] {text}")
if "Server disconnected" in text:
logger.error(f"[YouTube/RTMP输出异常] {text}")
elif "not supported" in text or "Invalid" in text or "invalid" in text:
logger.error(f"[转码异常] {text}")
logger.error(f"{worker_name} [输出异常] {text}")
elif ("not supported" in text) or ("Invalid" in text) or ("invalid" in text):
logger.error(f"{worker_name} [转码异常] {text}")
except Exception:
pass
import threading
t = threading.Thread(target=_reader, daemon=True)
t.start()
threading.Thread(target=_reader, daemon=True).start()
return p
def stop_ffmpeg(p: Optional[subprocess.Popen]):
if not p:
return
if not p: return
try:
if os.name == "nt":
if p.stdin:
try:
p.stdin.write(b"q")
p.stdin.flush()
p.stdin.write(b"q"); p.stdin.flush()
except Exception:
pass
p.wait(timeout=6)
@@ -402,24 +361,14 @@ def stop_ffmpeg(p: Optional[subprocess.Popen]):
p.send_signal(signal.SIGINT)
p.wait(timeout=10)
except Exception:
try:
p.kill()
except Exception:
pass
# --------------------- 主循环（单进程转推+切源） ---------------------
def check_ffmpeg_existence() -> bool:
try:
result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True)
if result.returncode == 0:
print(result.stdout.splitlines()[0])
except Exception:
pass
try: p.kill()
except Exception: pass
finally:
return bool(check_ffmpeg())
def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# ---- 新增：从全局集合移除 ----
with ALL_LOCK:
ALL_PROCS.discard(p)
# =============== 平台 headers/识别 ===============
def headers_for_platform(platform_name: str, live_url: str) -> Optional[str]:
live_domain = '/'.join(live_url.split('/')[0:3])
record_headers = {
@@ -441,12 +390,309 @@ def detect_platform(url: str) -> str:
if url.endswith(".flv") or url.endswith(".m3u8"): return "自定义录制直播"
return "未知平台"
# =============== 日志状态缓存（去重/限频） ===============
class LogState:
def __init__(self):
self.lock = threading.Lock()
self.url_status: Dict[str, str] = {} # url -> "unknown/offline/online/streaming"
self.no_source_worker: Dict[str, bool] = {} # worker_name -> idle printed flag
def set_status(self, url: str, status: str, text: Optional[str] = None, prefix: str = ""):
with self.lock:
prev = self.url_status.get(url, "unknown")
if prev != status:
self.url_status[url] = status
if text:
print(f"[{now_str()}]{prefix} {text}")
def mark_idle_once(self, worker_name: str):
with self.lock:
if not self.no_source_worker.get(worker_name, False):
print(f"[{now_str()}][{worker_name}] 当前无可用直播源，等待中…")
self.no_source_worker[worker_name] = True
def clear_idle_flag(self, worker_name: str):
with self.lock:
self.no_source_worker[worker_name] = False
LOGSTATE = LogState()
# =============== URL 协调器（包含 url 与 (url,key) 占用） ===============
class UrlCoordinator:
"""
- urls 顺序分配
- in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用）
- in_use_pairs: (url,key) 级别独占（禁止同一 url 推到同一个 key）
"""
def __init__(self, path: str):
self.path = path
self.urls: List[str] = parse_urls_from_file(path)
self.idx: int = 0
self.lock = threading.Lock()
self.last_md5 = check_md5(path)
self.in_use_urls: Set[str] = set()
self.in_use_pairs: Set[Tuple[str, str]] = set() # (url, key)
def refresh_if_changed(self):
try:
md5 = check_md5(self.path)
if md5 != self.last_md5:
self.urls = parse_urls_from_file(self.path) or self.urls
self.last_md5 = md5
print(f"[{now_str()}] 已重载 URL 列表，数量：{len(self.urls)}")
except Exception as e:
logger.warning(f"URL 列表热更新失败：{e}")
def acquire_next_live(self, worker_key: str, quality_code: str = "OD") -> Optional[Tuple[str, dict]]:
with self.lock:
if not self.urls:
return None
tried = 0
pick = None
while tried < len(self.urls):
url = self.urls[self.idx]
self.idx = (self.idx + 1) % len(self.urls)
tried += 1
# 已被其他路占用 / 同 url-key 已经在用
if url in self.in_use_urls:
continue
if (url, worker_key) in self.in_use_pairs:
continue
pick = url
break
if not pick:
return None
# 锁外解析
port = asyncio.run(fetch_port_info(pick, quality_code))
if not port.get("anchor_name"):
LOGSTATE.set_status(pick, "unknown",
text=f"解析失败：{pick}",
prefix="")
return None
if not port.get("is_live"):
anchor = port.get('anchor_name','')
LOGSTATE.set_status(pick, "offline",
text=f"未开播：{anchor} | {pick}",
prefix="")
return None
# 占用
with self.lock:
if (pick in self.in_use_urls) or ((pick, worker_key) in self.in_use_pairs):
return None
self.in_use_urls.add(pick)
self.in_use_pairs.add((pick, worker_key))
# 状态更新（online）
anchor = port.get('anchor_name','')
LOGSTATE.set_status(pick, "online",
text=f"已开播：{anchor} | {pick}",
prefix="")
return pick, port
def release(self, url: Optional[str], worker_key: Optional[str]):
if not url:
return
with self.lock:
self.in_use_urls.discard(url)
if worker_key:
self.in_use_pairs.discard((url, worker_key))
# =============== Worker（保留前缀日志） ===============
class StreamWorker(threading.Thread):
def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator):
super().__init__(daemon=True)
self.worker_id = worker_id
self.key = key
self.worker_name = f"Worker-{worker_id}"
self.resolution = resolution
self.coordinator = coordinator
self.current_url: Optional[str] = None
self.current_real_url: Optional[str] = None
self.current_proc: Optional[subprocess.Popen] = None
self.current_platform = ""
self.current_anchor = ""
self.record_quality_code = "OD"
# 与原逻辑一致的参数
self.watchdog_grace = 120
self.watchdog_silence = 600
self.refresh_real_url_every = 300
self.check_interval_live = 8
self.scan_interval_when_idle = 2
self._last_progress_print = -9999
self._stop_event = threading.Event()
def stop(self):
"""最小改动：收到停止请求时，立即尝试停掉当前 ffmpeg，尽快响应退出。"""
self._stop_event.set()
try:
stop_ffmpeg(self.current_proc)
except Exception:
pass
def _p(self, text: str):
print(f"[{now_str()}][{self.worker_name}] {text}")
def run(self):
self._p("启动")
try:
while not self._stop_event.is_set():
self.coordinator.refresh_if_changed()
if self.current_proc and self.current_url:
retcode = self.current_proc.poll()
if retcode is not None:
self._p(f"❌ ffmpeg 退出(code={retcode})，切换下一个源…")
stop_ffmpeg(self.current_proc)
self.coordinator.release(self.current_url, self.key)
self._clear_state()
LOGSTATE.clear_idle_flag(self.worker_name)
continue
# 检测是否仍在直播（保持原逻辑）
still_live = True
try:
port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code))
still_live = bool(port.get("is_live"))
except Exception as e:
logger.error(f"[{self.worker_name}] 检测当前源失败：{e}")
still_live = True
if still_live:
elapsed = int(time.time() - self.start_time)
# 每 60s 打印一次“正在转播”
if elapsed - self._last_progress_print >= 60:
h, m = divmod(elapsed, 3600)
m, s = divmod(m, 60)
dur_str = f"{h:02d}:{m:02d}:{s:02d}"
self._p(f"⏱ 正在转播({dur_str})：{self.current_platform} | {self.current_anchor}")
self._last_progress_print = elapsed
# 定期刷新真实播放地址（token）
try:
if elapsed >= self.refresh_real_url_every and (elapsed % self.refresh_real_url_every) < self.check_interval_live:
new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code))
new_real_url = select_source_url(self.current_url, new_port)
if new_real_url and new_real_url != self.current_real_url:
self._p("🔄 刷新播放地址（token 变化），重启 ffmpeg…")
stop_ffmpeg(self.current_proc)
out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution)
headers = headers_for_platform(self.current_platform, self.current_url)
self.current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args, a_args, self.worker_name)
self.current_real_url = new_real_url
self.start_time = time.time()
self._last_progress_print = -9999
time.sleep(self.check_interval_live)
continue
except Exception as e:
logger.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}")
# watchdog
silence = time.time() - getattr(self.current_proc, "last_out_ts", 0)
if elapsed >= self.watchdog_grace and silence > self.watchdog_silence:
self._p("⚠️ 超过10分钟未见输出，重启…")
stop_ffmpeg(self.current_proc)
self._clear_state(release=True)
LOGSTATE.clear_idle_flag(self.worker_name)
time.sleep(1)
continue
time.sleep(self.check_interval_live)
continue
else:
self._p("📴 主播停播，切换下一个源…")
stop_ffmpeg(self.current_proc)
self._clear_state(release=True)
LOGSTATE.clear_idle_flag(self.worker_name)
continue
# —— 未在推流：尝试分配一个新源（保持原顺序轮询 + 占用） ——
got = self.coordinator.acquire_next_live(self.key, self.record_quality_code)
if not got:
# 仅首次空闲时提示一次
LOGSTATE.mark_idle_once(self.worker_name)
time.sleep(self.scan_interval_when_idle)
continue
LOGSTATE.clear_idle_flag(self.worker_name)
url, port = got
real_url = select_source_url(url, port)
if not real_url:
self._p(f"未得到播放地址，跳过：{url}")
self.coordinator.release(url, self.key)
time.sleep(1)
continue
self.current_platform = detect_platform(url)
self.current_anchor = port.get("anchor_name", "")
headers = headers_for_platform(self.current_platform, url)
out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution)
self.current_proc = launch_ffmpeg(real_url, out_url, headers, v_args, a_args, self.worker_name)
self.current_url = url
self.current_real_url = real_url
self.start_time = time.time()
self._last_progress_print = -9999
self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}")
except Exception as e:
logger.error(f"[{self.worker_name}] 主循环异常：{e}")
finally:
stop_ffmpeg(self.current_proc)
self.coordinator.release(self.current_url, self.key)
self._p("已退出。")
def _clear_state(self, release: bool = False):
if release and self.current_url:
self.coordinator.release(self.current_url, self.key)
self.current_url = None
self.current_real_url = None
self.current_proc = None
self.current_platform = ""
self.current_anchor = ""
# =============== 启动入口（保持原行为） ===============
def check_ffmpeg_existence() -> bool:
try:
result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True)
if result.returncode == 0:
print(result.stdout.splitlines()[0])
except Exception:
pass
finally:
return bool(check_ffmpeg())
# ---- 新增：统一优雅退出 ----
def graceful_shutdown(signum=None, frame=None):
if not SHUTDOWN_EVENT.is_set():
print("\n收到中断，正在停止所有 worker 与 ffmpeg …")
SHUTDOWN_EVENT.set()
# 先发 stop 给所有 worker（会内部 stop 当前 ffmpeg）
for w in list(ALL_WORKERS):
try:
w.stop()
except Exception:
pass
# 再兜底停止全局登记的 ffmpeg 子进程
with ALL_LOCK:
procs = list(ALL_PROCS)
for p in procs:
try:
stop_ffmpeg(p)
except Exception:
pass
def main():
print("-----------------------------------------------------")
print("| DouyinLiveRecorder - 单进程转推版 |")
print("| DouyinLiveRecorder - 并行转推版 |")
print("-----------------------------------------------------")
print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder")
print("模式：只检测 + 单进程推流（停播自动切下一个 URL）")
print("模式：并行检测 + 并行推流（保持原有业务逻辑）")
print(".....................................................")
if not check_ffmpeg_existence():
@@ -458,160 +704,61 @@ def main():
logger.error("URL_config.ini 中没有可用 URL。")
sys.exit(1)
# 顺序轮询索引
idx = 0
current_url: Optional[str] = None
current_real_url: Optional[str] = None # 实际播放地址（flv/m3u8）
current_proc: Optional[subprocess.Popen] = None
current_platform = ""
current_anchor = ""
record_quality_code = "OD" # 固定“原画”；需要也可改为从 config.ini 读取
# keys（并行路数 = keys 数量）；兼容单 key
keys = _yt_get_list("youtube", "keys")
if not keys:
single_key = _yt_get("youtube", "key", fallback="", cast=str)
if single_key:
keys = [single_key]
if not keys:
logger.error("youtube.ini 没有配置 [youtube].keys 或 key")
sys.exit(1)
check_interval_live = 8 # 推流中检测当前源是否仍在直播的间隔秒
scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒
last_md5 = "" # URL 文件变化时热更新
# 可选：每路分辨率
per_res = _yt_get_list("youtube", "resolutions") # 如：1920x1080,1080x1920
while per_res and len(per_res) < len(keys):
per_res.append(per_res[-1])
# watchdog 参数（更稳）
watchdog_grace = 120 # 推流开始 120s 内不做静默判定
watchdog_silence = 600 # 10 分钟完全无输出才重启
refresh_real_url_every = 300 # 每 5 分钟刷新一次真实播放地址，避免 token 过期
# 打印 ffmpeg 版本（贴合旧风格）
try:
vline = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True).stdout.splitlines()[0]
print(vline)
except Exception:
pass
coordinator = UrlCoordinator(url_config_file)
# 信号处理（Ctrl+C / kill）
try:
signal.signal(signal.SIGINT, graceful_shutdown)
except Exception:
pass
try:
signal.signal(signal.SIGTERM, graceful_shutdown)
except Exception:
pass
# 启动多路（Worker 前缀区分）
workers: List[StreamWorker] = []
for i, key in enumerate(keys, start=1):
res_for_this = per_res[i-1] if per_res else None
w = StreamWorker(worker_id=i, key=key, resolution=res_for_this, coordinator=coordinator)
w.start()
workers.append(w)
# ---- 新增：登记全局，供优雅退出用 ----
ALL_WORKERS.append(w)
try:
while True:
# 动态热更新 URL 列表（URL_config.ini 有变更则重载）
try:
md5 = check_md5(url_config_file)
if md5 != last_md5:
urls = parse_urls_from_file(url_config_file) or urls
last_md5 = md5
print(f"[{now_str()}] 已重载 URL 列表，数量：{len(urls)}")
# 若当前 URL 已不在列表，强制切换
if current_url and current_url not in urls:
print(f"[{now_str()}] 当前URL已被移除，停止并重新选择。")
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
current_real_url = None
except Exception as e:
logger.warning(f"URL 列表热更新失败：{e}")
if current_proc and current_url:
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 = None
current_url = None
current_real_url = None
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 # 容错：无法判断时先认为还活着
if still_live:
elapsed = int(time.time() - start_time)
h, m = divmod(elapsed, 3600)
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="")
# 定期刷新真实播放地址，避免 token 过期
try:
if elapsed >= refresh_real_url_every and (elapsed % refresh_real_url_every) < check_interval_live:
new_port = asyncio.run(fetch_port_info(current_url, record_quality_code))
new_real_url = select_source_url(current_url, new_port)
if new_real_url and new_real_url != current_real_url:
color.print_colored(f"\n[{now_str()}] 刷新播放地址（token 变化），重启 ffmpeg…", color.YELLOW)
stop_ffmpeg(current_proc)
out_url, v_args, a_args = build_youtube_output()
headers = headers_for_platform(current_platform, current_url)
current_proc = launch_ffmpeg(new_real_url, out_url, headers, v_args=v_args, a_args=a_args)
current_real_url = new_real_url
start_time = time.time()
time.sleep(check_interval_live)
continue
except Exception as e:
logger.warning(f"刷新播放地址失败：{e}")
# ✅ watchdog（推流开始 > 120s 且 10 分钟完全无输出）
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()}] 超过10分钟未见 ffmpeg 输出，判定异常，重启…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
current_real_url = None
time.sleep(1)
continue
time.sleep(check_interval_live)
continue
else:
color.print_colored(f"\n[{now_str()}] 主播停播，切换下一个源…", color.YELLOW)
stop_ffmpeg(current_proc)
current_proc = None
current_url = None
current_real_url = None
else:
# 未在推流：从 urls 顺序找第一个 is_live 的
if not urls:
time.sleep(scan_interval_when_idle)
continue
tried = 0
found = False
start_idx = idx
while tried < len(urls):
url = urls[idx]
idx = (idx + 1) % len(urls)
tried += 1
port = asyncio.run(fetch_port_info(url, record_quality_code))
if not port.get("anchor_name"):
print(f"[{now_str()}] 获取失败，跳过：{url}")
continue
if not port.get("is_live"):
print(f"[{now_str()}] 未开播：{port.get('anchor_name','')} | {url}")
continue
# 选取真实播放地址
real_url = select_source_url(url, port)
if not real_url:
print(f"[{now_str()}] 未得到播放地址，跳过：{url}")
continue
current_platform = detect_platform(url)
current_anchor = port.get("anchor_name", "")
headers = headers_for_platform(current_platform, url)
# 启动 ffmpeg
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
current_real_url = real_url
start_time = time.time()
color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN)
found = True
break
if not found:
# 没有任何源在播，过会儿再试
print(f"[{now_str()}] 当前无可用直播源，等待中…")
time.sleep(scan_interval_when_idle)
continue
while not SHUTDOWN_EVENT.is_set():
time.sleep(0.5)
except KeyboardInterrupt:
print("\n收到中断，正在退出…")
except Exception as e:
logger.error(f"主循环异常：{e}")
graceful_shutdown()
finally:
stop_ffmpeg(current_proc)
print("已退出。")
# 二次保险，确保全部退出
graceful_shutdown()
for w in workers:
w.join(timeout=10)
print("全部 worker 已退出。")
if __name__ == "__main__":
main()