952 lines
39 KiB
Python
952 lines
39 KiB
Python
# -*- 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
|
||
import io
|
||
import logging
|
||
from logging.handlers import RotatingFileHandler
|
||
|
||
# ======================= 颜色日志（Windows Cmder 友好） =======================
|
||
def _enable_vt_on_windows():
|
||
"""在 Windows 控制台启用 ANSI VT（Cmder/CMD/PowerShell 颜色支持）"""
|
||
if os.name != "nt":
|
||
return
|
||
try:
|
||
import ctypes
|
||
kernel32 = ctypes.windll.kernel32
|
||
handle = kernel32.GetStdHandle(-11) # STD_OUTPUT_HANDLE = -11
|
||
mode = ctypes.c_uint32()
|
||
if kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
|
||
mode.value |= 0x0004 # ENABLE_VIRTUAL_TERMINAL_PROCESSING
|
||
kernel32.SetConsoleMode(handle, mode)
|
||
except Exception:
|
||
pass
|
||
|
||
_enable_vt_on_windows()
|
||
|
||
class ColorLog:
|
||
# ANSI colors
|
||
RESET = "\033[0m"
|
||
GREY = "\033[90m"
|
||
RED = "\033[91m"
|
||
GREEN = "\033[92m"
|
||
YELLOW = "\033[93m"
|
||
BLUE = "\033[94m"
|
||
MAGENTA = "\033[95m"
|
||
CYAN = "\033[96m"
|
||
WHITE = "\033[97m"
|
||
|
||
ICONS = {
|
||
"INFO": "green_circle",
|
||
"WARN": "yellow_circle",
|
||
"ERROR": "red_circle",
|
||
"DEBUG": "white_circle",
|
||
"OK": "check_mark",
|
||
"SWAP": "repeat",
|
||
"LIVE": "satellite",
|
||
"OFF": "no_entry",
|
||
"FFMPEG":"movie_camera",
|
||
"WATCH": "eyes",
|
||
}
|
||
|
||
COLORS = {
|
||
"INFO": GREEN,
|
||
"OK": GREEN,
|
||
"WARN": YELLOW,
|
||
"ERROR": RED,
|
||
"DEBUG": GREY,
|
||
"SWAP": CYAN,
|
||
"LIVE": BLUE,
|
||
"OFF": RED,
|
||
"FFMPEG": MAGENTA,
|
||
"WATCH": CYAN,
|
||
}
|
||
|
||
def __init__(self, log_file_path: str = "logs/runtime.log", max_bytes: int = 5*1024*1024, backup: int = 3):
|
||
os.makedirs(os.path.dirname(log_file_path), exist_ok=True)
|
||
|
||
self._logger = logging.getLogger("Runtime")
|
||
self._logger.setLevel(logging.INFO)
|
||
self._logger.propagate = False
|
||
# 控制台 Handler（给 src.utils.logger 让位，这里只用于文件）
|
||
if not self._logger.handlers:
|
||
file_handler = RotatingFileHandler(log_file_path, maxBytes=max_bytes, backupCount=backup, encoding="utf-8")
|
||
fmt = logging.Formatter("%(asctime)s | %(levelname)-5s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
|
||
file_handler.setFormatter(fmt)
|
||
self._logger.addHandler(file_handler)
|
||
|
||
def _stamp(self) -> str:
|
||
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
def _compose(self, level: str, msg: str, worker: Optional[str] = None) -> str:
|
||
icon = self.ICONS.get(level, "")
|
||
color = self.COLORS.get(level, self.WHITE)
|
||
prefix = f"[{self._stamp()}]"
|
||
if worker:
|
||
prefix += f"[{worker}]"
|
||
return f"{prefix} {color}[{icon} {level}]{self.RESET} {msg}"
|
||
|
||
def info(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("INFO", msg, worker)
|
||
print(text)
|
||
self._logger.info(msg)
|
||
|
||
def ok(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("OK", msg, worker)
|
||
print(text)
|
||
self._logger.info(msg)
|
||
|
||
def warn(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("WARN", msg, worker)
|
||
print(text)
|
||
self._logger.warning(msg)
|
||
|
||
def error(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("ERROR", msg, worker)
|
||
print(text, file=sys.stderr)
|
||
self._logger.error(msg)
|
||
|
||
def debug(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("DEBUG", msg, worker)
|
||
print(text)
|
||
self._logger.debug(msg)
|
||
|
||
def swap(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("SWAP", msg, worker)
|
||
print(text)
|
||
self._logger.info("[SWAP] " + msg)
|
||
|
||
def live(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("LIVE", msg, worker)
|
||
print(text)
|
||
self._logger.info("[LIVE] " + msg)
|
||
|
||
def off(self, msg: str, worker: Optional[str] = None):
|
||
text = self._compose("OFF", msg, worker)
|
||
print(text)
|
||
self._logger.info("[OFF] " + msg)
|
||
|
||
CLOG = ColorLog() # 全局彩色日志
|
||
|
||
# ======================= 这里开始是原有工程的 import ===========================
|
||
from src import spider, stream
|
||
from src.utils import logger, Color, check_md5
|
||
from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path
|
||
|
||
# =============== 重要：先定义 build_youtube_output_for_key，避免 NameError ===============
|
||
# 读取 YouTube 配置（需要在函数内部访问 _yt_get，故先声明占位；稍后会正式赋值）
|
||
_yt_cfg: Optional[configparser.ConfigParser] = None
|
||
|
||
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}"
|
||
|
||
# ======================= QSV 优化：build_youtube_output_for_key =======================
|
||
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)
|
||
|
||
# === QSV 硬件编码（N5105 超低 CPU）===
|
||
# v_args = [
|
||
# "-c:v", "h264_qsv",
|
||
# "-preset", "veryfast",
|
||
# "-global_quality", "22",
|
||
# "-b:v", f"{bv}k",
|
||
# "-maxrate", f"{bv}k",
|
||
# "-bufsize", f"{max(bv*2, 10000)}k",
|
||
# "-g", str(gop//2),
|
||
# "-bf", "0",
|
||
# "-pix_fmt", "nv12",
|
||
# "-look_ahead", "0",
|
||
# "-look_ahead_depth", "0",
|
||
# "-vprofile", "high",
|
||
# "-level", "4.1",
|
||
# ]
|
||
v_args = [
|
||
"-c:v", "h264_qsv",
|
||
"-preset", "veryfast",
|
||
"-global_quality", "23", # 22→23 再降一点画质换性能
|
||
"-b:v", "3500k", # 你原 5000k → 降码率更省
|
||
"-maxrate", "3500k",
|
||
"-bufsize", "7000k",
|
||
"-g", "60",
|
||
"-bf", "0",
|
||
"-pix_fmt", "nv12",
|
||
"-look_ahead", "0",
|
||
]
|
||
|
||
if res_str:
|
||
try:
|
||
w, h = map(int, res_str.lower().split("x"))
|
||
w, h = _round_res(w, h)
|
||
v_args = ["-vf", f"scale={w}:{h}:flags=lanczos"] + v_args
|
||
except Exception:
|
||
logger.warning(f"resolution '{res_str}' 非法，忽略")
|
||
|
||
# a_args = [
|
||
# "-c:a", "aac",
|
||
# "-b:a", f"{ba}k",
|
||
# "-ar", "44100",
|
||
# "-ac", "2",
|
||
# ]
|
||
a_args = ["-c:a", "copy"] # 关键！不重编码音频
|
||
|
||
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 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)
|
||
CLOG.info("你的网络似乎直连可用（无需代理）。")
|
||
except Exception:
|
||
if not use_proxy:
|
||
print("\033[93m提示：无法直连海外站，如需 TikTok 请在 config.ini 开启并填写代理地址。\033[0m")
|
||
|
||
# =============== URL 列表 ===============
|
||
def parse_urls_from_file(path: str) -> List[str]:
|
||
urls, seen = [], set()
|
||
if not os.path.isfile(path):
|
||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||
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]:
|
||
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):
|
||
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:
|
||
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/自定义） ===============
|
||
async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict:
|
||
proxy = proxy_addr
|
||
delay = 1.2
|
||
for attempt in range(1, 4):
|
||
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)
|
||
|
||
if not port_info.get("is_live"):
|
||
handlers = []
|
||
if proxy:
|
||
handlers.append(urllib.request.ProxyHandler({"http": proxy, "https": proxy}))
|
||
opener = urllib.request.build_opener(*handlers) if handlers else urllib.request.build_opener()
|
||
headers = {
|
||
"User-Agent": random.choice([
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127 Safari/537.36",
|
||
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile Safari/604.1",
|
||
]),
|
||
"Referer": f"https://live.douyin.com/{re.sub(r'\\D','',url)}",
|
||
"Accept": "*/*",
|
||
}
|
||
req = urllib.request.Request(url, headers=headers)
|
||
with contextlib.closing(opener.open(req, timeout=6)) as resp:
|
||
html = resp.read(512*1024).decode("utf-8", errors="ignore")
|
||
|
||
m = re.search(r'"roomStore"\s*:\s*({.*?})\s*,\s*"userStore"', html)
|
||
if m:
|
||
try:
|
||
data = json.loads(m.group(1))
|
||
stream_data = data.get("roomInfo", {}).get("stream_url", {})
|
||
flv = (stream_data.get("flv_pull_url", {}).get("FULL_HD1")
|
||
or stream_data.get("flv_pull_url", {}).get("HD1"))
|
||
m3u8 = (stream_data.get("hls_pull_url", {}).get("FULL_HD1")
|
||
or stream_data.get("hls_pull_url", {}).get("HD1"))
|
||
if flv or m3u8:
|
||
port_info = {"anchor_name": "JSONFallback", "is_live": True}
|
||
if flv: port_info["flv_url"] = flv
|
||
else: port_info["m3u8_url"] = m3u8
|
||
CLOG.warn("JSON fallback 成功解析流地址", worker=None)
|
||
except Exception as e:
|
||
CLOG.debug(f"JSON fallback 解析失败: {e}")
|
||
|
||
elif "tiktok.com" in url:
|
||
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:
|
||
port_info = {"anchor_name": "", "is_live": False}
|
||
logger.error("TikTok 需要代理。")
|
||
|
||
elif (".m3u8" in url) or (".flv" in url):
|
||
port_info = {"anchor_name": "自定义", "is_live": True}
|
||
port_info["flv_url" if url.endswith(".flv") else "m3u8_url"] = url
|
||
|
||
else:
|
||
port_info = {"anchor_name": "", "is_live": False}
|
||
|
||
return port_info
|
||
|
||
except urllib.error.HTTPError as e:
|
||
if e.code == 403:
|
||
CLOG.warn(f"[fetch_port_info] Douyin 403 第{attempt}次延时重试")
|
||
time.sleep(delay)
|
||
delay *= 2
|
||
continue
|
||
CLOG.error(f"[fetch_port_info] HTTP {e.code}: {url}")
|
||
break
|
||
except Exception as e:
|
||
CLOG.warn(f"[fetch_port_info] 异常: {e}")
|
||
time.sleep(delay)
|
||
delay *= 2
|
||
|
||
return {"anchor_name": "", "is_live": False}
|
||
|
||
# ======================= 精简版 launch_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:
|
||
base = [
|
||
"ffmpeg", "-nostdin", "-hide_banner", "-stats", "-stats_period", "60",
|
||
"-rw_timeout", "15000000", "-reconnect", "1", "-reconnect_at_eof", "1", "-reconnect_delay_max", "30",
|
||
"-user_agent", "Mozilla/5.0", "-protocol_whitelist", "file,http,https,tcp,tls,crypto,udp,rtmp,rtp,httpproxy",
|
||
"-thread_queue_size", "2048", "-analyzeduration", "5000000", "-probesize", "5000000",
|
||
]
|
||
if extra_headers: base += ["-headers", extra_headers]
|
||
if proxy_addr: base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:]
|
||
base += ["-re", "-i", input_url]
|
||
base += ["-fflags", "+genpts+igndts", "-avoid_negative_ts", "make_zero"]
|
||
base += v_args + a_args
|
||
base += ["-f", "flv", "-flvflags", "no_duration_filesize", out_url]
|
||
|
||
env = os.environ.copy()
|
||
if proxy_addr:
|
||
env["http_proxy"] = proxy_addr
|
||
env["https_proxy"] = proxy_addr
|
||
|
||
creationflags = subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0
|
||
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()
|
||
|
||
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()
|
||
if any(k in text for k in ["bitrate=", "frame=", "speed="]):
|
||
CLOG.info(f"[统计] {text}", worker=worker_name)
|
||
except: 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] = {}
|
||
self.no_source_worker: Dict[str, bool] = {}
|
||
|
||
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):
|
||
CLOG.debug("当前无可用直播源，等待中…", worker=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:
|
||
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
|
||
CLOG.info(f"已重载 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:
|
||
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))
|
||
anchor = port.get('anchor_name', '')
|
||
if not port.get("anchor_name"):
|
||
LOGSTATE.set_status(url, "unknown", text=f"解析失败：{url}")
|
||
tried += 1
|
||
continue
|
||
if not port.get("is_live"):
|
||
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)
|
||
|
||
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
|
||
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()
|
||
self.start_time = time.time()
|
||
|
||
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):
|
||
self._stop_event.set()
|
||
with contextlib.suppress(Exception):
|
||
stop_ffmpeg(self.current_proc)
|
||
|
||
def _soft_restart_ffmpeg(self, reason: str):
|
||
now = time.time()
|
||
if now - self._last_soft_restart_ts < self._soft_restart_min_interval:
|
||
CLOG.warn(f"跳过软重启（冷却中）: {reason}", worker=self.worker_name)
|
||
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)
|
||
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:
|
||
CLOG.error("软重启失败：无法获取播放 URL。", worker=self.worker_name)
|
||
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
|
||
CLOG.ok(f"已软重启 ffmpeg：{reason}", worker=self.worker_name)
|
||
return True
|
||
|
||
def run(self):
|
||
CLOG.ok("启动", worker=self.worker_name)
|
||
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:
|
||
CLOG.error(f"ffmpeg 退出(code={retcode})，切换下一个源…", worker=self.worker_name)
|
||
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:
|
||
CLOG.warn(f"检测当前源异常：{e}", worker=self.worker_name)
|
||
is_live_raw = True
|
||
|
||
if not is_live_raw:
|
||
alive_by_probe = probe_stream_url(self.current_real_url)
|
||
if alive_by_probe:
|
||
CLOG.warn("API 返回下播但探活成功，视为仍在播。", worker=self.worker_name)
|
||
is_live_raw = True
|
||
|
||
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:
|
||
CLOG.warn(f"短期异常 is_live=False（{self._offline_count}），已忽略", worker=self.worker_name)
|
||
still_live = True
|
||
elif self._offline_count < 3:
|
||
CLOG.warn(f"疑似下播 {self._offline_count}/3 次，继续观望", worker=self.worker_name)
|
||
still_live = True
|
||
else:
|
||
still_live = False
|
||
|
||
if still_live:
|
||
elapsed = int(time.time() - self.start_time)
|
||
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}"
|
||
CLOG.live(f"正在转播({dur_str})：{self.current_platform} | {self.current_anchor}", worker=self.worker_name)
|
||
self._last_progress_print = elapsed
|
||
|
||
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:
|
||
CLOG.swap("刷新播放地址（token 变化），软重启 ffmpeg…", worker=self.worker_name)
|
||
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:
|
||
CLOG.warn(f"刷新播放地址失败：{e}", worker=self.worker_name)
|
||
|
||
silence = time.time() - getattr(self.current_proc, "last_out_ts", 0)
|
||
if elapsed >= self.watchdog_grace and silence > self.watchdog_silence:
|
||
CLOG.warn("超过10分钟未见输出，尝试软重启…", worker=self.worker_name)
|
||
ok = self._soft_restart_ffmpeg("watchdog silence")
|
||
if not ok:
|
||
CLOG.warn("软重启失败，释放当前源并切换…", worker=self.worker_name)
|
||
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:
|
||
CLOG.off("多次确认主播下播，切换下一个源…", worker=self.worker_name)
|
||
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)
|
||
|
||
|
||
LOGSTATE.clear_idle_flag(self.worker_name)
|
||
url, port = got
|
||
real_url = select_source_url(url, port)
|
||
if not real_url:
|
||
CLOG.warn(f"未得到播放地址，跳过：{url}", worker=self.worker_name)
|
||
self.coordinator.release(url, self.key)
|
||
time.sleep(1)
|
||
continue
|
||
|
||
if not probe_stream_url(real_url):
|
||
CLOG.warn(f"播放 URL 探活失败（可能刚开播或接口延迟），暂缓此源：{url}", worker=self.worker_name)
|
||
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._offline_count = 0
|
||
self._last_live_ts = time.time()
|
||
|
||
CLOG.ok(f"开始转播：{self.current_platform} | {self.current_anchor} | {url}", worker=self.worker_name)
|
||
|
||
except Exception as e:
|
||
CLOG.error(f"内部异常：{e}", worker=self.worker_name)
|
||
time.sleep(2)
|
||
continue
|
||
|
||
finally:
|
||
with contextlib.suppress(Exception):
|
||
stop_ffmpeg(self.current_proc)
|
||
self.coordinator.release(self.current_url, self.key)
|
||
CLOG.off("已退出。", worker=self.worker_name)
|
||
|
||
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()
|
||
for w in list(ALL_WORKERS):
|
||
with contextlib.suppress(Exception):
|
||
w.stop()
|
||
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 = _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")
|
||
while per_res and len(per_res) < len(keys):
|
||
per_res.append(per_res[-1])
|
||
|
||
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)
|
||
|
||
with contextlib.suppress(Exception):
|
||
signal.signal(signal.SIGINT, graceful_shutdown)
|
||
with contextlib.suppress(Exception):
|
||
signal.signal(signal.SIGTERM, graceful_shutdown)
|
||
|
||
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() |