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

's'

Browse Source
This commit is contained in:
eric
2025-10-15 16:15:46 +08:00
parent 3c92b691db
commit 019f68a91a
3 changed files with 1410 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

907
back/mult5 Normal file
View File

@@ -0,0 +1,907 @@
# -*- 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()

7
config/URL_config.ini
View File

@@ -1,4 +1,8 @@
https://live.douyin.com/16531926455
https://live.douyin.com/782322393954
https://live.douyin.com/16531926455
https://live.douyin.com/739670432305
https://live.douyin.com/486789870173
https://live.douyin.com/15652099787
https://live.douyin.com/572442812068
https://live.douyin.com/477961427964
https://live.douyin.com/686507122231
@@ -14,6 +18,7 @@ https://live.douyin.com/122383435921
https://live.douyin.com/92876979328
https://live.douyin.com/517715534931
https://live.douyin.com/460525712926
https://live.douyin.com/339454333072
https://live.douyin.com/456576000931
https://live.douyin.com/745606325880
https://live.douyin.com/749687541944

734
main.py
View File

@@ -1,12 +1,15 @@
# -*- encoding: utf-8 -*-
"""
并行转推（日志优化版，保持原业务逻辑）
- 保留 Worker-1/2 等前缀
- 日志状态变化才打印，避免刷屏
- 禁止同一个 URL 推到同一个 YouTube key（(url,key) 级别防重复）
- 仍然避免同一个 URL 被多路并行占用（url 级别独占）
- 新增：Ctrl+C/TERM 时优雅停止所有 Worker 与 ffmpeg 子进程（仅最小增改）
并行转推（稳定性+体验增强版，保持原业务逻辑与对外接口）
- 核心修复：降低“误判下播”的概率（抖音仍在播但 API 偶发失败）
* fetch_port_info：加入重试（3 次，指数退避），不改签名/调用处
* “下播防抖”：只有在“连续N次 is_live=False 且最近M秒没有任何活跃迹象”才判定下播
* 交叉验证：若 is_live=False，但当前 real_url 可探测（探活成功），则判为仍在播
- FFmpeg 看门狗：卡顿/长时间无输出 -> 先软重启（保留 URL 占用），再必要时释放
- 细节增强：日志更明确，解释每次判定原因；优雅退出确保无僵尸进程
- 兼容性：不改配置文件/外部模块（spider、stream 等）的接口与行为
- 修复：确保函数定义顺序，避免 NameError
"""
import asyncio
@@ -20,49 +23,145 @@ 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 msvcrt # noqa: F401
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": "🟢",
"WARN": "🟡",
"ERROR": "🔴",
"DEBUG": "⚪",
"OK": "✅",
"SWAP": "🔁",
"LIVE": "📡",
"OFF": "📴",
"FFMPEG":"🎬",
"WATCH": "👀",
}
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
# =============== 基础环境 ===============
version = "relay-parallel-2.0.3-optimized"
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]
# =============== 重要：先定义 build_youtube_output_for_key，避免 NameError ===============
# 读取 YouTube 配置（需要在函数内部访问 _yt_get，故先声明占位；稍后会正式赋值）
_yt_cfg: Optional[configparser.ConfigParser] = None
def _safe_int(v, lo, hi, defv):
try:
@@ -72,7 +171,6 @@ def _safe_int(v, lo, hi, defv):
except Exception:
return defv
# =============== 工具函数 ===============
def _round_res(w: int, h: int) -> Tuple[int, int]:
# 偶数对齐，避免 1088x1920 类问题
return (w & ~1, h & ~1)
@@ -84,121 +182,11 @@ 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)
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]:
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
# =============== 构建输出参数（每个 key） ===============
def build_youtube_output_for_key(key: str, resolution: Optional[str]) -> Tuple[str, list, list]:
"""
！！！此函数放在文件靠前位置，确保任何地方调用都不再出现 NameError。
与旧版保持一致；仅对异常进行更稳妥处理与注释增强。
"""
if not key:
logger.error("错误: youtube.ini 未配置 key/keys")
sys.exit(1)
@@ -250,6 +238,222 @@ def build_youtube_output_for_key(key: str, resolution: Optional[str]) -> Tuple[s
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]:
"""
选源优先级保持不变：抖音/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,
@@ -302,12 +506,12 @@ def launch_ffmpeg(input_url: str,
# 通用容错 + 时间戳修复（优化版）
base += [
"-fflags", "+genpts+igndts+flush_packets", # 自动生成时间戳 + 忽略乱序 + 刷新缓冲
"-use_wallclock_as_timestamps", "1", # 用系统时钟修正时间戳
"-avioflags", "direct", # 减少延迟
"-rtbufsize", "100M", # 输入缓冲
"-err_detect", "ignore_err", # 忽略解码错误
"-avoid_negative_ts", "make_zero", # 避免负时间戳
"-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 的结果）
@@ -318,10 +522,9 @@ def launch_ffmpeg(input_url: str,
# 输出（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] + " ...")
# 控制台+文件日志
CLOG.live(f"FFmpeg CMD({worker_name}): {cmd_str if len(cmd_str) < 500 else cmd_str[:500] + ' ...'}", worker=worker_name)
logger.info(f"{worker_name} 启动推流：{input_url} -> {out_url}")
creationflags = 0
@@ -339,7 +542,7 @@ def launch_ffmpeg(input_url: str,
)
p.last_out_ts = time.time()
# ---- 新增：注册进程，便于 Ctrl+C 统一清理 ----
# ---- 注册进程，便于 Ctrl+C 统一清理 ----
with ALL_LOCK:
ALL_PROCS.add(p)
@@ -353,11 +556,11 @@ def launch_ffmpeg(input_url: str,
# 只保留关键统计/错误（前缀带 Worker）
if ("bitrate=" in text) or ("frame=" in text) or ("speed=" in text) or ("libx264" in text):
logger.info(f"{worker_name} [ffmpeg统计] {text}")
CLOG.info(f"[ffmpeg统计] {text}", worker=worker_name)
if "Server disconnected" in text:
logger.error(f"{worker_name} [输出异常] {text}")
CLOG.error(f"[输出异常] {text}", worker=worker_name)
elif ("not supported" in text) or ("Invalid" in text) or ("invalid" in text):
logger.error(f"{worker_name} [转码异常] {text}")
CLOG.error(f"[转码异常] {text}", worker=worker_name)
except Exception:
pass
@@ -369,19 +572,17 @@ def stop_ffmpeg(p: Optional[subprocess.Popen]):
try:
if os.name == "nt":
if p.stdin:
try:
with contextlib.suppress(Exception):
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
with contextlib.suppress(Exception):
p.kill()
finally:
# ---- 新增：从全局集合移除 ----
# ---- 从全局集合移除 ----
with ALL_LOCK:
ALL_PROCS.discard(p)
@@ -425,7 +626,7 @@ class LogState:
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}] 当前无可用直播源，等待中…")
CLOG.debug("当前无可用直播源，等待中…", worker=worker_name)
self.no_source_worker[worker_name] = True
def clear_idle_flag(self, worker_name: str):
@@ -434,7 +635,7 @@ class LogState:
LOGSTATE = LogState()
# =============== URL 协调器（只保留 url 独占） ===============
# =============== URL 协调器（url 级独占，保持原顺序） ===============
class UrlCoordinator:
"""
- urls 顺序分配
@@ -446,8 +647,7 @@ class UrlCoordinator:
self.idx: int = 0
self.lock = threading.Lock()
self.last_md5 = check_md5(path)
self.in_use_urls: Set[str] = set() # 只保留 url 独占
self.in_use_urls: Set[str] = set()
def refresh_if_changed(self):
try:
@@ -455,7 +655,7 @@ class UrlCoordinator:
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)}")
CLOG.info(f"已重载 URL 列表，数量：{len(self.urls)}")
except Exception as e:
logger.warning(f"URL 列表热更新失败：{e}")
@@ -472,19 +672,18 @@ class UrlCoordinator:
url = self.urls[self.idx]
self.idx = (self.idx + 1) % total
# 只要 url 已占用就跳过
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"):
anchor = port.get('anchor_name', '')
LOGSTATE.set_status(url, "offline", text=f"未开播：{anchor} | {url}")
tried += 1
continue
@@ -496,7 +695,6 @@ class UrlCoordinator:
continue
self.in_use_urls.add(url)
anchor = port.get('anchor_name', '')
LOGSTATE.set_status(url, "online", text=f"已开播：{anchor} | {url}")
return url, port
@@ -508,7 +706,7 @@ class UrlCoordinator:
with self.lock:
self.in_use_urls.discard(url)
# =============== Worker（保留前缀日志） ===============
# =============== Worker（保持原业务逻辑框架，加入防抖与探活） ===============
class StreamWorker(threading.Thread):
def __init__(self, worker_id: int, key: str, resolution: Optional[str], coordinator: UrlCoordinator):
super().__init__(daemon=True)
@@ -525,53 +723,112 @@ class StreamWorker(threading.Thread):
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.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，尽快响应退出。"""
"""收到停止请求时，尽快停掉当前 ffmpeg。"""
self._stop_event.set()
with contextlib.suppress(Exception):
stop_ffmpeg(self.current_proc)
def _soft_restart_ffmpeg(self, reason: str):
"""
尝试“软重启”ffmpeg：仅重启进程，不释放 URL 占用，优先保持连续性。
"""
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
def _p(self, text: str):
print(f"[{now_str()}][{self.worker_name}] {text}")
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):
self._p("启动")
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:
self._p(f"❌ ffmpeg 退出(code={retcode})，切换下一个源…")
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))
still_live = bool(port.get("is_live"))
is_live_raw = bool(port.get("is_live"))
except Exception as e:
logger.error(f"[{self.worker_name}] 检测当前源失败：{e}")
CLOG.warn(f"检测当前源异常：{e}", worker=self.worker_name)
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:
CLOG.warn("API 返回下播但探活成功，视为仍在播。", worker=self.worker_name)
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:
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)
@@ -580,7 +837,7 @@ class StreamWorker(threading.Thread):
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}")
CLOG.live(f"正在转播({dur_str})：{self.current_platform} | {self.current_anchor}", worker=self.worker_name)
self._last_progress_print = elapsed
# 定期刷新真实播放地址（token）
@@ -589,7 +846,7 @@ class StreamWorker(threading.Thread):
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…")
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)
@@ -600,28 +857,33 @@ class StreamWorker(threading.Thread):
time.sleep(self.check_interval_live)
continue
except Exception as e:
logger.warning(f"[{self.worker_name}] 刷新播放地址失败：{e}")
CLOG.warn(f"刷新播放地址失败：{e}", worker=self.worker_name)
# 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)
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:
self._p("📴 主播停播，切换下一个源…")
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)
@@ -632,7 +894,14 @@ class StreamWorker(threading.Thread):
url, port = got
real_url = select_source_url(url, port)
if not real_url:
self._p(f"未得到播放地址，跳过：{url}")
CLOG.warn(f"未得到播放地址，跳过：{url}", worker=self.worker_name)
self.coordinator.release(url, self.key)
time.sleep(1)
continue
# 新源上线前：探活（避免刚拿到 URL 就空推）
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
@@ -646,19 +915,22 @@ class StreamWorker(threading.Thread):
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}")
CLOG.ok(f"开始转播：{self.current_platform} | {self.current_anchor} | {url}", worker=self.worker_name)
except Exception as e:
# 🔧 关键：任何内部异常不让线程退出
logger.error(f"[{self.worker_name}] 内部异常：{e}")
# 关键：任何内部异常不让线程退出
CLOG.error(f"内部异常：{e}", worker=self.worker_name)
time.sleep(2)
continue
finally:
stop_ffmpeg(self.current_proc)
with contextlib.suppress(Exception):
stop_ffmpeg(self.current_proc)
self.coordinator.release(self.current_url, self.key)
self._p("已退出。")
CLOG.off("已退出。", worker=self.worker_name)
def _clear_state(self, release: bool = False):
if release and self.current_url:
@@ -674,39 +946,31 @@ 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])
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:
with contextlib.suppress(Exception):
w.stop()
except Exception:
pass
# 再兜底停止全局登记的 ffmpeg 子进程
with ALL_LOCK:
procs = list(ALL_PROCS)
for p in procs:
try:
with contextlib.suppress(Exception):
stop_ffmpeg(p)
except Exception:
pass
def main():
print("-----------------------------------------------------")
print("| DouyinLiveRecorder - 并行转推版 |")
print("-----------------------------------------------------")
print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder")
print("模式：并行检测 + 并行推流（保持原有业务逻辑）")
print(".....................................................")
print("DouyinRelay", version)
print("-----------------------------------------")
if not check_ffmpeg_existence():
logger.error("缺少 ffmpeg，程序退出")
@@ -742,14 +1006,10 @@ def main():
coordinator = UrlCoordinator(url_config_file)
# 信号处理（Ctrl+C / kill）
try:
with contextlib.suppress(Exception):
signal.signal(signal.SIGINT, graceful_shutdown)
except Exception:
pass
try:
with contextlib.suppress(Exception):
signal.signal(signal.SIGTERM, graceful_shutdown)
except Exception:
pass
# 启动多路（Worker 前缀区分）
workers: List[StreamWorker] = []