907 lines
37 KiB
Plaintext
907 lines
37 KiB
Plaintext
# -*- encoding: utf-8 -*-
|
||
|
||
"""
|
||
并行转推（稳定性增强版，保持原业务逻辑与对外接口）
|
||
- 核心修复：降低“误判下播”的概率（抖音仍在播但 API 偶发失败）
|
||
* fetch_port_info：加入重试（3 次，指数退避），不改签名/调用处
|
||
* “下播防抖”：只有在“连续N次 is_live=False 且最近M秒没有任何活跃迹象”才判定下播
|
||
* 交叉验证：若 is_live=False，但当前 real_url 可探测（探活成功），则判为仍在播
|
||
- FFmpeg 看门狗：卡顿/长时间无输出 -> 先软重启（保留 URL 占用），再必要时释放
|
||
- 细节增强：日志更明确，解释每次判定原因；优雅退出确保无僵尸进程
|
||
- 兼容性：不改配置文件/外部模块（spider、stream 等）的接口与行为
|
||
- 修复：确保函数定义顺序，避免 NameError
|
||
"""
|
||
|
||
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 contextlib
|
||
|
||
import configparser
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
from src import spider, stream
|
||
from src.utils import logger, Color, check_md5
|
||
from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
|
||
|
||
# =============== 构建输出参数（每个 key） ———— 确保在此定义，早于 StreamWorker ===============
|
||
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
|
||
|
||
|
||
# =============== 基础环境 ===============
|
||
version = "vrelay-parallel-2.1.1-pro"
|
||
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()
|
||
|
||
# ---- 全局 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(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, defv):
|
||
try:
|
||
v = int(v)
|
||
if v < lo or v > hi: return defv
|
||
return v
|
||
except Exception:
|
||
return defv
|
||
|
||
# =============== 工具函数 ===============
|
||
def _round_res(w: int, h: int) -> Tuple[int, int]:
|
||
# 偶数对齐，避免 1088x1920 类问题
|
||
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 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)
|
||
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
|
||
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
|
||
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]:
|
||
"""
|
||
选源优先级保持不变：抖音/TikTok 优先 FLV（且避免 H265），否则 record_url/m3u8/flv。
|
||
"""
|
||
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")
|
||
|
||
# =============== 辅助：探测播放 URL 是否可用（用于误判纠正） ===============
|
||
def _open_with_proxy(url: str, timeout: float = 5.0):
|
||
"""
|
||
使用 urllib 在当前代理环境下打开 URL。
|
||
仅抓取少量字节用于“活跃性探测”，避免大流量。
|
||
"""
|
||
handlers = []
|
||
if proxy_addr:
|
||
handlers.append(urllib.request.ProxyHandler({"http": proxy_addr, "https": proxy_addr}))
|
||
opener = urllib.request.build_opener(*handlers) if handlers else urllib.request.build_opener()
|
||
req = urllib.request.Request(
|
||
url,
|
||
headers={
|
||
"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",
|
||
"Accept": "*/*",
|
||
"Connection": "close",
|
||
}
|
||
)
|
||
return opener.open(req, timeout=timeout)
|
||
|
||
def probe_stream_url(real_url: Optional[str], timeout: float = 5.0) -> bool:
|
||
"""
|
||
轻量级“探活”：
|
||
- m3u8: 取前若干字节并检查 "#EXTM3U"
|
||
- flv: 取前3字节是否 "FLV"
|
||
- 其它：只要 200 且有数据即可视为“活着”
|
||
失败不抛异常，返回 False
|
||
"""
|
||
if not real_url:
|
||
return False
|
||
try:
|
||
with contextlib.closing(_open_with_proxy(real_url, timeout=timeout)) as resp:
|
||
code = getattr(resp, "status", None) or getattr(resp, "code", None)
|
||
if code and int(code) >= 400:
|
||
return False
|
||
head = resp.read(256) # 少量字节足够判定格式
|
||
u = real_url.lower()
|
||
if ".m3u8" in u:
|
||
return b"#EXTM3U" in head.upper()
|
||
if ".flv" in u:
|
||
return head.startswith(b"FLV")
|
||
# 其它类型：有数据即认为仍在
|
||
return bool(head)
|
||
except Exception:
|
||
return False
|
||
|
||
# =============== 解析与选源（抖音/TikTok/自定义） ===============
|
||
# 修复点①：增强 fetch_port_info —— 重试 + 指数退避（不改签名/调用处）
|
||
async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
|
||
proxy = proxy_addr
|
||
delay = 1.5 # 初始退避
|
||
for attempt in range(1, 4): # 最多 3 次
|
||
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}
|
||
|
||
# 返回前补充：若 is_live True 但未给出真实播放地址，也作为暂不可用
|
||
if port_info.get("is_live") and not select_source_url(url, port_info):
|
||
logger.warning(f"[fetch_port_info] 在播但暂未拿到播放 URL，可能接口延迟：{url}")
|
||
return port_info
|
||
|
||
except Exception as e:
|
||
logger.warning(f"[fetch_port_info] 第{attempt}次解析失败: {e}")
|
||
if attempt < 3:
|
||
try:
|
||
await asyncio.sleep(delay)
|
||
except Exception:
|
||
# 若在同步上下文内调用（意外），兜底
|
||
time.sleep(delay)
|
||
delay *= 2
|
||
else:
|
||
logger.error(f"[fetch_port_info] 多次失败，返回空状态: {url}")
|
||
return {"anchor_name": "", "is_live": False}
|
||
|
||
|
||
# =============== 启动/停止 ffmpeg ===============
|
||
def launch_ffmpeg(input_url: str,
|
||
out_url: str,
|
||
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",
|
||
"-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+flush_packets",
|
||
"-use_wallclock_as_timestamps", "1",
|
||
"-avioflags", "direct",
|
||
"-rtbufsize", "100M",
|
||
"-err_detect", "ignore_err",
|
||
"-avoid_negative_ts", "make_zero",
|
||
]
|
||
|
||
# 视频/音频编码参数（使用 build_youtube_output 的结果）
|
||
v_args = v_args or []
|
||
a_args = a_args or []
|
||
base += v_args + a_args
|
||
|
||
# 输出（flv 到 YT，更宽容）
|
||
base += ["-f", "flv", "-flvflags", "no_duration_filesize", out_url]
|
||
|
||
cmd_str = " ".join(base)
|
||
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":
|
||
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()
|
||
|
||
# ---- 注册进程，便于 Ctrl+C 统一清理 ----
|
||
with ALL_LOCK:
|
||
ALL_PROCS.add(p)
|
||
|
||
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()
|
||
|
||
# 只保留关键统计/错误（前缀带 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"{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
|
||
|
||
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:
|
||
with contextlib.suppress(Exception):
|
||
p.stdin.write(b"q"); p.stdin.flush()
|
||
p.wait(timeout=6)
|
||
else:
|
||
p.send_signal(signal.SIGINT)
|
||
p.wait(timeout=10)
|
||
except Exception:
|
||
with contextlib.suppress(Exception):
|
||
p.kill()
|
||
finally:
|
||
# ---- 从全局集合移除 ----
|
||
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 = {
|
||
'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 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 级独占，保持原顺序） ===============
|
||
class UrlCoordinator:
|
||
"""
|
||
- urls 顺序分配
|
||
- in_use_urls: url 级别独占（防止同一 url 被不同 worker 同时占用）
|
||
"""
|
||
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()
|
||
|
||
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]]:
|
||
"""尝试遍历一轮 URL 表，找到首个在播的；否则返回 None。"""
|
||
with self.lock:
|
||
total = len(self.urls)
|
||
if total == 0:
|
||
return None
|
||
|
||
tried = 0
|
||
while tried < total:
|
||
with self.lock:
|
||
url = self.urls[self.idx]
|
||
self.idx = (self.idx + 1) % total
|
||
|
||
if url in self.in_use_urls:
|
||
tried += 1
|
||
continue
|
||
|
||
# 锁外解析（避免阻塞）
|
||
port = asyncio.run(fetch_port_info(url, quality_code))
|
||
if not port.get("anchor_name"):
|
||
LOGSTATE.set_status(url, "unknown", text=f"解析失败：{url}")
|
||
tried += 1
|
||
continue
|
||
if not port.get("is_live"):
|
||
anchor = port.get('anchor_name', '')
|
||
LOGSTATE.set_status(url, "offline", text=f"未开播：{anchor} | {url}")
|
||
tried += 1
|
||
continue
|
||
|
||
# 回到锁内占用
|
||
with self.lock:
|
||
if url in self.in_use_urls:
|
||
tried += 1
|
||
continue
|
||
self.in_use_urls.add(url)
|
||
|
||
anchor = port.get('anchor_name', '')
|
||
LOGSTATE.set_status(url, "online", text=f"已开播：{anchor} | {url}")
|
||
return url, port
|
||
|
||
return None
|
||
|
||
def release(self, url: Optional[str], worker_key: Optional[str]):
|
||
if not url:
|
||
return
|
||
with self.lock:
|
||
self.in_use_urls.discard(url)
|
||
|
||
# =============== 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 # 10 分钟无输出视为卡死
|
||
self.refresh_real_url_every = 300 # 5 分钟刷新一次真实地址（token 可能过期）
|
||
self.check_interval_live = 8 # 在播时的检测间隔
|
||
self.scan_interval_when_idle = 2 # 空闲时的扫描间隔
|
||
|
||
self._last_progress_print = -9999
|
||
self._stop_event = threading.Event()
|
||
self.start_time = time.time()
|
||
|
||
# 修复点②：新增下播防抖状态（连续多次 is_live=False 才判定）
|
||
self._offline_count = 0
|
||
self._last_live_ts = time.time()
|
||
|
||
# 软重启冷却（避免频繁重启）
|
||
self._last_soft_restart_ts = 0.0
|
||
self._soft_restart_min_interval = 90.0
|
||
|
||
def stop(self):
|
||
"""收到停止请求时，尽快停掉当前 ffmpeg。"""
|
||
self._stop_event.set()
|
||
with contextlib.suppress(Exception):
|
||
stop_ffmpeg(self.current_proc)
|
||
|
||
def _p(self, text: str):
|
||
print(f"[{now_str()}][{self.worker_name}] {text}")
|
||
|
||
def _soft_restart_ffmpeg(self, reason: str):
|
||
"""
|
||
尝试“软重启”ffmpeg：仅重启进程，不释放 URL 占用，优先保持连续性。
|
||
"""
|
||
now = time.time()
|
||
if now - self._last_soft_restart_ts < self._soft_restart_min_interval:
|
||
self._p(f"跳过软重启（冷却中）: {reason}")
|
||
return False
|
||
self._last_soft_restart_ts = now
|
||
|
||
try:
|
||
stop_ffmpeg(self.current_proc)
|
||
except Exception:
|
||
pass
|
||
|
||
out_url, v_args, a_args = build_youtube_output_for_key(self.key, self.resolution) # 修复：用 self.resolution，非 None
|
||
headers = headers_for_platform(self.current_platform, self.current_url or "")
|
||
if not self.current_real_url:
|
||
# 尝试重新解析一遍
|
||
new_port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code)) if self.current_url else {}
|
||
self.current_real_url = select_source_url(self.current_url or "", new_port)
|
||
if not self.current_real_url:
|
||
self._p("软重启失败：无法获取播放 URL。")
|
||
return False
|
||
|
||
self.current_proc = launch_ffmpeg(self.current_real_url, out_url, headers, v_args, a_args, self.worker_name)
|
||
self.start_time = time.time()
|
||
self._last_progress_print = -9999
|
||
self._p(f"已软重启 ffmpeg：{reason}")
|
||
return True
|
||
|
||
def run(self):
|
||
self._p("启动")
|
||
try:
|
||
while not self._stop_event.is_set():
|
||
try:
|
||
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
|
||
is_live_raw = True
|
||
try:
|
||
port = asyncio.run(fetch_port_info(self.current_url, self.record_quality_code))
|
||
is_live_raw = bool(port.get("is_live"))
|
||
except Exception as e:
|
||
logger.warning(f"[{self.worker_name}] 检测当前源异常：{e}")
|
||
is_live_raw = True # 网络/接口异常默认不下播
|
||
|
||
if not is_live_raw:
|
||
# 交叉探活：若当前 real_url 还能探测成功，认为仍在播
|
||
alive_by_probe = probe_stream_url(self.current_real_url)
|
||
if alive_by_probe:
|
||
logger.info(f"[{self.worker_name}] API 返回下播但探活成功，视为仍在播。")
|
||
is_live_raw = True
|
||
|
||
# 下播防抖：连续 N 次失败 且 距最后一次活跃已超过 M 秒
|
||
if is_live_raw:
|
||
self._offline_count = 0
|
||
self._last_live_ts = time.time()
|
||
still_live = True
|
||
else:
|
||
self._offline_count += 1
|
||
if time.time() - self._last_live_ts < 120:
|
||
logger.info(f"[{self.worker_name}] 短期异常 is_live=False（{self._offline_count}），已忽略")
|
||
still_live = True
|
||
elif self._offline_count < 3:
|
||
logger.info(f"[{self.worker_name}] 疑似下播 {self._offline_count}/3 次，继续观望")
|
||
still_live = True
|
||
else:
|
||
still_live = False # 多次确认才下播
|
||
|
||
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) # 修复：用 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}")
|
||
|
||
# 看门狗：长时间无输出 -> 先软重启，如果软重启失败再释放
|
||
silence = time.time() - getattr(self.current_proc, "last_out_ts", 0)
|
||
if elapsed >= self.watchdog_grace and silence > self.watchdog_silence:
|
||
self._p("⚠️ 超过10分钟未见输出，尝试软重启…")
|
||
ok = self._soft_restart_ffmpeg("watchdog silence")
|
||
if not ok:
|
||
self._p("软重启失败，释放当前源并切换…")
|
||
stop_ffmpeg(self.current_proc)
|
||
self._clear_state(release=True)
|
||
LOGSTATE.clear_idle_flag(self.worker_name)
|
||
time.sleep(1)
|
||
else:
|
||
time.sleep(self.check_interval_live)
|
||
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
|
||
|
||
# 新源上线前：探活（避免刚拿到 URL 就空推）
|
||
if not probe_stream_url(real_url):
|
||
self._p(f"播放 URL 探活失败（可能刚开播或接口延迟），暂缓此源：{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.resolution，非 None
|
||
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._offline_count = 0
|
||
self._last_live_ts = time.time()
|
||
|
||
self._p(f"✅ 开始转播：{self.current_platform} | {self.current_anchor} | {url}")
|
||
|
||
except Exception as e:
|
||
# 关键：任何内部异常不让线程退出
|
||
logger.error(f"[{self.worker_name}] 内部异常：{e}")
|
||
time.sleep(2)
|
||
continue
|
||
|
||
finally:
|
||
with contextlib.suppress(Exception):
|
||
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]) # 修复：统一用 stdout
|
||
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):
|
||
with contextlib.suppress(Exception):
|
||
w.stop()
|
||
# 再兜底停止全局登记的 ffmpeg 子进程
|
||
with ALL_LOCK:
|
||
procs = list(ALL_PROCS)
|
||
for p in procs:
|
||
with contextlib.suppress(Exception):
|
||
stop_ffmpeg(p)
|
||
|
||
def main():
|
||
print("DouyinRelay", version)
|
||
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（并行路数 = 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)
|
||
|
||
# 可选：每路分辨率
|
||
per_res = _yt_get_list("youtube", "resolutions") # 如：1920x1080,1080x1920
|
||
while per_res and len(per_res) < len(keys):
|
||
per_res.append(per_res[-1])
|
||
|
||
# 打印 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）
|
||
with contextlib.suppress(Exception):
|
||
signal.signal(signal.SIGINT, graceful_shutdown)
|
||
with contextlib.suppress(Exception):
|
||
signal.signal(signal.SIGTERM, graceful_shutdown)
|
||
|
||
# 启动多路（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 not SHUTDOWN_EVENT.is_set():
|
||
time.sleep(0.5)
|
||
except KeyboardInterrupt:
|
||
graceful_shutdown()
|
||
finally:
|
||
graceful_shutdown()
|
||
for w in workers:
|
||
w.join(timeout=10)
|
||
print("全部 worker 已退出。")
|
||
|
||
if __name__ == "__main__":
|
||
main() |