diff --git a/back/back2 b/back/back2 index 67bcfd0..530b0be 100644 --- a/back/back2 +++ b/back/back2 @@ -1,61 +1,56 @@ # -*- encoding: utf-8 -*- + +""" +单进程 ffmpeg 推流：从 URL_config.ini 顺序检测直播源， +一旦当前源停播/异常，自动切到下一个正在直播的源并推流到 YouTube。 +仅做“检测+转推”，不做录制、分段、通知等。 +""" + import asyncio import os import sys -import builtins -import subprocess import signal -import threading import time import datetime import re -import shutil -import random import uuid +import subprocess from pathlib import Path +from typing import Any, List, Tuple, Optional + +import configparser import urllib.request from urllib.error import URLError, HTTPError -from typing import Any -import configparser + import httpx from src import spider, stream -from src.proxy import ProxyDetector -from src.utils import logger -from src import utils -from colorama import Fore, Style +from src.utils import logger, Color, check_md5 +from ffmpeg_install import check_ffmpeg, ffmpeg_path, current_env_path -from msg_push import ( - dingtalk, xizhi, tg_bot, send_email, bark, ntfy, pushplus -) -from ffmpeg_install import ( - check_ffmpeg, ffmpeg_path, current_env_path -) -import configparser -import shlex -import subprocess +# --------------------- 基础环境 --------------------- +version = "relay-1.0.0" +os.environ['PATH'] = ffmpeg_path + os.pathsep + current_env_path +script_path = os.path.split(os.path.realpath(sys.argv[0]))[0] +config_file = f'{script_path}/config/config.ini' +url_config_file = f'{script_path}/config/URL_config.ini' +text_encoding = 'utf-8-sig' +color = Color() - - - -# 建议在程序初始化阶段全局加载一次 -_config = configparser.ConfigParser() -with open("config/youtube.ini", "r", encoding="utf-8-sig") as f: - _config.read_file(f) -# _config.read("config/config.ini", encoding="utf-8") +# ---------- YouTube 推流配置（读取 youtube.ini） ---------- +_yt_cfg = configparser.ConfigParser() +with open(f"{script_path}/config/youtube.ini", "r", encoding="utf-8-sig") as f: + _yt_cfg.read_file(f) def _yt_get(section: str, key: str, fallback=None, cast=str): try: if cast is bool: - return _config.getboolean(section, key, fallback=fallback) + return _yt_cfg.getboolean(section, key, fallback=fallback) if cast is int: - return _config.getint(section, key, fallback=fallback) - return _config.get(section, key, fallback=fallback) + return _yt_cfg.getint(section, key, fallback=fallback) + return _yt_cfg.get(section, key, fallback=fallback) except Exception: return fallback -# ===【单进程推流协调模式】全局控制 === -yt_single_mode = _yt_get("youtube", "single_mode", fallback=True, cast=bool) # 建议默认 True -yt_thread_started = False -yt_coordinator_lock = threading.Lock() + def _safe_int(v, lo, hi, default): try: v = int(v) @@ -65,10 +60,170 @@ def _safe_int(v, lo, hi, default): except Exception: return default +def _build_youtube_rtmp_url(key: str, secure: bool, ingest: str | None): + node = (ingest or "a").strip().lower() + node = node if node.isalpha() and len(node) == 1 else "a" + scheme = "rtmps" if secure else "rtmp" + host = f"{node}.rtmps.youtube.com" if secure else f"{node}.rtmp.youtube.com" + return f"{scheme}://{host}/live2/{key}" +# --------------------- 简化配置读取 --------------------- +def read_config_value(cfg: configparser.RawConfigParser, section: str, option: str, default_value: Any) -> Any: + try: + cfg.read(config_file, encoding=text_encoding) + if section not in cfg.sections(): + cfg.add_section(section) + return cfg.get(section, option) + except (configparser.NoSectionError, configparser.NoOptionError): + cfg.set(section, option, str(default_value)) + with open(config_file, 'w', encoding=text_encoding) as f: + cfg.write(f) + return default_value -def _yt_build_output_args(port_info, res_str, bv, ba, fps, gop, allow_copy): - # 输出端（向 YouTube）编码链 +_CN_BOOL = {"是": True, "否": False} + +cfg = configparser.RawConfigParser() +use_proxy = _CN_BOOL.get(read_config_value(cfg, '录制设置', '是否使用代理ip(是/否)', "否"), False) +proxy_addr = read_config_value(cfg, '录制设置', '代理地址', "") +proxy_addr = proxy_addr if use_proxy and proxy_addr.strip() else None +# Cookie（按需） +dy_cookie = read_config_value(cfg, 'Cookie', '抖音cookie', '') +tiktok_cookie = read_config_value(cfg, 'Cookie', 'tiktok_cookie', '') + +# 可选：启动时检测系统代理（仅提示） +try: + if not use_proxy: + urllib.request.urlopen("https://www.google.com/", timeout=5) + print("提示：你的网络似乎直连可用（无需代理）。") +except Exception: + if not use_proxy: + color.print_colored("提示：无法直连海外站，如需 TikTok 请在 config.ini 开启代理并填写代理地址。", color.YELLOW) + +# --------------------- URL 列表读取 --------------------- +def parse_urls_from_file(path: str) -> List[str]: + """ + 读取 URL_config.ini，返回去重且保持相对顺序的 URL 列表。 + 支持行内格式： + - 纯 URL + - quality,url + - url,主播: xxx + - # 注释行 忽略 + """ + urls, seen = [], set() + if not os.path.isfile(path): + open(path, 'w', encoding=text_encoding).close() + print("URL_config.ini 为空，请写入直播间地址后再运行。") + return urls + + with open(path, "r", encoding=text_encoding, errors='ignore') as f: + for line in f: + s = line.strip() + if not s or s.startswith("#"): + continue + # 拆分（最多3段，取包含 https 或域名的那段） + parts = re.split(r"[，,]\s*", s) + cand = None + for p in parts: + if '://' in p or re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", p): + cand = p + break + if not cand: + continue + url = cand if '://' in cand else f"https://{cand}" + if url not in seen: + seen.add(url) + urls.append(url) + return urls + +# --------------------- 直播解析与选源 --------------------- +def is_flv_preferred_platform(link: str) -> bool: + # 抖音/TikTok 优先走 FLV；但 H265 不走 FLV + return any(i in link for i in ["douyin", "tiktok"]) + +def select_source_url(link: str, stream_info: dict) -> Optional[str]: + if not stream_info: + return None + if is_flv_preferred_platform(link): + flv = stream_info.get("flv_url") + if flv: + # 若 codec=h265 则不走 flv + try: + from urllib.parse import urlparse, parse_qs + qs = parse_qs(urlparse(flv).query or "") + if (qs.get("codec") or [""])[0].lower() == "h265": + logger.warning("FLV 不支持 h265，改用 HLS。") + else: + return flv + except Exception: + return flv + return stream_info.get("record_url") or stream_info.get("m3u8_url") or stream_info.get("flv_url") + +async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: + """ + 调用项目现有 spider/stream，返回： + { + "anchor_name": str, + "is_live": bool, + "title": str?, + "record_url": "...", + "flv_url": "...", + "m3u8_url": "..." + } + """ + proxy = proxy_addr + port_info = {} + try: + if "douyin.com/" in url: + if 'v.douyin.com' not in url and '/user/' not in url: + j = await spider.get_douyin_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie) + else: + j = await spider.get_douyin_app_stream_data(url=url, proxy_addr=proxy, cookies=dy_cookie) + port_info = await stream.get_douyin_stream_url(j, record_quality_code, proxy) + elif url.startswith("https://www.tiktok.com/"): + if proxy: + j = await spider.get_tiktok_stream_data(url=url, proxy_addr=proxy, cookies=tiktok_cookie) + port_info = await stream.get_tiktok_stream_url(j, record_quality_code, proxy) + else: + logger.error("TikTok 需要代理，请在 config.ini 开启并配置代理。") + port_info = {"anchor_name": "", "is_live": False} + + elif (".m3u8" in url) or (".flv" in url): + # 自定义直链 + port_info = { + "anchor_name": "自定义_" + str(uuid.uuid4())[:8], + "is_live": True, + } + if url.endswith(".flv"): + port_info["flv_url"] = url + else: + port_info["m3u8_url"] = url + else: + # 其它站点可按需扩展 + port_info = {"anchor_name": "", "is_live": False} + except Exception as e: + logger.error(f"解析失败: {e}") + port_info = {"anchor_name": "", "is_live": False} + return port_info + +# --------------------- ffmpeg 单进程推流 --------------------- +def build_youtube_output() -> Tuple[str, list, list]: + key = _yt_get("youtube", "key", fallback="", cast=str) + if not key: + logger.error("错误: config/youtube.ini 未配置 key") + sys.exit(1) + + secure = _yt_get("youtube", "secure", fallback=False, cast=bool) + ingest = _yt_get("youtube", "ingest", fallback="a", cast=str) + res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() + + 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) + + # —— 默认强制转码（最稳），如果想省 CPU 可把 -c:v libx264 改为 copy —— # v_args = [ "-c:v", "libx264", "-preset", "veryfast", @@ -87,818 +242,112 @@ def _yt_build_output_args(port_info, res_str, bv, ba, fps, gop, allow_copy): "-ar", "44100", "-ac", "2", ] - vf_chain = [] + + # 可选缩放/留黑边，保持 YT 吃得更稳 if res_str: try: w, h = map(int, res_str.lower().split("x")) - vf_chain += [ - f"scale={w}:{h}:force_original_aspect_ratio=decrease", - f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2", - "format=yuv420p", - ] + vf = f"scale={w}:{h}:force_original_aspect_ratio=decrease,pad={w}:{h}:(ow-iw)/2:(oh-ih)/2,format=yuv420p" + v_args = ["-vf", vf] + v_args except Exception: - logger.warning(f"警告: resolution '{res_str}' 非法，已忽略") + logger.warning(f"resolution '{res_str}' 非法，忽略") - # 源若已是 h264+aac 且未强制缩放，尝试零转码 copy - if allow_copy and not vf_chain and _can_copy_h264_aac(port_info): - v_args = ["-c:v", "copy"] - a_args = ["-c:a", "copy"] + return out_url, v_args, a_args - base = ["-map", "0"] - if vf_chain: - base += ["-vf", ",".join(vf_chain)] - base += v_args + a_args + [ - "-flvflags", "no_duration_filesize", - "-rtmp_live", "live", - ] - return base - - -def _yt_make_input_chain(input_url, proxy_address, headers, - rw_timeout="15000000", analyzeduration="20000000", - probesize="10000000", bufsize="8000k", - max_muxing_queue_size="1024"): - # 输入端（从抖音等源拉流） +def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = None) -> subprocess.Popen: + """ + 启动一个 ffmpeg 进程：输入 input_url，输出 RTMP 到 out_url。 + """ user_agent = ("Mozilla/5.0 (Linux; Android 11; SAMSUNG SM-G973U) AppleWebKit/537.36 " "(KHTML, like Gecko) SamsungBrowser/14.2 Chrome/87.0.4280.141 Mobile Safari/537.36") - cmd = [ - "ffmpeg", "-y", - "-v", "verbose", + + rw_timeout = "15000000" + analyzeduration = "20000000" + probesize = "10000000" + bufsize = "8000k" + + base = [ + "ffmpeg", "-nostdin", "-hide_banner", + # 超时 & 重连（必须在 -i 前面） "-rw_timeout", rw_timeout, - "-loglevel", "error", - "-hide_banner", + "-timeout", "10000000", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_at_eof", "1", + "-reconnect_delay_max", "60", + + # UA + 协议 "-user_agent", user_agent, "-protocol_whitelist", "rtmp,crypto,file,http,https,tcp,tls,udp,rtp,httpproxy", + + # 缓冲 & 分析 "-thread_queue_size", "1024", "-analyzeduration", analyzeduration, "-probesize", probesize, "-fflags", "+discardcorrupt", + + # 输入流（放在重连参数之后） "-re", "-i", input_url, + + # 视频/音频编解码 + "-c:v", "copy", # 直接拷贝视频，省 CPU，低延迟 + "-c:a", "aac", # 音频转码成 AAC（保证 YouTube 兼容） + "-ar", "44100", # 音频采样率 + "-b:a", "128k", # 音频码率 + "-ac", "2", # 双声道 "-bufsize", bufsize, - "-sn", "-dn", - "-reconnect_delay_max", "60", - "-reconnect_streamed", "-reconnect_at_eof", - "-max_muxing_queue_size", max_muxing_queue_size, - "-correct_ts_overflow", "1", + "-max_muxing_queue_size", "2048", "-avoid_negative_ts", "1", + "-fflags", "+genpts", # 确保时间戳连续，防止花屏 + + # 输出容器 + "-f", "flv", out_url ] - if headers: - cmd.insert(11, "-headers") - cmd.insert(12, headers) - if proxy_address: - cmd.insert(1, "-http_proxy") - cmd.insert(2, proxy_address) - return cmd + if extra_headers: + base[ base.index("-user_agent") : base.index("-protocol_whitelist") ] += ["-headers", extra_headers] -def _try_open_rtmp_out(key, secure, ingest): - # 优先 rtMPS；失败自动轮换节点与协议 - nodes = [ingest] if ingest and len(ingest) == 1 else ["a"] - nodes += [c for c in "bcdef" if c not in nodes] - schemes = ["rtmps", "rtmp"] if secure else ["rtmp", "rtmps"] - for sch in schemes: - for node in nodes: - yield _build_youtube_rtmp_url(key, sch == "rtmps", node) + if proxy_addr: + base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:] + logger.info(f"启动推流：{input_url} -> YouTube") + p = subprocess.Popen(base, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + return p -def yt_single_coordinator(record_quality, proxy_address): - key = (_yt_get("youtube", "key", fallback="", cast=str) or "").strip() - if not key: - logger.error("[YT协调] ❌ youtube.ini[youtube].key 未配置或为空") +def stop_ffmpeg(p: Optional[subprocess.Popen]): + if not p: return - - secure = _yt_get("youtube", "secure", fallback=True, cast=bool) - ingest = (_yt_get("youtube", "ingest", fallback="a", cast=str) or "a").strip().lower() - res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() - bv = _safe_int(_yt_get("youtube", "bitrate", fallback=5000, cast=int), 300, 20000, 5000) - ba = _safe_int(_yt_get("youtube", "audio_bitrate", fallback=128, cast=int), 64, 384, 128) - fps = _safe_int(_yt_get("youtube", "fps", fallback=30, cast=int), 10, 60, 30) - gop = _safe_int(_yt_get("youtube", "gop", fallback=2*fps, cast=int), fps, 4*fps, 2*fps) - allow_copy = _yt_get("youtube", "copy_when_safe", fallback=True, cast=bool) - - out_url_candidates = list(_try_open_rtmp_out(key, secure, ingest)) - out_url_idx = 0 - - # 输入端时序阈值（海外可加大） - rw_timeout = "15000000"; analyzeduration = "20000000"; probesize = "10000000" - bufsize = "8000k"; max_muxing_queue_size = "1024" - - current_display_name = None # 面板上当前显示的条目名（带主播） - current_anchor = None - - while True: - # 1) 找到“第一条在播”的源 - real_url, port_info = find_next_live_source(None, proxy_address, record_quality) - if not real_url: - if current_display_name: - logger.info(f"[YT协调] ⏹️ {current_anchor or '未知主播'} 下播，清理面板显示") - recording.discard(current_display_name) - recording_time_list.pop(current_display_name, None) - current_display_name = None - current_anchor = None - time.sleep(3) - continue - - # 2) 更新面板显示名称为“转播·主播名” - anchor = clean_name(port_info.get("anchor_name") or "直播") - display_name = f"转播·{anchor}" - if display_name != current_display_name: - if current_display_name: - logger.info(f"[YT协调] 🔄 切换主播 {current_anchor} → {anchor}") - recording.discard(current_display_name) - recording_time_list.pop(current_display_name, None) - logger.info(f"[YT协调] ▶️ 开始转播 {anchor}") - recording.add(display_name) - # recording_time_list[display_name] = [datetime.datetime.now(), video_record_quality, real_url] - page_url = port_info.get("url") or real_url # 优先用抖音原始直播间链接 - recording_time_list[display_name] = [datetime.datetime.now(), video_record_quality, page_url] - - - current_display_name = display_name - current_anchor = anchor - - headers = None - in_cmd = _yt_make_input_chain(real_url, proxy_address, headers, - rw_timeout, analyzeduration, probesize, - bufsize, max_muxing_queue_size) - out_args = _yt_build_output_args(port_info, res_str, bv, ba, fps, gop, allow_copy) - - # 3) 逐个尝试 YouTube 出口（节点/协议） - tried = 0 - while tried < len(out_url_candidates): - out_url = out_url_candidates[out_url_idx % len(out_url_candidates)] - out_url_idx += 1 - tried += 1 - - run_cmd = list(in_cmd) + out_args + ["-f", "flv", out_url] - logger.info(f"[YT协调] 📡 {anchor} → {out_url} (尝试 {tried}/{len(out_url_candidates)})") - - proc = subprocess.Popen( - run_cmd, stdin=subprocess.PIPE, stderr=subprocess.STDOUT, - startupinfo=get_startup_info(os_type) - ) - - while proc.poll() is None: - time.sleep(1) - - rc = proc.returncode - if rc == 0: - logger.info(f"[YT协调] ✅ {anchor} 推流进程正常退出，重新寻找在播源") - break - else: - logger.warning(f"[YT协调] ⚠️ {anchor} 推流异常退出 rc={rc}，切换节点重试…") - - # 4) 当前源结束/异常：回到循环顶部去找下一路在播 - time.sleep(1) - -def read_all_urls_from_list(url_file: str) -> list[tuple[str, str]]: - """ - 读取 URL_config.ini，返回 [(url, anchor_hint), ...] - anchor_hint 可能为空字符串 - """ - items = [] try: - if not os.path.isfile(url_file): - return items - with open(url_file, 'r', encoding=text_encoding, errors='ignore') as f: - for origin_line in f: - line = origin_line.strip() - if not line or line.startswith('#'): - continue - # 兼容 "url,主播: 名字" / "清晰度,url,主播: 名字" - parts = re.split('[,，]', line) - if len(parts) == 1: - url, anchor = parts[0], '' - elif len(parts) == 2: - if '://' in parts[0]: - url, anchor = parts - else: - # parts[0] 是清晰度，忽略它 - url, anchor = parts[1], '' - else: - # 质量, url, 主播 - _, url, anchor = parts[0], parts[1], parts[2] if len(parts) > 2 else '' - url = 'https://' + url if '://' not in url else url - items.append((url.strip(), anchor.strip())) - except Exception as e: - # logger.error(f"读取 URL 列表失败: {e}") - line = getattr(e, "__traceback__", None).tb_lineno if getattr(e, "__traceback__", None) else "?" - logger.error(f"{str(e)} (line {line})") - - return items - -def probe_douyin_live_and_stream(url: str, proxy_address: str | None, record_quality: str | None): - """ - 只针对抖音/抖音短链，复用你现有的 spider/stream，返回 (is_live, real_url, port_info) - 失败返回 (False, None, None) - """ - try: - if 'douyin.com/' not in url: - return False, None, None - if 'v.douyin.com' not in url and '/user/' not in url: - json_data = asyncio.run(spider.get_douyin_stream_data(url=url, proxy_addr=proxy_address, cookies=dy_cookie)) + if os.name == "nt": + if p.stdin: + try: + p.stdin.write(b"q") + p.stdin.flush() + except Exception: + pass else: - json_data = asyncio.run(spider.get_douyin_app_stream_data(url=url, proxy_addr=proxy_address, cookies=dy_cookie)) - port_info = asyncio.run(stream.get_douyin_stream_url(json_data, record_quality, proxy_address)) - if not port_info: - return False, None, None - if not port_info.get("is_live"): - return False, None, port_info - real_url = select_source_url(url, port_info) - return True, real_url, port_info - except Exception as e: - logger.debug(f"探测抖音直播失败: {e}") - return False, None, None - - - -async def parse_stream(url: str, proxy_addr=None, cookies=None) -> dict: - """根据 URL 调用不同平台的解析函数""" - if "live.douyin.com" in url: - return await get_douyin_stream_data(url, proxy_addr=proxy_addr, cookies=cookies) - elif "tiktok.com" in url: - return await get_tiktok_stream_data(url, proxy_addr=proxy_addr, cookies=cookies) - elif "kuaishou.com" in url: - return await get_kuaishou_stream_data(url, proxy_addr=proxy_addr, cookies=cookies) - elif "huya.com" in url: - return await get_huya_stream_data(url, proxy_addr=proxy_addr, cookies=cookies) - elif "bilibili.com" in url: - return await get_bilibili_stream_data(url, proxy_addr=proxy_addr, cookies=cookies) - else: - raise ValueError(f"暂不支持的直播平台: {url}") - - -async def find_next_live_source(url: str, proxy_addr=None, cookies=None) -> dict | None: - """检测直播状态并返回直播信息""" - logger.debug(f"{Fore.CYAN}[检测] 正在检测 {url} ...{Style.RESET_ALL}") - try: - data = await parse_stream(url, proxy_addr=proxy_addr, cookies=cookies) - - if not data: - logger.warning(f"{Fore.YELLOW}[检测] {url} 返回空数据{Style.RESET_ALL}") - return None - - # 通用字段处理 - anchor = data.get("anchor_name", "未知主播") - title = data.get("title", "") - is_live = False - - # 不同平台的开播字段 - if data.get("status") == 2 or data.get("is_live") is True or data.get("live_status") == 1: - is_live = True - - if is_live: - # 优先取抖音原始 URL - record_url = None - if "douyin" in url and "stream_url" in data: - flv_map = data["stream_url"].get("flv_pull_url") or {} - record_url = flv_map.get("ORIGIN") or list(flv_map.values())[0] - - if not record_url: - record_url = data.get("record_url") or data.get("flv_url") or data.get("m3u8_url") - - logger.info( - f"{Fore.GREEN}[检测成功] {anchor} 已开播 | 标题: {title} | 地址: {record_url}{Style.RESET_ALL}" - ) - return { - "anchor_name": anchor, - "title": title, - "record_url": record_url, - "raw_data": data, - } - else: - logger.debug(f"{Fore.MAGENTA}[检测未开播] {anchor} -> {url}{Style.RESET_ALL}") - return None - - except Exception as e: - logger.warning(f"{Fore.RED}[检测失败] {url} 解析异常: {e}{Style.RESET_ALL}") - return None - -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 _can_copy_h264_aac(port_info: dict) -> bool: - """ - 可选：根据你前面已解析到的源信息判断是否允许 copy。 - 如果你没有ffprobe，这里保守返回False；若你在port_info里塞了codec信息，按需判断。 - """ - # 示例：只有在明确知道是 h264 + aac 时才放行 - v = (port_info or {}).get("vcodec", "").lower() - a = (port_info or {}).get("acodec", "").lower() - if not v or not a: - return False - if v != "h264": - return False - # YouTube 推荐 44.1k/48k，立体声 OK - return a in ("aac", "mp4a") -version = "v4.0.6" -platforms = ("\n抖音") - -recording = set() -error_count = 0 -pre_max_request = 10 -max_request_lock = threading.Lock() -error_window = [] -error_window_size = 10 -error_threshold = 5 -monitoring = 0 -running_list = [] -url_tuples_list = [] -url_comments = [] -text_no_repeat_url = [] -create_var = locals() -first_start = True -exit_recording = False -need_update_line_list = [] -first_run = True -not_record_list = [] -start_display_time = datetime.datetime.now() -global_proxy = False -recording_time_list = {} -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' -backup_dir = f'{script_path}/backup_config' -text_encoding = 'utf-8-sig' -rstr = r"[\/\\\:\*\？?\"\<\>\|&#.。,， ~！· ]" -default_path = f'{script_path}/downloads' -os.makedirs(default_path, exist_ok=True) -file_update_lock = threading.Lock() -os_type = os.name -clear_command = "cls" if os_type == 'nt' else "clear" -color_obj = utils.Color() -os.environ['PATH'] = ffmpeg_path + os.pathsep + current_env_path - - -def signal_handler(_signal, _frame): - sys.exit(0) - - -signal.signal(signal.SIGTERM, signal_handler) - - - - -def _fmt_td(td: datetime.timedelta) -> str: - # 统一格式：HH:MM:SS（超过24h也按总小时数显示） - total = int(td.total_seconds()) - if total < 0: - total = 0 - h = total // 3600 - m = (total % 3600) // 60 - s = total % 60 - return f"{h:02d}:{m:02d}:{s:02d}" - -def _hr(char: str = "─") -> str: - cols = shutil.get_terminal_size((100, 20)).columns - return char * max(20, cols) - -def display_info() -> None: - """控制台【直播动态】面板（定时清屏重绘）""" - global start_display_time - time.sleep(1.0) # 稍等程序其余线程初始化 - start_display_time = datetime.datetime.now() - - while True: + p.send_signal(signal.SIGINT) + p.wait(timeout=10) + except Exception: try: - # 清屏 - if Path(sys.executable).name != 'pythonw.exe': - os.system(clear_command) + p.kill() + except Exception: + pass - now_dt = datetime.datetime.now() - now_str = now_dt.strftime("%Y-%m-%d %H:%M:%S") - - # 顶部抬头 - print("【直播动态】") - print(_hr()) - - # 统计区 - stat_parts = [] - stat_parts.append(f"监测中: {monitoring}") - stat_parts.append(f"并发线程: {max_request}") - stat_parts.append(f"代理直播: {'是' if use_proxy else '否'}") - stat_parts.append(f"分段: {f'{split_time}s' if split_video_by_time else '否'}") - if create_time_file: - stat_parts.append("时间字幕: 是") - stat_parts.append(f"质量: {video_record_quality}") - stat_parts.append(f"格式: {video_save_type}") - stat_parts.append(f"瞬时错误: {error_count}") - stat_line = " | ".join(stat_parts) - print(stat_line) - print(f"当前时间: {now_str}") - print(_hr()) - - # 列表区 - if not recording: - if monitoring == 0: - print("暂无监测任务。") - else: - print(f"暂无正在直播的直播。下次检测间隔：{delay_default} 秒") - else: - no_repeat = list(set(recording)) - print(f"正在直播：{len(no_repeat)} 个") - print(_hr("·")) - - # 标题行 - print(f"{'序号/主播名':<28} {'清晰度':<6} {'开始时间':<19} {'已直播':<10}") - print(_hr("·")) - - for name in sorted(no_repeat): - start_time, qa = recording_time_list.get(name, (now_dt, "未知")) - elapsed = _fmt_td(now_dt - start_time) - start_str = start_time.strftime("%H:%M:%S") if isinstance(start_time, datetime.datetime) else "--:--:--" - - # name 可能很长，截断以适配终端宽度 - cols = shutil.get_terminal_size((100, 20)).columns - name_col = 28 - # 留余量：28 + 1 + 6 + 1 + 19 + 1 + 10 ≈ 65 列基础宽度 - disp_name = name if len(name) <= name_col else (name[:name_col-2] + "…") - print(f"{disp_name:<28} {str(qa):<6} {start_str:<19} {elapsed:<10}") - - print(_hr()) - - # 运行时长（可选） - uptime = _fmt_td(now_dt - start_display_time) - print(f"已运行时长：{uptime}") - - # 降低刷新频率，避免刷屏 - sys.stdout.flush() - time.sleep(2.0) - - except KeyboardInterrupt: - # 主线程Ctrl+C时，优雅退出这个展示线程 - break - except Exception as e: - # 面板渲染异常不能影响主流程 - try: - logger.error(f"显示面板异常: {e} | 行: {getattr(e, '__traceback__', None).tb_lineno if getattr(e, '__traceback__', None) else '未知'}") - except Exception: - pass - time.sleep(1.0) - - -def update_file(file_path: str, old_str: str, new_str: str, start_str: str = None) -> str | None: - if old_str == new_str and start_str is None: - return old_str - with file_update_lock: - file_data = [] - with open(file_path, "r", encoding=text_encoding) as f: - try: - for text_line in f: - if old_str in text_line: - text_line = text_line.replace(old_str, new_str) - if start_str: - text_line = f'{start_str}{text_line}' - if text_line not in file_data: - file_data.append(text_line) - except RuntimeError as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - if ini_URL_content: - with open(file_path, "w", encoding=text_encoding) as f2: - f2.write(ini_URL_content) - return old_str - if file_data: - with open(file_path, "w", encoding=text_encoding) as f: - f.write(''.join(file_data)) - return new_str - - -def delete_line(file_path: str, del_line: str, delete_all: bool = False) -> None: - with file_update_lock: - with open(file_path, 'r+', encoding=text_encoding) as f: - lines = f.readlines() - f.seek(0) - f.truncate() - skip_line = False - for txt_line in lines: - if del_line in txt_line: - if delete_all or not skip_line: - skip_line = True - continue - else: - skip_line = False - f.write(txt_line) - - -def get_startup_info(system_type: str): - if system_type == 'nt': - startup_info = subprocess.STARTUPINFO() - startup_info.dwFlags |= subprocess.STARTF_USESHOWWINDOW - else: - startup_info = None - return startup_info - - -def segment_video(converts_file_path: str, segment_save_file_path: str, segment_format: str, segment_time: str, - is_original_delete: bool = True) -> None: +# --------------------- 主循环（单进程转推+切源） --------------------- +def check_ffmpeg_existence() -> bool: try: - if os.path.exists(converts_file_path) and os.path.getsize(converts_file_path) > 0: - ffmpeg_command = [ - "ffmpeg", - "-i", converts_file_path, - "-c:v", "copy", - "-c:a", "copy", - "-map", "0", - "-f", "segment", - "-segment_time", segment_time, - "-segment_format", segment_format, - "-reset_timestamps", "1", - "-movflags", "+frag_keyframe+empty_moov", - segment_save_file_path, - ] - _output = subprocess.check_output( - ffmpeg_command, stderr=subprocess.STDOUT, startupinfo=get_startup_info(os_type) - ) - if is_original_delete: - time.sleep(1) - if os.path.exists(converts_file_path): - os.remove(converts_file_path) - except subprocess.CalledProcessError as e: - logger.error(f'Error occurred during conversion: {e}') - except Exception as e: - logger.error(f'An unknown error occurred: {e}') + result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True) + if result.returncode == 0: + print(result.stdout.splitlines()[0]) + except Exception: + pass + finally: + return bool(check_ffmpeg()) +def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") -def converts_mp4(converts_file_path: str, is_original_delete: bool = True) -> None: - try: - if os.path.exists(converts_file_path) and os.path.getsize(converts_file_path) > 0: - if converts_to_h264: - color_obj.print_colored("正在转码为MP4格式并重新编码为h264\n", color_obj.YELLOW) - ffmpeg_command = [ - "ffmpeg", "-i", converts_file_path, - "-c:v", "libx264", - "-preset", "veryfast", - "-crf", "23", - "-vf", "format=yuv420p", - "-c:a", "copy", - "-f", "mp4", converts_file_path.rsplit('.', maxsplit=1)[0] + ".mp4", - ] - else: - color_obj.print_colored("正在转码为MP4格式\n", color_obj.YELLOW) - ffmpeg_command = [ - "ffmpeg", "-i", converts_file_path, - "-c:v", "copy", - "-c:a", "copy", - "-f", "mp4", converts_file_path.rsplit('.', maxsplit=1)[0] + ".mp4", - ] - _output = subprocess.check_output( - ffmpeg_command, stderr=subprocess.STDOUT, startupinfo=get_startup_info(os_type) - ) - if is_original_delete: - time.sleep(1) - if os.path.exists(converts_file_path): - os.remove(converts_file_path) - except subprocess.CalledProcessError as e: - logger.error(f'Error occurred during conversion: {e}') - except Exception as e: - logger.error(f'An unknown error occurred: {e}') - - -def converts_m4a(converts_file_path: str, is_original_delete: bool = True) -> None: - try: - if os.path.exists(converts_file_path) and os.path.getsize(converts_file_path) > 0: - _output = subprocess.check_output([ - "ffmpeg", "-i", converts_file_path, - "-n", "-vn", - "-c:a", "aac", "-bsf:a", "aac_adtstoasc", "-ab", "320k", - converts_file_path.rsplit('.', maxsplit=1)[0] + ".m4a", - ], stderr=subprocess.STDOUT, startupinfo=get_startup_info(os_type)) - if is_original_delete: - time.sleep(1) - if os.path.exists(converts_file_path): - os.remove(converts_file_path) - except subprocess.CalledProcessError as e: - logger.error(f'Error occurred during conversion: {e}') - except Exception as e: - logger.error(f'An unknown error occurred: {e}') - - -def generate_subtitles(record_name: str, ass_filename: str, sub_format: str = 'srt') -> None: - index_time = 0 - today = datetime.datetime.now() - re_datatime = today.strftime('%Y-%m-%d %H:%M:%S') - - def transform_int_to_time(seconds: int) -> str: - m, s = divmod(seconds, 60) - h, m = divmod(m, 60) - return f"{h:02d}:{m:02d}:{s:02d}" - - while True: - index_time += 1 - txt = str(index_time) + "\n" + transform_int_to_time(index_time) + ',000 --> ' + transform_int_to_time( - index_time + 1) + ',000' + "\n" + str(re_datatime) + "\n\n" - - with open(f"{ass_filename}.{sub_format.lower()}", 'a', encoding=text_encoding) as f: - f.write(txt) - - if record_name not in recording: - return - time.sleep(1) - today = datetime.datetime.now() - re_datatime = today.strftime('%Y-%m-%d %H:%M:%S') - - -def adjust_max_request() -> None: - global max_request, error_count, pre_max_request, error_window - preset = max_request - - while True: - time.sleep(5) - with max_request_lock: - if error_window: - error_rate = sum(error_window) / len(error_window) - else: - error_rate = 0 - - if error_rate > error_threshold: - max_request = max(1, max_request - 1) - elif error_rate < error_threshold / 2 and max_request < preset: - max_request += 1 - else: - pass - - if pre_max_request != max_request: - pre_max_request = max_request - print(f"\r同一时间访问网络的线程数动态改为 {max_request}") - - error_window.append(error_count) - if len(error_window) > error_window_size: - error_window.pop(0) - error_count = 0 - - -def push_message(record_name: str, live_url: str, content: str) -> None: - msg_title = push_message_title.strip() or "直播间状态更新通知" - push_functions = { - '微信': lambda: xizhi(xizhi_api_url, msg_title, content), - '钉钉': lambda: dingtalk(dingtalk_api_url, content, dingtalk_phone_num, dingtalk_is_atall), - '邮箱': lambda: send_email( - email_host, login_email, email_password, sender_email, sender_name, - to_email, msg_title, content, smtp_port, open_smtp_ssl - ), - 'TG': lambda: tg_bot(tg_chat_id, tg_token, content), - 'BARK': lambda: bark( - bark_msg_api, title=msg_title, content=content, level=bark_msg_level, sound=bark_msg_ring - ), - 'NTFY': lambda: ntfy( - ntfy_api, title=msg_title, content=content, tags=ntfy_tags, action_url=live_url, email=ntfy_email - ), - 'PUSHPLUS': lambda: pushplus(pushplus_token, msg_title, content), - } - - for platform, func in push_functions.items(): - if platform in live_status_push.upper(): - try: - result = func() - print(f'提示信息：已经将[{record_name}]直播状态消息推送至你的{platform},' - f' 成功{len(result["success"])}, 失败{len(result["error"])}') - except Exception as e: - color_obj.print_colored(f"直播消息推送到{platform}失败: {e}", color_obj.RED) - - -def run_script(command: str) -> None: - try: - process = subprocess.Popen( - command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, startupinfo=get_startup_info(os_type) - ) - stdout, stderr = process.communicate() - stdout_decoded = stdout.decode('utf-8') - stderr_decoded = stderr.decode('utf-8') - if stdout_decoded.strip(): - print(stdout_decoded) - if stderr_decoded.strip(): - print(stderr_decoded) - except PermissionError as e: - logger.error(e) - logger.error('脚本无执行权限!, 若是Linux环境, 请先执行:chmod +x your_script.sh 授予脚本可执行权限') - except OSError as e: - logger.error(e) - logger.error('Please add `#!/bin/bash` at the beginning of your bash script file.') - - -def clear_record_info(record_name: str, record_url: str) -> None: - global monitoring - recording.discard(record_name) - if record_url in url_comments and record_url in running_list: - running_list.remove(record_url) - monitoring -= 1 - color_obj.print_colored(f"[{record_name}]已经从直播列表中移除\n", color_obj.YELLOW) - - -def direct_download_stream(source_url: str, save_path: str, record_name: str, live_url: str, platform: str) -> bool: - - try: - with open(save_path, 'wb') as f: - client = httpx.Client(timeout=None) - - headers = {} - header_params = get_record_headers(platform, live_url) - if header_params: - key, value = header_params.split(":", 1) - headers[key] = value - - with client.stream('GET', source_url, headers=headers, follow_redirects=True) as response: - if response.status_code != 200: - logger.error(f"请求直播流失败，状态码: {response.status_code}") - return False - - downloaded = 0 - chunk_size = 1024 * 16 - - for chunk in response.iter_bytes(chunk_size): - if live_url in url_comments or exit_recording: - color_obj.print_colored(f"[{record_name}]直播时已被注释或请求停止,下载中断", color_obj.YELLOW) - clear_record_info(record_name, live_url) - return False - - if chunk: - f.write(chunk) - downloaded += len(chunk) - print() - return True - except Exception as e: - logger.error(f"FLV下载错误: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - return False - -import asyncio -import datetime -import subprocess - -async def check_subprocess_with_heartbeat(record_name: str, - record_url: str, - ffmpeg_command: list, - save_type: str, - script_command: str | None = None, - heartbeat_timeout: int = 10) -> bool: - """ - 启动 ffmpeg 子进程并进行心跳检测： - - 每 heartbeat_timeout 秒内必须有输出，否则认为卡死，直接 kill。 - - 进程退出 returncode=0 正常，否则异常。 - """ - process = await asyncio.create_subprocess_exec( - *ffmpeg_command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - - last_heartbeat = datetime.datetime.now() - - async def read_stream(stream, name): - nonlocal last_heartbeat - async for line in stream: - decoded = line.decode(errors="ignore").strip() - if decoded: - last_heartbeat = datetime.datetime.now() - # 只保留关键指标 - if "frame=" in decoded and "bitrate=" in decoded: - logger.info(f"[心跳] {record_name} {decoded}") - elif "error" in decoded.lower(): - logger.error(f"[FFmpeg错误] {record_name}: {decoded}") - else: - logger.debug(f"[{record_name}][{name}] {decoded}") - - asyncio.create_task(read_stream(process.stdout, "stdout")) - asyncio.create_task(read_stream(process.stderr, "stderr")) - - while True: - if process.returncode is not None: - break - - now = datetime.datetime.now() - if (now - last_heartbeat).total_seconds() > heartbeat_timeout: - logger.warning(f"[心跳超时] {record_name}: {heartbeat_timeout}s 无数据 → 强制重启 ffmpeg") - process.kill() - await process.wait() - return False - - await asyncio.sleep(2) - - return process.returncode == 0 - -def clean_name(input_text): - cleaned_name = re.sub(rstr, "_", input_text.strip()).strip('_') - cleaned_name = cleaned_name.replace("（", "(").replace("）", ")") - if clean_emoji: - cleaned_name = utils.remove_emojis(cleaned_name, '_').strip('_') - return cleaned_name or '空白昵称' - - -def get_quality_code(qn): - QUALITY_MAPPING = { - "原画": "OD", - "蓝光": "BD", - "超清": "UHD", - "高清": "HD", - "标清": "SD", - "流畅": "LD" - } - return QUALITY_MAPPING.get(qn) - -def get_record_headers(platform, live_url): +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', @@ -911,1171 +360,149 @@ def get_record_headers(platform, live_url): 'shopee': f'origin:{live_domain}', 'Blued直播': 'referer:https://app.blued.cn' } - return record_headers.get(platform) - - -def is_flv_preferred_platform(link): - return any(i in link for i in ["douyin", "tiktok"]) - - -def select_source_url(link, stream_info): - if is_flv_preferred_platform(link): - codec = utils.get_query_params(stream_info.get('flv_url'), "codec") - if codec and codec[0] == 'h265': - logger.warning("FLV is not supported for h265 codec, use HLS source instead") - else: - return stream_info.get('flv_url') - - return stream_info.get('record_url') - - -def start_record(url_data: tuple, count_variable: int = -1) -> None: - global error_count - - while True: - try: - record_finished = False - run_once = False - start_pushed = False - new_record_url = '' - count_time = time.time() - retry = 0 - record_quality_zh, record_url, anchor_name = url_data - record_quality = get_quality_code(record_quality_zh) - proxy_address = proxy_addr - platform = '未知平台' - live_domain = '/'.join(record_url.split('/')[0:3]) - - if proxy_addr: - proxy_address = None - for platform in enable_proxy_platform_list: - if platform and platform.strip() in record_url: - proxy_address = proxy_addr - break - - if not proxy_address: - if extra_enable_proxy_platform_list: - for pt in extra_enable_proxy_platform_list: - if pt and pt.strip() in record_url: - proxy_address = proxy_addr_bak or None - - # print(f'\r代理地址:{proxy_address}') - # print(f'\r全局代理:{global_proxy}') - while True: - try: - port_info = [] - if record_url.find("douyin.com/") > -1: - platform = '抖音直播' - with semaphore: - if 'v.douyin.com' not in record_url and '/user/' not in record_url: - json_data = asyncio.run(spider.get_douyin_stream_data( - url=record_url, - proxy_addr=proxy_address, - cookies=dy_cookie)) - else: - json_data = asyncio.run(spider.get_douyin_app_stream_data( - url=record_url, - proxy_addr=proxy_address, - cookies=dy_cookie)) - port_info = asyncio.run( - stream.get_douyin_stream_url(json_data, record_quality, proxy_address)) - - elif record_url.find("https://www.tiktok.com/") > -1: - platform = 'TikTok直播' - with semaphore: - if global_proxy or proxy_address: - json_data = asyncio.run(spider.get_tiktok_stream_data( - url=record_url, - proxy_addr=proxy_address, - cookies=tiktok_cookie)) - port_info = asyncio.run( - stream.get_tiktok_stream_url(json_data, record_quality, proxy_address)) - else: - logger.error("错误信息: 网络异常，请检查网络是否能正常访问TikTok平台") - - - elif record_url.find(".m3u8") > -1 or record_url.find(".flv") > -1: - platform = '自定义直播直播' - port_info = { - "anchor_name": platform + '_' + str(uuid.uuid4())[:8], - "is_live": True, - "record_url": record_url, - } - if '.flv' in record_url: - port_info['flv_url'] = record_url - else: - port_info['m3u8_url'] = record_url - - else: - logger.error(f'{record_url} {platform}直播地址') - return - - if anchor_name: - if '主播:' in anchor_name: - anchor_split: list = anchor_name.split('主播:') - if len(anchor_split) > 1 and anchor_split[1].strip(): - anchor_name = anchor_split[1].strip() - else: - anchor_name = port_info.get("anchor_name", '') - else: - anchor_name = port_info.get("anchor_name", '') - - if not port_info.get("anchor_name", ''): - print(f'序号{count_variable} 网址内容获取失败,进行重试中...获取失败的地址是:{url_data}') - with max_request_lock: - error_count += 1 - error_window.append(1) - else: - anchor_name = clean_name(anchor_name) - record_name = f'序号{count_variable} {anchor_name}' - - if record_url in url_comments: - print(f"[{anchor_name}]已被注释,本条线程将会退出") - clear_record_info(record_name, record_url) - return - - if not url_data[-1] and run_once is False: - if new_record_url: - need_update_line_list.append( - f'{record_url}|{new_record_url},主播: {anchor_name.strip()}') - not_record_list.append(new_record_url) - else: - need_update_line_list.append(f'{record_url}|{record_url},主播: {anchor_name.strip()}') - run_once = True - - push_at = datetime.datetime.today().strftime('%Y-%m-%d %H:%M:%S') - if port_info['is_live'] is False: - print(f"\r{record_name} 等待直播... ") - - if start_pushed: - if over_show_push: - push_content = "直播间状态更新：[直播间名称] 直播已结束！时间：[时间]" - if over_push_message_text: - push_content = over_push_message_text - - push_content = (push_content.replace('[直播间名称]', record_name). - replace('[时间]', push_at)) - threading.Thread( - target=push_message, - args=(record_name, record_url, push_content.replace(r'\n', '\n')), - daemon=True - ).start() - start_pushed = False - - else: - content = f"\r{record_name} 正在直播中..." - print(content) - - if live_status_push and not start_pushed: - if begin_show_push: - push_content = "直播间状态更新：[直播间名称] 正在直播中，时间：[时间]" - if begin_push_message_text: - push_content = begin_push_message_text - - push_content = (push_content.replace('[直播间名称]', record_name). - replace('[时间]', push_at)) - threading.Thread( - target=push_message, - args=(record_name, record_url, push_content.replace(r'\n', '\n')), - daemon=True - ).start() - start_pushed = True - - if disable_record: - time.sleep(push_check_seconds) - continue - - real_url = select_source_url(record_url, port_info) - full_path = f'{default_path}/{platform}' - if real_url: - now = datetime.datetime.today().strftime("%Y-%m-%d_%H-%M-%S") - live_title = port_info.get('title') - title_in_name = '' - if live_title: - live_title = clean_name(live_title) - title_in_name = live_title + '_' if filename_by_title else '' - - try: - if len(video_save_path) > 0: - if not video_save_path.endswith(('/', '\\')): - full_path = f'{video_save_path}/{platform}' - else: - full_path = f'{video_save_path}{platform}' - - full_path = full_path.replace("\\", '/') - if folder_by_author: - full_path = f'{full_path}/{anchor_name}' - if folder_by_time: - full_path = f'{full_path}/{now[:10]}' - if folder_by_title and port_info.get('title'): - if folder_by_time: - full_path = f'{full_path}/{live_title}_{anchor_name}' - else: - full_path = f'{full_path}/{now[:10]}_{live_title}' - if not os.path.exists(full_path): - os.makedirs(full_path) - except Exception as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - - if platform != '自定义直播直播': - if enable_https_recording and real_url.startswith("http://"): - real_url = real_url.replace("http://", "https://") - - http_record_list = ['shopee', "migu"] - if platform in http_record_list: - real_url = real_url.replace("https://", "http://") - - user_agent = ("Mozilla/5.0 (Linux; Android 11; SAMSUNG SM-G973U) AppleWebKit/537.36 (" - "KHTML, like Gecko) SamsungBrowser/14.2 Chrome/87.0.4280.141 Mobile " - "Safari/537.36") - - rw_timeout = "15000000" - analyzeduration = "20000000" - probesize = "10000000" - bufsize = "8000k" - max_muxing_queue_size = "1024" - for pt_host in overseas_platform_host: - if pt_host in record_url: - rw_timeout = "50000000" - analyzeduration = "40000000" - probesize = "20000000" - bufsize = "15000k" - max_muxing_queue_size = "2048" - break - - ffmpeg_command = [ - 'ffmpeg', "-y", - "-v", "verbose", - "-rw_timeout", rw_timeout, - "-loglevel", "error", - "-hide_banner", - "-user_agent", user_agent, - "-protocol_whitelist", "rtmp,crypto,file,http,https,tcp,tls,udp,rtp,httpproxy", - "-thread_queue_size", "1024", - "-analyzeduration", analyzeduration, - "-probesize", probesize, - "-fflags", "+discardcorrupt", - "-re", "-i", real_url, - "-bufsize", bufsize, - "-sn", "-dn", - "-reconnect_delay_max", "60", - "-reconnect_streamed", "-reconnect_at_eof", - "-max_muxing_queue_size", max_muxing_queue_size, - "-correct_ts_overflow", "1", - "-avoid_negative_ts", "1" - ] - - headers = get_record_headers(platform, record_url) - if headers: - ffmpeg_command.insert(11, "-headers") - ffmpeg_command.insert(12, headers) - - if proxy_address: - ffmpeg_command.insert(1, "-http_proxy") - ffmpeg_command.insert(2, proxy_address) - - recording.add(record_name) - start_record_time = datetime.datetime.now() - recording_time_list[record_name] = [start_record_time, record_quality_zh] - rec_info = f"\r{anchor_name} 准备开始直播视频: {full_path}" - if show_url: - re_plat = ('WinkTV', 'PandaTV', 'ShowRoom', 'CHZZK', 'Youtube') - if platform in re_plat: - logger.info( - f"{platform} | {anchor_name} | 直播源地址: {port_info.get('m3u8_url')}") - else: - logger.info( - f"{platform} | {anchor_name} | 直播源地址: {real_url}") - - only_flv_record = False - only_flv_platform_list = ['shopee', '花椒直播'] - if platform in only_flv_platform_list: - logger.debug(f"提示: {platform} 将强制使用FLV格式直播") - only_flv_record = True - - only_audio_record = False - only_audio_platform_list = ['猫耳FM直播', 'Look直播'] - if platform in only_audio_platform_list: - only_audio_record = True - - record_save_type = video_save_type - - if is_flv_preferred_platform(record_url) and port_info.get('flv_url'): - codec = utils.get_query_params(port_info['flv_url'], "codec") - if codec and codec[0] == 'h265': - logger.warning("FLV is not supported for h265 codec, use TS format instead") - record_save_type = "TS" - - if only_audio_record or any(i in record_save_type for i in ['MP3', 'M4A']): - try: - now = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) - extension = "mp3" if "m4a" not in record_save_type.lower() else "m4a" - name_format = "_%03d" if split_video_by_time else "" - save_file_path = (f"{full_path}/{anchor_name}_{title_in_name}{now}" - f"{name_format}.{extension}") - - if split_video_by_time: - print(f'\r{anchor_name} 准备开始直播音频: {save_file_path}') - - if "MP3" in record_save_type: - command = [ - "-map", "0:a", - "-c:a", "libmp3lame", - "-ab", "320k", - "-f", "segment", - "-segment_time", split_time, - "-reset_timestamps", "1", - save_file_path, - ] - else: - command = [ - "-map", "0:a", - "-c:a", "aac", - "-bsf:a", "aac_adtstoasc", - "-ab", "320k", - "-f", "segment", - "-segment_time", split_time, - "-segment_format", 'mpegts', - "-reset_timestamps", "1", - save_file_path, - ] - - else: - if "MP3" in record_save_type: - command = [ - "-map", "0:a", - "-c:a", "libmp3lame", - "-ab", "320k", - save_file_path, - ] - - else: - command = [ - "-map", "0:a", - "-c:a", "aac", - "-bsf:a", "aac_adtstoasc", - "-ab", "320k", - "-movflags", "+faststart", - save_file_path, - ] - - ffmpeg_command.extend(command) - # comment_end = check_subprocess( - # record_name, - # record_url, - # ffmpeg_command, - # record_save_type, - # custom_script - # ) - - # comment_end = asyncio.run( - # check_subprocess_with_heartbeat( - # record_name, record_url, ffmpeg_command, record_save_type, custom_script, heartbeat_timeout=15 - # )) - - comment_end = asyncio.run( - check_subprocess_with_heartbeat( - record_name, - record_url, - ffmpeg_command, - record_save_type, - custom_script, - heartbeat_timeout=10 # 建议 10s - ) - ) - - - - if comment_end: - return - - except subprocess.CalledProcessError as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - - if only_flv_record: - logger.info(f"Use Direct Downloader to Download FLV Stream: {record_url}") - filename = anchor_name + f'_{title_in_name}' + now + '.flv' - save_file_path = f'{full_path}/{filename}' - print(f'{rec_info}/{filename}') - - subs_file_path = save_file_path.rsplit('.', maxsplit=1)[0] - subs_thread_name = f'subs_{Path(subs_file_path).name}' - if create_time_file: - create_var[subs_thread_name] = threading.Thread( - target=generate_subtitles, args=(record_name, subs_file_path) - ) - create_var[subs_thread_name].daemon = True - create_var[subs_thread_name].start() - - try: - flv_url = port_info.get('flv_url') - if flv_url: - recording.add(record_name) - start_record_time = datetime.datetime.now() - recording_time_list[record_name] = [start_record_time, record_quality_zh] - - download_success = direct_download_stream( - flv_url, save_file_path, record_name, record_url, platform - ) - - if download_success: - record_finished = True - print(f"\n{anchor_name} {time.strftime('%Y-%m-%d %H:%M:%S')} 直播直播完成\n") - - recording.discard(record_name) - else: - logger.debug("未找到FLV直播流，跳过直播") - except Exception as e: - clear_record_info(record_name, record_url) - color_obj.print_colored( - f"\n{anchor_name} {time.strftime('%Y-%m-%d %H:%M:%S')} 直播直播出错,请检查网络\n", - color_obj.RED) - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - - elif record_save_type == "MP4": - try: - print('youtube------------------') - - # === 读取 YouTube 推流配置（你已在文件头加载了 _config）=== - key_raw = _yt_get("youtube", "key", fallback="", cast=str) - key = (key_raw or "").strip() - if not key: - logger.error("错误: youtube.ini[youtube].key 未配置或为空") - return - - secure = _yt_get("youtube", "secure", fallback=True, cast=bool) # 建议默认 True=rtmps - ingest = (_yt_get("youtube", "ingest", fallback="a", cast=str) or "a").strip().lower() - res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() - - # 边界保护 - bv = _safe_int(_yt_get("youtube", "bitrate", fallback=5000, cast=int), 300, 20000, 5000) - ba = _safe_int(_yt_get("youtube", "audio_bitrate", fallback=128, cast=int), 64, 384, 128) - fps = _safe_int(_yt_get("youtube", "fps", fallback=30, cast=int), 10, 60, 30) - gop = _safe_int(_yt_get("youtube", "gop", fallback=2*fps, cast=int), fps, 4*fps, 2*fps) - allow_copy = _yt_get("youtube", "copy_when_safe", fallback=True, cast=bool) - - # 出口 URL（rtmps 优先，失败时 check_subprocess 返回后再探测其他节点/协议） - out_url = _build_youtube_rtmp_url(key, secure, ingest) - - # === 输出端参数 === - v_args = [ - "-c:v", "libx264", - "-preset", "veryfast", - "-profile:v", "high", - "-level:v", "4.2", - "-pix_fmt", "yuv420p", - "-r", str(fps), - "-g", str(gop), - "-b:v", f"{bv}k", - "-maxrate", f"{bv}k", - "-bufsize", f"{max(2 * bv, 4000)}k", - ] - a_args = [ - "-c:a", "aac", - "-b:a", f"{ba}k", - "-ar", "44100", - "-ac", "2", - ] - vf_chain = [] - if res_str: - try: - w, h = map(int, res_str.lower().split("x")) - vf_chain += [ - f"scale={w}:{h}:force_original_aspect_ratio=decrease", - f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2", - "format=yuv420p", - ] - except Exception: - logger.warning(f"警告: resolution '{res_str}' 非法，已忽略") - - # 如果允许copy且源确实是 h264+aac，且未强制改分辨率，则走copy省CPU - use_copy = False - if allow_copy and not vf_chain and _can_copy_h264_aac(port_info): - use_copy = True - v_args = ["-c:v", "copy"] - a_args = ["-c:a", "copy"] - - # 输出端公共部分（注意 flvflags 与 rtmp_live） - base_cmd = ["-map", "0"] - if vf_chain: - base_cmd += ["-vf", ",".join(vf_chain)] - base_cmd += v_args + a_args + [ - "-flvflags", "no_duration_filesize", - "-rtmp_live", "live", - "-f", "flv", out_url - ] - - # === Failover 主循环：当前源挂了，立刻找下一路在播，继续推到同一 out_url === - current_input_url = real_url - current_source_url = record_url # 记原始页面URL，用于排除自己 - - while True: - # 组装一次完整命令：输入端(你的 ffmpeg_command) + 输出端(base_cmd) - run_cmd = list(ffmpeg_command) # 不污染原始 - # 替换输入URL（确保 -re -i 的后面跟的是当前 real_url） - try: - # 找到 "-re", "-i", 这三元组中的 并替换 - idx = run_cmd.index("-i") - run_cmd[idx + 1] = current_input_url - except ValueError: - # 兜底：直接塞在末尾前 - run_cmd += ["-re", "-i", current_input_url] - - run_cmd += base_cmd - - logger.info(f"{platform} | {anchor_name} | 推流启动，源={current_input_url[:80]}...") - comment_end = check_subprocess( - record_name, - current_source_url, - run_cmd, - "YTLIVE", - custom_script - ) - - # 如果是被用户注释/请求停止，check_subprocess 会 True，直接退出 - if comment_end: - return - - # FFmpeg 退出：源可能下播/断线。快速找下一路在播 - logger.info("当前源已结束或异常，尝试切换到下一路直播源...") - next_real, next_port = find_next_live_source(current_source_url, proxy_address, record_quality) - if not next_real: - logger.warning("未找到可用的下一路直播源，结束 YouTube 推流。") - break - - # 替换为下一路 - current_input_url = next_real - port_info = next_port - # 保持同一个 record_name/anchor_name；如需也替换显示名，可以在这里更新 - continue # 立即重启下一路 - - # 主循环退出：没有可切的源 - return - - except subprocess.CalledProcessError as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - - else: - if split_video_by_time: - now = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) - filename = anchor_name + f'_{title_in_name}' + now + ".ts" - print(f'{rec_info}/{filename}') - - try: - save_file_path = f"{full_path}/{anchor_name}_{title_in_name}{now}_%03d.ts" - command = [ - "-c:v", "copy", - "-c:a", "copy", - "-map", "0", - "-f", "segment", - "-segment_time", split_time, - "-segment_format", 'mpegts', - "-reset_timestamps", "1", - save_file_path, - ] - - ffmpeg_command.extend(command) - comment_end = check_subprocess( - record_name, - record_url, - ffmpeg_command, - record_save_type, - custom_script - ) - if comment_end: - if converts_to_mp4: - file_paths = utils.get_file_paths(os.path.dirname(save_file_path)) - prefix = os.path.basename(save_file_path).rsplit('_', maxsplit=1)[0] - for path in file_paths: - if prefix in path: - try: - threading.Thread( - target=converts_mp4, - args=(path, delete_origin_file) - ).start() - except subprocess.CalledProcessError as e: - logger.error(f"转码失败: {e} ") - return - - except subprocess.CalledProcessError as e: - logger.error( - f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - - else: - filename = anchor_name + f'_{title_in_name}' + now + ".ts" - print(f'{rec_info}/{filename}') - save_file_path = full_path + '/' + filename - - try: - command = [ - "-c:v", "copy", - "-c:a", "copy", - "-map", "0", - "-f", "mpegts", - save_file_path, - ] - - ffmpeg_command.extend(command) - comment_end = check_subprocess( - record_name, - record_url, - ffmpeg_command, - record_save_type, - custom_script - ) - if comment_end: - threading.Thread( - target=converts_mp4, args=(save_file_path, delete_origin_file) - ).start() - return - - except subprocess.CalledProcessError as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - - count_time = time.time() - - except Exception as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - - num = random.randint(-5, 5) + delay_default - if num < 0: - num = 0 - x = num - - if error_count > 20: - x = x + 60 - color_obj.print_colored("\r瞬时错误太多,延迟加60秒", color_obj.YELLOW) - - # 这里是.如果直播结束后,循环时间会暂时变成30s后检测一遍. 这样一定程度上防止主播卡顿造成少录 - # 当30秒过后检测一遍后. 会回归正常设置的循环秒数 - if record_finished: - count_time_end = time.time() - count_time - if count_time_end < 60: - x = 30 - record_finished = False - - else: - x = num - - # 这里是正常循环 - while x: - x = x - 1 - if loop_time: - print(f'\r{anchor_name}循环等待{x}秒 ', end="") - time.sleep(1) - - if loop_time: - print('\r检测直播间中...', end="") - except Exception as e: - logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}") - with max_request_lock: - error_count += 1 - error_window.append(1) - time.sleep(2) - - -def backup_file(file_path: str, backup_dir_path: str, limit_counts: int = 6) -> None: - try: - if not os.path.exists(backup_dir_path): - os.makedirs(backup_dir_path) - - timestamp = datetime.datetime.now().strftime('%Y-%m-%d_%H-%M-%S') - backup_file_name = os.path.basename(file_path) + '_' + timestamp - backup_file_path = os.path.join(backup_dir_path, backup_file_name).replace("\\", "/") - shutil.copy2(file_path, backup_file_path) - - files = os.listdir(backup_dir_path) - _files = [f for f in files if f.startswith(os.path.basename(file_path))] - _files.sort(key=lambda x: os.path.getmtime(os.path.join(backup_dir_path, x))) - - while len(_files) > limit_counts: - oldest_file = _files[0] - os.remove(os.path.join(backup_dir_path, oldest_file)) - _files = _files[1:] - - except Exception as e: - logger.error(f'\r备份配置文件 {file_path} 失败：{str(e)}') - - -def backup_file_start() -> None: - config_md5 = '' - url_config_md5 = '' - - while True: - try: - if os.path.exists(config_file): - new_config_md5 = utils.check_md5(config_file) - if new_config_md5 != config_md5: - backup_file(config_file, backup_dir) - config_md5 = new_config_md5 - - if os.path.exists(url_config_file): - new_url_config_md5 = utils.check_md5(url_config_file) - if new_url_config_md5 != url_config_md5: - backup_file(url_config_file, backup_dir) - url_config_md5 = new_url_config_md5 - time.sleep(600) - except Exception as e: - logger.error(f"备份配置文件失败, 错误信息: {e}") - - -def check_ffmpeg_existence() -> bool: - try: - result = subprocess.run(['ffmpeg', '-version'], check=True, capture_output=True, text=True) - if result.returncode == 0: - lines = result.stdout.splitlines() - version_line = lines[0] - built_line = lines[1] - print(version_line) - print(built_line) - except subprocess.CalledProcessError as e: - logger.error(e) - except FileNotFoundError: - pass - finally: - if check_ffmpeg(): - time.sleep(1) - return True - return False - - -# --------------------------初始化程序------------------------------------- -if not check_ffmpeg_existence(): - logger.error("缺少ffmpeg无法进行直播，程序退出") - sys.exit(1) -os.makedirs(os.path.dirname(config_file), exist_ok=True) -t3 = threading.Thread(target=backup_file_start, args=(), daemon=True) -t3.start() -utils.remove_duplicate_lines(url_config_file) - - -def read_config_value(config_parser: configparser.RawConfigParser, section: str, option: str, default_value: Any) \ - -> Any: - try: - - config_parser.read(config_file, encoding=text_encoding) - if '直播设置' not in config_parser.sections(): - config_parser.add_section('直播设置') - if '推送配置' not in config_parser.sections(): - config_parser.add_section('推送配置') - if 'Cookie' not in config_parser.sections(): - config_parser.add_section('Cookie') - if 'Authorization' not in config_parser.sections(): - config_parser.add_section('Authorization') - if '账号密码' not in config_parser.sections(): - config_parser.add_section('账号密码') - return config_parser.get(section, option) - except (configparser.NoSectionError, configparser.NoOptionError): - config_parser.set(section, option, str(default_value)) - with open(config_file, 'w', encoding=text_encoding) as f: - config_parser.write(f) - return default_value - - -options = {"是": True, "否": False} -config = configparser.RawConfigParser() -language = read_config_value(config, '直播设置', 'language(zh_cn/en)', "zh_cn") -skip_proxy_check = options.get(read_config_value(config, '直播设置', '是否跳过代理检测(是/否)', "否"), False) -if language and 'en' not in language.lower(): - from i18n import translated_print - - builtins.print = translated_print - -try: - if skip_proxy_check: - global_proxy = True - else: - print('系统代理检测中，请耐心等待...') - response_g = urllib.request.urlopen("https://www.google.com/", timeout=15) - global_proxy = True - print('\r全局/规则网络代理已开启√') - pd = ProxyDetector() - if pd.is_proxy_enabled(): - proxy_info = pd.get_proxy_info() - print("System Proxy: http://{}:{}".format(proxy_info.ip, proxy_info.port)) -except HTTPError as err: - print(f"HTTP error occurred: {err.code} - {err.reason}") -except URLError: - color_obj.print_colored("INFO：未检测到全局/规则网络代理，请检查代理配置（若无需直播海外直播请忽略此条提示）", - color_obj.YELLOW) -except Exception as err: - print("An unexpected error occurred:", err) - -while True: + return record_headers.get(platform_name) + +def detect_platform(url: str) -> str: + if "douyin.com" in url: return "抖音直播" + if "tiktok.com" in url: return "TikTok直播" + if url.endswith(".flv") or url.endswith(".m3u8"): return "自定义录制直播" + return "未知平台" + +def main(): + print("-----------------------------------------------------") + print("| DouyinLiveRecorder - 单进程转推版 |") + print("-----------------------------------------------------") + print(f"版本号: {version}\nGitHub: https://github.com/ihmily/DouyinLiveRecorder") + print("模式：只检测 + 单进程推流（停播自动切下一个 URL）") + print(".....................................................") + + if not check_ffmpeg_existence(): + logger.error("缺少 ffmpeg，程序退出") + sys.exit(1) + + urls = parse_urls_from_file(url_config_file) + if not urls: + logger.error("URL_config.ini 中没有可用 URL。") + sys.exit(1) + + # 顺序轮询索引 + idx = 0 + current_url = None + current_proc: Optional[subprocess.Popen] = None + current_platform = "" + current_anchor = "" + record_quality_code = "OD" # 固定“原画”；需要也可改为从 config.ini 读取 + + check_interval_live = 8 # 推流中检测当前源是否仍在直播的间隔秒 + scan_interval_when_idle = 2 # 没有任何可用源时的扫描间隔秒 + last_md5 = "" # URL 文件变化时热更新 try: - if not os.path.isfile(config_file): - with open(config_file, 'w', encoding=text_encoding) as file: - pass + while True: + # 动态热更新 URL 列表（URL_config.ini 有变更则重载） + try: + md5 = check_md5(url_config_file) + if md5 != last_md5: + urls = parse_urls_from_file(url_config_file) or urls + last_md5 = md5 + print(f"[{now_str()}] 已重载 URL 列表，数量：{len(urls)}") + # 若当前 URL 已不在列表，强制切换 + if current_url and current_url not in urls: + print(f"[{now_str()}] 当前URL已被移除，停止并重新选择。") + stop_ffmpeg(current_proc) + current_proc = None + current_url = None + except Exception as e: + logger.warning(f"URL 列表热更新失败：{e}") - ini_URL_content = '' - if os.path.isfile(url_config_file): - with open(url_config_file, 'r', encoding=text_encoding) as file: - ini_URL_content = file.read().strip() - - if not ini_URL_content.strip(): - input_url = input('请输入要直播的主播直播间网址（尽量使用PC网页端的直播间地址）:\n') - with open(url_config_file, 'w', encoding=text_encoding) as file: - file.write(input_url) - except OSError as err: - logger.error(f"发生 I/O 错误: {err}") - - video_save_path = read_config_value(config, '直播设置', '直播保存路径(不填则默认)', "") - folder_by_author = options.get(read_config_value(config, '直播设置', '保存文件夹是否以作者区分', "是"), False) - folder_by_time = options.get(read_config_value(config, '直播设置', '保存文件夹是否以时间区分', "否"), False) - folder_by_title = options.get(read_config_value(config, '直播设置', '保存文件夹是否以标题区分', "否"), False) - filename_by_title = options.get(read_config_value(config, '直播设置', '保存文件名是否包含标题', "否"), False) - clean_emoji = options.get(read_config_value(config, '直播设置', '是否去除名称中的表情符号', "是"), True) - video_save_type = read_config_value(config, '直播设置', '视频保存格式ts|mkv|flv|mp4|mp3音频|m4a音频', "mp4") - video_record_quality = read_config_value(config, '直播设置', '原画|超清|高清|标清|流畅', "原画") - use_proxy = options.get(read_config_value(config, '直播设置', '是否使用代理ip(是/否)', "是"), False) - proxy_addr_bak = read_config_value(config, '直播设置', '代理地址', "") - proxy_addr = None if not use_proxy else proxy_addr_bak - max_request = int(read_config_value(config, '直播设置', '同一时间访问网络的线程数', 3)) - semaphore = threading.Semaphore(max_request) - delay_default = int(read_config_value(config, '直播设置', '循环时间(秒)', 30)) - local_delay_default = int(read_config_value(config, '直播设置', '排队读取网址时间(秒)', 0)) - loop_time = options.get(read_config_value(config, '直播设置', '是否显示循环秒数', "否"), False) - show_url = options.get(read_config_value(config, '直播设置', '是否显示直播源地址', "否"), False) - split_video_by_time = options.get(read_config_value(config, '直播设置', '分段直播是否开启', "否"), False) - enable_https_recording = options.get(read_config_value(config, '直播设置', '是否强制启用https直播', "否"), False) - disk_space_limit = float(read_config_value(config, '直播设置', '直播空间剩余阈值(gb)', 1.0)) - split_time = str(read_config_value(config, '直播设置', '视频分段时间(秒)', 1800)) - converts_to_mp4 = options.get(read_config_value(config, '直播设置', '直播完成后自动转为mp4格式', "否"), False) - converts_to_h264 = options.get(read_config_value(config, '直播设置', 'mp4格式重新编码为h264', "否"), False) - delete_origin_file = options.get(read_config_value(config, '直播设置', '追加格式后删除原文件', "否"), False) - create_time_file = options.get(read_config_value(config, '直播设置', '生成时间字幕文件', "否"), False) - is_run_script = options.get(read_config_value(config, '直播设置', '是否直播完成后执行自定义脚本', "否"), False) - custom_script = read_config_value(config, '直播设置', '自定义脚本执行命令', "") if is_run_script else None - enable_proxy_platform = read_config_value( - config, '直播设置', '使用代理直播的平台(逗号分隔)', - 'tiktok, soop, pandalive, winktv, flextv, popkontv, twitch, liveme, showroom, chzzk, shopee, shp, youtu, faceit' - ) - enable_proxy_platform_list = enable_proxy_platform.replace('，', ',').split(',') if enable_proxy_platform else None - extra_enable_proxy = read_config_value(config, '直播设置', '额外使用代理直播的平台(逗号分隔)', '') - extra_enable_proxy_platform_list = extra_enable_proxy.replace('，', ',').split(',') if extra_enable_proxy else None - live_status_push = read_config_value(config, '推送配置', '直播状态推送渠道', "") - dingtalk_api_url = read_config_value(config, '推送配置', '钉钉推送接口链接', "") - xizhi_api_url = read_config_value(config, '推送配置', '微信推送接口链接', "") - bark_msg_api = read_config_value(config, '推送配置', 'bark推送接口链接', "") - bark_msg_level = read_config_value(config, '推送配置', 'bark推送中断级别', "active") - bark_msg_ring = read_config_value(config, '推送配置', 'bark推送铃声', "bell") - dingtalk_phone_num = read_config_value(config, '推送配置', '钉钉通知@对象(填手机号)', "") - dingtalk_is_atall = options.get(read_config_value(config, '推送配置', '钉钉通知@全体(是/否)', "否"), False) - tg_token = read_config_value(config, '推送配置', 'tgapi令牌', "") - tg_chat_id = read_config_value(config, '推送配置', 'tg聊天id(个人或者群组id)', "") - email_host = read_config_value(config, '推送配置', 'SMTP邮件服务器', "") - open_smtp_ssl = options.get(read_config_value(config, '推送配置', '是否使用SMTP服务SSL加密(是/否)', "是"), True) - smtp_port = read_config_value(config, '推送配置', 'SMTP邮件服务器端口', "") - login_email = read_config_value(config, '推送配置', '邮箱登录账号', "") - email_password = read_config_value(config, '推送配置', '发件人密码(授权码)', "") - sender_email = read_config_value(config, '推送配置', '发件人邮箱', "") - sender_name = read_config_value(config, '推送配置', '发件人显示昵称', "") - to_email = read_config_value(config, '推送配置', '收件人邮箱', "") - ntfy_api = read_config_value(config, '推送配置', 'ntfy推送地址', "") - ntfy_tags = read_config_value(config, '推送配置', 'ntfy推送标签', "tada") - ntfy_email = read_config_value(config, '推送配置', 'ntfy推送邮箱', "") - pushplus_token = read_config_value(config, '推送配置', 'pushplus推送token', "") - push_message_title = read_config_value(config, '推送配置', '自定义推送标题', "直播间状态更新通知") - begin_push_message_text = read_config_value(config, '推送配置', '自定义开播推送内容', "") - over_push_message_text = read_config_value(config, '推送配置', '自定义关播推送内容', "") - disable_record = options.get(read_config_value(config, '推送配置', '只推送通知不直播(是/否)', "否"), False) - push_check_seconds = int(read_config_value(config, '推送配置', '直播推送检测频率(秒)', 1800)) - begin_show_push = options.get(read_config_value(config, '推送配置', '开播推送开启(是/否)', "是"), True) - over_show_push = options.get(read_config_value(config, '推送配置', '关播推送开启(是/否)', "否"), False) - sooplive_username = read_config_value(config, '账号密码', 'sooplive账号', '') - sooplive_password = read_config_value(config, '账号密码', 'sooplive密码', '') - flextv_username = read_config_value(config, '账号密码', 'flextv账号', '') - flextv_password = read_config_value(config, '账号密码', 'flextv密码', '') - popkontv_username = read_config_value(config, '账号密码', 'popkontv账号', '') - popkontv_partner_code = read_config_value(config, '账号密码', 'partner_code', 'P-00001') - popkontv_password = read_config_value(config, '账号密码', 'popkontv密码', '') - twitcasting_account_type = read_config_value(config, '账号密码', 'twitcasting账号类型', 'normal') - twitcasting_username = read_config_value(config, '账号密码', 'twitcasting账号', '') - twitcasting_password = read_config_value(config, '账号密码', 'twitcasting密码', '') - popkontv_access_token = read_config_value(config, 'Authorization', 'popkontv_token', '') - dy_cookie = read_config_value(config, 'Cookie', '抖音cookie', '') - ks_cookie = read_config_value(config, 'Cookie', '快手cookie', '') - tiktok_cookie = read_config_value(config, 'Cookie', 'tiktok_cookie', '') - hy_cookie = read_config_value(config, 'Cookie', '虎牙cookie', '') - douyu_cookie = read_config_value(config, 'Cookie', '斗鱼cookie', '') - yy_cookie = read_config_value(config, 'Cookie', 'yy_cookie', '') - bili_cookie = read_config_value(config, 'Cookie', 'B站cookie', '') - xhs_cookie = read_config_value(config, 'Cookie', '小红书cookie', '') - bigo_cookie = read_config_value(config, 'Cookie', 'bigo_cookie', '') - blued_cookie = read_config_value(config, 'Cookie', 'blued_cookie', '') - sooplive_cookie = read_config_value(config, 'Cookie', 'sooplive_cookie', '') - netease_cookie = read_config_value(config, 'Cookie', 'netease_cookie', '') - qiandurebo_cookie = read_config_value(config, 'Cookie', '千度热播_cookie', '') - pandatv_cookie = read_config_value(config, 'Cookie', 'pandatv_cookie', '') - maoerfm_cookie = read_config_value(config, 'Cookie', '猫耳fm_cookie', '') - winktv_cookie = read_config_value(config, 'Cookie', 'winktv_cookie', '') - flextv_cookie = read_config_value(config, 'Cookie', 'flextv_cookie', '') - look_cookie = read_config_value(config, 'Cookie', 'look_cookie', '') - twitcasting_cookie = read_config_value(config, 'Cookie', 'twitcasting_cookie', '') - baidu_cookie = read_config_value(config, 'Cookie', 'baidu_cookie', '') - weibo_cookie = read_config_value(config, 'Cookie', 'weibo_cookie', '') - kugou_cookie = read_config_value(config, 'Cookie', 'kugou_cookie', '') - twitch_cookie = read_config_value(config, 'Cookie', 'twitch_cookie', '') - liveme_cookie = read_config_value(config, 'Cookie', 'liveme_cookie', '') - huajiao_cookie = read_config_value(config, 'Cookie', 'huajiao_cookie', '') - liuxing_cookie = read_config_value(config, 'Cookie', 'liuxing_cookie', '') - showroom_cookie = read_config_value(config, 'Cookie', 'showroom_cookie', '') - acfun_cookie = read_config_value(config, 'Cookie', 'acfun_cookie', '') - changliao_cookie = read_config_value(config, 'Cookie', 'changliao_cookie', '') - yinbo_cookie = read_config_value(config, 'Cookie', 'yinbo_cookie', '') - yingke_cookie = read_config_value(config, 'Cookie', 'yingke_cookie', '') - zhihu_cookie = read_config_value(config, 'Cookie', 'zhihu_cookie', '') - chzzk_cookie = read_config_value(config, 'Cookie', 'chzzk_cookie', '') - haixiu_cookie = read_config_value(config, 'Cookie', 'haixiu_cookie', '') - vvxqiu_cookie = read_config_value(config, 'Cookie', 'vvxqiu_cookie', '') - yiqilive_cookie = read_config_value(config, 'Cookie', '17live_cookie', '') - langlive_cookie = read_config_value(config, 'Cookie', 'langlive_cookie', '') - pplive_cookie = read_config_value(config, 'Cookie', 'pplive_cookie', '') - six_room_cookie = read_config_value(config, 'Cookie', '6room_cookie', '') - lehaitv_cookie = read_config_value(config, 'Cookie', 'lehaitv_cookie', '') - huamao_cookie = read_config_value(config, 'Cookie', 'huamao_cookie', '') - shopee_cookie = read_config_value(config, 'Cookie', 'shopee_cookie', '') - youtube_cookie = read_config_value(config, 'Cookie', 'youtube_cookie', '') - taobao_cookie = read_config_value(config, 'Cookie', 'taobao_cookie', '') - jd_cookie = read_config_value(config, 'Cookie', 'jd_cookie', '') - faceit_cookie = read_config_value(config, 'Cookie', 'faceit_cookie', '') - migu_cookie = read_config_value(config, 'Cookie', 'migu_cookie', '') - lianjie_cookie = read_config_value(config, 'Cookie', 'lianjie_cookie', '') - laixiu_cookie = read_config_value(config, 'Cookie', 'laixiu_cookie', '') - picarto_cookie = read_config_value(config, 'Cookie', 'picarto_cookie', '') - - video_save_type_list = ("FLV", "MKV", "TS", "MP4", "MP3音频", "M4A音频", "MP3", "M4A") - if video_save_type and video_save_type.upper() in video_save_type_list: - video_save_type = video_save_type.upper() - else: - # video_save_type = "TS" - video_save_type = "MP4" - - - - check_path = video_save_path or default_path - if utils.check_disk_capacity(check_path, show=first_run) < disk_space_limit: - exit_recording = True - if not recording: - logger.warning(f"Disk space remaining is below {disk_space_limit} GB. " - f"Exiting program due to the disk space limit being reached.") - sys.exit(-1) - - - def contains_url(string: str) -> bool: - pattern = r"(https?://)?(www\.)?[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+(:\d+)?(/.*)?" - return re.search(pattern, string) is not None - - - try: - url_comments, line_list, url_line_list = [[] for _ in range(3)] - with (open(url_config_file, "r", encoding=text_encoding, errors='ignore') as file): - for origin_line in file: - if origin_line in line_list: - delete_line(url_config_file, origin_line) - line_list.append(origin_line) - line = origin_line.strip() - if len(line) < 18: + if current_proc and current_url: + # 已在推流：检测是否仍直播 & 进程状态 + if current_proc.poll() is not None: + color.print_colored(f"[{now_str()}] ffmpeg 退出，尝试切源…", color.YELLOW) + current_proc = None + current_url = None continue - line_spilt = line.split('主播: ') - if len(line_spilt) > 2: - line = update_file(url_config_file, line, f'{line_spilt[0]}主播: {line_spilt[-1]}') + # 轻量检测：仍直播？ + try: + port = asyncio.run(fetch_port_info(current_url, record_quality_code)) + still_live = bool(port.get("is_live")) + except Exception as e: + logger.error(f"检测当前源失败：{e}") + still_live = False - is_comment_line = line.startswith("#") - if is_comment_line: - line = line.lstrip('#') + if still_live: + elapsed = int(time.time() - start_time) + h, m = divmod(elapsed, 3600) + m, s = divmod(m, 60) + dur_str = f"{h:02d}:{m:02d}:{s:02d}" + + print( + f"\r[{now_str()}] 正在转播({dur_str})：{current_platform} | {current_anchor} | {current_url}", + end="" + ) + time.sleep(check_interval_live) + continue - if re.search('[,，]', line): - split_line = re.split('[,，]', line) else: - split_line = [line, ''] + color.print_colored(f"\n[{now_str()}] 当前源已停播，切换下一个…", color.YELLOW) + stop_ffmpeg(current_proc) + current_proc = None + current_url = None + # 落下去选择下一个 + else: + # 未在推流：从 urls 顺序找第一个 is_live 的 + if not urls: + time.sleep(scan_interval_when_idle) + continue - if len(split_line) == 1: - url = split_line[0] - quality, name = [video_record_quality, ''] - elif len(split_line) == 2: - if contains_url(split_line[0]): - quality = video_record_quality - url, name = split_line - else: - quality, url = split_line - name = '' - else: - quality, url, name = split_line + tried = 0 + found = False + start_idx = idx + while tried < len(urls): + url = urls[idx] + idx = (idx + 1) % len(urls) + tried += 1 - if quality not in ("原画", "蓝光", "超清", "高清", "标清", "流畅"): - quality = '原画' - - if url not in url_line_list: - url_line_list.append(url) - else: - delete_line(url_config_file, origin_line) - - url = 'https://' + url if '://' not in url else url - url_host = url.split('/')[2] - - platform_host = [ - 'live.douyin.com', - 'v.douyin.com', - 'www.douyin.com', - 'live.kuaishou.com', - 'www.huya.com', - 'www.douyu.com', - ] - overseas_platform_host = [ - 'www.tiktok.com', - ] - - platform_host.extend(overseas_platform_host) - clean_url_host_list = ( - "live.douyin.com", - ) - - if 'live.shopee.' in url_host or '.shp.ee' in url_host: - url_host = 'live.shopee.' if 'live.shopee.' in url_host else '.shp.ee' - - if url_host in platform_host or any(ext in url for ext in (".flv", ".m3u8")): - if url_host in clean_url_host_list: - url = update_file(url_config_file, old_str=url, new_str=url.split('?')[0]) - - if 'xiaohongshu' in url: - host_id = re.search('&host_id=(.*?)(?=&|$)', url) - if host_id: - new_url = url.split('?')[0] + f'?host_id={host_id.group(1)}' - url = update_file(url_config_file, old_str=url, new_str=new_url) - - url_comments = [i for i in url_comments if url not in i] - if is_comment_line: - url_comments.append(url) - else: - new_line = (quality, url, name) - url_tuples_list.append(new_line) - else: - if not origin_line.startswith('#'): - color_obj.print_colored(f"\r{origin_line.strip()} 本行包含未知链接.此条跳过", color_obj.YELLOW) - update_file(url_config_file, old_str=origin_line, new_str=origin_line, start_str='#') - - while len(need_update_line_list): - a = need_update_line_list.pop() - replace_words = a.split('|') - if replace_words[0] != replace_words[1]: - if replace_words[1].startswith("#"): - start_with = '#' - new_word = replace_words[1][1:] - else: - start_with = None - new_word = replace_words[1] - update_file(url_config_file, old_str=replace_words[0], new_str=new_word, start_str=start_with) - - # 保序去重（文件顺序 = 备源优先级） - text_no_repeat_url = list(dict.fromkeys(url_tuples_list)) - - # 监测数量显示为“URL 条数”（哪怕不启动直播线程） - monitoring = len(text_no_repeat_url) - - if yt_single_mode and video_save_type.upper() == "MP4": - # ===== 单进程协调模式：只启 1 个协调器，不再为每个 URL 启动 start_record ===== - with yt_coordinator_lock: - if not yt_thread_started: - print("\n启用【单进程 YouTube 转播协调模式】（只保留 1 个 ffmpeg 推流进程）") - record_quality = get_quality_code(video_record_quality) - args = (record_quality, proxy_addr if use_proxy else None) - t_yt = threading.Thread(target=yt_single_coordinator, args=args, daemon=True) - t_yt.start() - yt_thread_started = True - else: - if len(text_no_repeat_url) > 0: - for url_tuple in text_no_repeat_url: - if url_tuple[1] in not_record_list: + port = asyncio.run(fetch_port_info(url, record_quality_code)) + if not port.get("anchor_name"): + print(f"[{now_str()}] 获取失败，跳过：{url}") + continue + if not port.get("is_live"): + print(f"[{now_str()}] 未开播：{port.get('anchor_name','')} | {url}") continue - if url_tuple[1] not in running_list: - print(f"\r{'新增' if not first_start else '传入'}地址: {url_tuple[1]}") - idx = len(running_list) + 1 - args = [url_tuple, idx] - create_var[f'thread_{idx}'] = threading.Thread(target=start_record, args=args) - create_var[f'thread_{idx}'].daemon = True - create_var[f'thread_{idx}'].start() - running_list.append(url_tuple[1]) - time.sleep(local_delay_default) - url_tuples_list = [] - first_start = False + # 选取真实播放地址 + real_url = select_source_url(url, port) + if not real_url: + print(f"[{now_str()}] 未得到播放地址，跳过：{url}") + continue + current_platform = detect_platform(url) + current_anchor = port.get("anchor_name", "") + headers = headers_for_platform(current_platform, url) - except Exception as err: - logger.error(f"错误信息: {err} 发生错误的行数: {err.__traceback__.tb_lineno}") + # 启动 ffmpeg + out_url, _, _ = build_youtube_output() + current_proc = launch_ffmpeg(real_url, out_url, headers) + current_url = url + start_time = time.time() + color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN) + found = True + break - def status_panel(): - """实时终端面板，覆盖刷新（不刷屏，5s更新一次）""" - while True: - try: - os.system(clear_command) - total_urls = len(text_no_repeat_url) - total_recording = len(recording) + if not found: + # 没有任何源在播，过会儿再试 + print(f"[{now_str()}] 当前无可用直播源，等待中…") + time.sleep(scan_interval_when_idle) + continue + except KeyboardInterrupt: + print("\n收到中断，正在退出…") + except Exception as e: + logger.error(f"主循环异常：{e}") + finally: + stop_ffmpeg(current_proc) + print("已退出。") - now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - - print(Fore.CYAN + "【系统状态】" + Style.RESET_ALL) - print(_hr()) - print( - f"监测中: {Fore.YELLOW}{total_urls}{Style.RESET_ALL} | " - f"并发线程: {Fore.YELLOW}{threading.active_count()}{Style.RESET_ALL} | " - f"正在直播: {Fore.GREEN}{total_recording}{Style.RESET_ALL}" - ) - print(f"当前时间: {Fore.CYAN}{now}{Style.RESET_ALL}") - print(_hr()) - - if recording_time_list: - for name, values in recording_time_list.items(): - if len(values) == 2: - start_time, quality = values - live_url = "N/A" - else: - start_time, quality, live_url = values - dur = datetime.datetime.now() - start_time - elapsed = str(dur).split(".")[0] - print( - f"{Fore.GREEN}{name}{Style.RESET_ALL} " - f"[{Fore.BLUE}{live_url}{Style.RESET_ALL}] | " - f"{quality} | 已直播 {Fore.YELLOW}{elapsed}{Style.RESET_ALL}" - ) - else: - print(Fore.MAGENTA + "暂无正在直播" + Style.RESET_ALL) - - print(_hr()) - sys.stdout.flush() - time.sleep(5) - - except KeyboardInterrupt: - break - except Exception as e: - logger.error(f"状态面板异常: {e}") - time.sleep(2) - - - if first_run: - t2 = threading.Thread(target=adjust_max_request, args=(), daemon=True) - t2.start() - t_status = threading.Thread(target=status_panel, daemon=True) - t_status.start() - first_run = False +if __name__ == "__main__": + main() diff --git a/main.py b/main.py index f770081..530b0be 100644 --- a/main.py +++ b/main.py @@ -207,11 +207,11 @@ async def fetch_port_info(url: str, record_quality_code: str = "OD") -> dict: # --------------------- ffmpeg 单进程推流 --------------------- def build_youtube_output() -> Tuple[str, list, list]: - """读取 youtube.ini，返回 (out_url, v_args, a_args)""" key = _yt_get("youtube", "key", fallback="", cast=str) if not key: logger.error("错误: config/youtube.ini 未配置 key") sys.exit(1) + secure = _yt_get("youtube", "secure", fallback=False, cast=bool) ingest = _yt_get("youtube", "ingest", fallback="a", cast=str) res_str = (_yt_get("youtube", "resolution", fallback="", cast=str) or "").strip() @@ -223,6 +223,7 @@ def build_youtube_output() -> Tuple[str, list, list]: out_url = _build_youtube_rtmp_url(key, secure, ingest) + # —— 默认强制转码（最稳），如果想省 CPU 可把 -c:v libx264 改为 copy —— # v_args = [ "-c:v", "libx264", "-preset", "veryfast", @@ -242,18 +243,16 @@ def build_youtube_output() -> Tuple[str, list, list]: "-ac", "2", ] - # 可选缩放/留黑边 - vf_chain = [] + # 可选缩放/留黑边，保持 YT 吃得更稳 if res_str: try: w, h = map(int, res_str.lower().split("x")) - vf_chain.append(f"scale={w}:{h}:force_original_aspect_ratio=decrease") - vf_chain.append(f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2") - vf_chain.append("format=yuv420p") + vf = f"scale={w}:{h}:force_original_aspect_ratio=decrease,pad={w}:{h}:(ow-iw)/2:(oh-ih)/2,format=yuv420p" + v_args = ["-vf", vf] + v_args except Exception: logger.warning(f"resolution '{res_str}' 非法，忽略") - return out_url, v_args, a_args if not vf_chain else (out_url, ["-vf", ",".join(vf_chain)] + v_args, a_args) + return out_url, v_args, a_args def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = None) -> subprocess.Popen: """ @@ -269,40 +268,50 @@ def launch_ffmpeg(input_url: str, out_url: str, extra_headers: Optional[str] = N base = [ "ffmpeg", "-nostdin", "-hide_banner", + # 超时 & 重连（必须在 -i 前面） "-rw_timeout", rw_timeout, - "-loglevel", "error", + "-timeout", "10000000", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_at_eof", "1", + "-reconnect_delay_max", "60", + + # UA + 协议 "-user_agent", user_agent, "-protocol_whitelist", "rtmp,crypto,file,http,https,tcp,tls,udp,rtp,httpproxy", + + # 缓冲 & 分析 "-thread_queue_size", "1024", "-analyzeduration", analyzeduration, "-probesize", probesize, "-fflags", "+discardcorrupt", + + # 输入流（放在重连参数之后） "-re", "-i", input_url, + + # 视频/音频编解码 + "-c:v", "copy", # 直接拷贝视频，省 CPU，低延迟 + "-c:a", "aac", # 音频转码成 AAC（保证 YouTube 兼容） + "-ar", "44100", # 音频采样率 + "-b:a", "128k", # 音频码率 + "-ac", "2", # 双声道 "-bufsize", bufsize, - "-sn", "-dn", - "-reconnect_delay_max", "60", - "-reconnect_streamed", "-reconnect_at_eof", "-max_muxing_queue_size", "2048", - "-correct_ts_overflow", "1", "-avoid_negative_ts", "1", - "-rtmp_live", "live", - "-timeout", "10000000" + "-fflags", "+genpts", # 确保时间戳连续，防止花屏 + + # 输出容器 + "-f", "flv", out_url ] if extra_headers: - base[ base.index("-user_agent") : base.index("-thread_queue_size") ] += ["-headers", extra_headers] + base[ base.index("-user_agent") : base.index("-protocol_whitelist") ] += ["-headers", extra_headers] - out_url2, v_args, a_args = build_youtube_output() - if out_url: - out_url2 = out_url # 允许外部传入覆盖 - - cmd = base + v_args + a_args + ["-f", "flv", out_url2] if proxy_addr: - cmd = ["ffmpeg", "-http_proxy", proxy_addr] + cmd[1:] + base = ["ffmpeg", "-http_proxy", proxy_addr] + base[1:] logger.info(f"启动推流：{input_url} -> YouTube") - # Windows 优雅关闭：后续喂 'q'；Linux 发 SIGINT - p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + p = subprocess.Popen(base, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) return p def stop_ffmpeg(p: Optional[subprocess.Popen]): @@ -423,9 +432,18 @@ def main(): still_live = False if still_live: - print(f"\r[{now_str()}] 正在转播：{current_platform} | {current_anchor} | {current_url}", end="") + elapsed = int(time.time() - start_time) + h, m = divmod(elapsed, 3600) + m, s = divmod(m, 60) + dur_str = f"{h:02d}:{m:02d}:{s:02d}" + + print( + f"\r[{now_str()}] 正在转播({dur_str})：{current_platform} | {current_anchor} | {current_url}", + end="" + ) time.sleep(check_interval_live) continue + else: color.print_colored(f"\n[{now_str()}] 当前源已停播，切换下一个…", color.YELLOW) stop_ffmpeg(current_proc) @@ -468,6 +486,7 @@ def main(): out_url, _, _ = build_youtube_output() current_proc = launch_ffmpeg(real_url, out_url, headers) current_url = url + start_time = time.time() color.print_colored(f"[{now_str()}] 开始转播：{current_platform} | {current_anchor} | {url}", color.GREEN) found = True break