diff --git a/backup/back13 b/backup/back13 new file mode 100644 index 0000000..df687b6 --- /dev/null +++ b/backup/back13 @@ -0,0 +1,443 @@ +import os +import subprocess +import threading +import asyncio +import uuid +import time +import signal +import sys +import logging +import aiohttp +import requests +import spider +import stream +import configparser +from typing import Any, Optional +from rich.console import Console +from rich.progress import Progress +from rich.table import Table +import pyttsx3 +from watchdog.observers import Observer +from watchdog.events import FileSystemEventHandler +import typer +import json + +# ----------------- 日志和 Rich 设置 ----------------- +console = Console() +logger = logging.getLogger() +logger.setLevel(logging.INFO) +handler = logging.StreamHandler() +handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S")) +logger.addHandler(handler) +logging.getLogger('comtypes').setLevel(logging.ERROR) + +# ----------------- 配置和状态 ----------------- +class Config: + youtube_key = None + live_urls = [] + config_changed = False + streamer_instance = None + +CONFIG = Config() + +# ----------------- 文件监控 ----------------- +class ConfigFileHandler(FileSystemEventHandler): + def on_modified(self, event): + if event.src_path.endswith(("config.ini", "list.ini")): + console.log(f"[yellow]🛠️ 检测到 {event.src_path} 变更，正在重新加载配置...[/yellow]") + CONFIG.config_changed = True + +def start_file_watcher(): + observer = Observer() + observer.schedule(ConfigFileHandler(), path=".", recursive=False) + observer.start() + console.log("[cyan]📂 已启动文件监控，实时检测 config.ini 和 list.ini 变更[/cyan]") + return observer + +# ----------------- 抖音到 YouTube 推流类 ----------------- +class DouyinToYouTubeStreamer: + def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', fps=30, gop=60, max_minutes=None, enable_recording=False): + self.push_proc = None + self.record_proc = None + self.stop_flag = threading.Event() + self.bitrate = bitrate + self.audio_bitrate = audio_bitrate + self.resolution = resolution + self.fps = fps + self.gop = gop + self.start_time = None + self.room_name = None + self.pushing = False + self.out_file = None + self.max_minutes = max_minutes + self.enable_recording = enable_recording + try: + self.tts_engine = pyttsx3.init() + except Exception as e: + console.log(f"[red]❌ 初始化 pyttsx3 失败: {e}，语音播报不可用[/red]") + self.tts_engine = None + os.makedirs("records", exist_ok=True) + + def start_stream(self, real_url, youtube_key, room_name=None): + self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" + console.log(f"[green]🚀 [{self.room_name}] 启动推流到 YouTube[/green]") + if self.tts_engine: + self.tts_engine.say(f"正在为直播间 {self.room_name} 启动推流") + self.tts_engine.runAndWait() + threading.Thread(target=self._run_stream_loop, args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._monitor_thread, daemon=True).start() + + def _build_push_cmd(self, real_url, youtube_key, attempt=0): + user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36" + headers = "Referer: https://www.douyin.com/\r\nOrigin: https://www.douyin.com\r\n" + rtmp_servers = [f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}', f'rtmp://b.rtmp.youtube.com/live2/{youtube_key}'] + cmd = [ + 'ffmpeg', '-y', '-rw_timeout', '10000000', '-analyzeduration', '1000000', '-probesize', '5000000', + '-user_agent', user_agent, '-headers', headers, '-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto', + '-fflags', '+discardcorrupt+genpts', '-flags', 'low_delay', '-http_seekable', '0', + '-reconnect', '1', '-reconnect_streamed', '1', '-i', real_url, + '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-b:v', f'{self.bitrate}k', '-maxrate', f'{self.bitrate}k', '-bufsize', f'{self.bitrate * 2}k', + '-r', str(self.fps), '-g', str(self.gop), '-s', self.resolution, + '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-f', 'flv', + rtmp_servers[min(attempt, len(rtmp_servers)-1)] + ] + return cmd + + def _build_record_cmd(self, real_url): + base_name = f"{self.room_name}_{int(time.time())}.mp4" + self.out_file = os.path.join("records", base_name).replace("\\", "/") + cmd = [ + 'ffmpeg', '-y', '-i', real_url, '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-movflags', '+faststart+frag_keyframe', + self.out_file + ] + return cmd + + def _run_stream_loop(self, real_url, youtube_key): + attempt = 0 + max_attempts = 2 + while not self.stop_flag.is_set() and attempt < max_attempts: + try: + self.start_time = time.time() + self.pushing = False + push_cmd = self._build_push_cmd(real_url, youtube_key, attempt) + self.push_proc = subprocess.Popen(push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=1, universal_newlines=True) + console.log(f"[cyan]🔄 [{self.room_name}] 推流进程 PID: {self.push_proc.pid}[/cyan]") + if self.enable_recording: + record_cmd = self._build_record_cmd(real_url) + self.record_proc = subprocess.Popen(record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=1, universal_newlines=True) + console.log(f"[cyan]📹 [{self.room_name}] 本地录制: {self.out_file}[/cyan]") + + def monitor_proc(proc, name): + try: + while not self.stop_flag.is_set() and proc and proc.poll() is None: + line = proc.stderr.readline() + if not line: + continue + if name == "push" and not self.pushing and "frame=" in line and "fps=" in line: + self.pushing = True + console.log(f"[green]✅ [{self.room_name}] 推流成功[/green]") + if self.tts_engine: + self.tts_engine.say(f"直播间 {self.room_name} 推流成功") + self.tts_engine.runAndWait() + except Exception as e: + console.log(f"[red]❌ [{self.room_name}] {name} stderr 异常: {e}[/red]") + self._send_ntfy_notification(title="推流错误", message=f"{self.room_name} {name} 异常: {str(e)}") + + threading.Thread(target=monitor_proc, args=(self.push_proc, "push"), daemon=True).start() + if self.enable_recording: + threading.Thread(target=monitor_proc, args=(self.record_proc, "record"), daemon=True).start() + + while not self.stop_flag.is_set(): + elapsed = time.time() - self.start_time + if self.max_minutes and elapsed >= self.max_minutes * 60: + console.log(f"[yellow]⏰ [{self.room_name}] 超过 {self.max_minutes} 分钟，停止推流[/yellow]") + self.stop() + break + if self.push_proc.poll() is not None: + console.log(f"[yellow]⚠️ [{self.room_name}] 推流进程退出，正在重试...[/yellow]") + self._send_ntfy_notification(title="推流中断", message=f"{self.room_name} 推流进程退出") + attempt += 1 + break + if self.enable_recording and self.record_proc and self.record_proc.poll() is not None: + console.log(f"[yellow]⚠️ [{self.room_name}] 录制进程退出，1秒后重启...[/yellow]") + self._terminate_procs() + break + time.sleep(1) + except Exception as e: + console.log(f"[red]❌ [{self.room_name}] FFmpeg 异常: {e}[/red]") + self._send_ntfy_notification(title="FFmpeg 错误", message=f"{self.room_name} FFmpeg 异常: {str(e)}") + attempt += 1 + finally: + self._terminate_procs() + if not self.stop_flag.is_set() and attempt < max_attempts: + time.sleep(2) + elif attempt >= max_attempts: + console.log(f"[red]❌ [{self.room_name}] 达到最大重试次数，停止推流[/red]") + self.stop_flag.set() + + def _terminate_procs(self): + for proc in [self.push_proc, self.record_proc]: + if proc and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + proc.kill() + self.push_proc, self.record_proc = None, None + + def _send_ntfy_notification(self, title: str, message: str): + ntfy_url = "https://ntfy.sh/your_topic" + try: + requests.post(ntfy_url, data=message.encode('utf-8'), headers={"Title": title.encode('utf-8')}) + console.log(f"[cyan]📢 发送 ntfy 通知: {title}[/cyan]") + except Exception as e: + console.log(f"[red]❌ 发送 ntfy 通知失败: {e}[/red]") + + def _monitor_thread(self): + with Progress(console=console) as progress: + task = progress.add_task(f"[cyan]📡 [{self.room_name}] 推流状态[/cyan]", total=None) + while not self.stop_flag.is_set(): + uptime = int(time.time() - self.start_time) if self.start_time else 0 + m, s = divmod(uptime, 60) + status = "推流中" if self.pushing else "启动中" + progress.update(task, description=f"[cyan]📡 [{self.room_name}] {status} | 时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k[/cyan]") + time.sleep(1) + + def stop(self): + console.log(f"[yellow]🛑 [{self.room_name}] 停止推流和录制[/yellow]") + if self.tts_engine: + self.tts_engine.say(f"直播间 {self.room_name} 停止推流") + self.tts_engine.runAndWait() + self.stop_flag.set() + self._terminate_procs() + +# ----------------- 抖音流解析 ----------------- +async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=None, cookies='', max_retries=2): + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36", + "Referer": "https://www.douyin.com/", + "Origin": "https://www.douyin.com", + "Accept": "application/json, text/javascript, */*; q=0.01", + } + session_args = {"headers": headers} + if proxy_addr: + session_args["proxy"] = proxy_addr + if cookies: + session_args["cookies"] = cookies + + try: + async with aiohttp.ClientSession(**session_args) as session: + for attempt in range(max_retries): + try: + async with session.get(record_url, timeout=5) as resp: + console.log(f"[cyan]🔗 HTTP 请求: {record_url} -> 状态码 {resp.status}[/cyan]") + text = await resp.text() + if not text.strip() or "captcha" in text.lower(): + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 验证码或空响应[/yellow]") + await asyncio.sleep(2) + continue + if 'v.douyin.com' not in record_url and '/user/' not in record_url: + try: + json_data = await spider.get_douyin_stream_data(url=record_url, proxy_addr=proxy_addr, cookies=cookies) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] spider.get_douyin_stream_data 异常: {e}[/yellow]") + await asyncio.sleep(2) + continue + else: + try: + json_data = await spider.get_douyin_app_stream_data(url=record_url, proxy_addr=proxy_addr, cookies=cookies) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] spider.get_douyin_app_stream_data 异常: {e}[/yellow]") + await asyncio.sleep(2) + continue + if json_data is None: + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 解析返回 None[/yellow]") + await asyncio.sleep(2) + continue + try: + port_info = await stream.get_douyin_stream_url(json_data, record_quality, proxy_addr) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] stream.get_douyin_stream_url 异常: {e}[/yellow]") + await asyncio.sleep(2) + continue + if port_info and isinstance(port_info, dict) and port_info.get("record_url"): + console.log(f"[green]✅ [{record_url}] 成功解析直播流[/green]") + return port_info.get("record_url") + else: + anchor_name = port_info.get('anchor_name', '未知') + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: {anchor_name} 未开播[/yellow]") + await asyncio.sleep(2) + continue + except json.JSONDecodeError as e: + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: JSON 解析失败 - {e}[/yellow]") + await asyncio.sleep(2) + except aiohttp.ClientError as e: + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 网络请求失败 - {e}[/yellow]") + await asyncio.sleep(2) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 抓取异常 - {e}[/yellow]") + await asyncio.sleep(2) + console.log(f"[red]❌ [{record_url}] 多次尝试后无法获取直播流[/red]") + DouyinToYouTubeStreamer()._send_ntfy_notification(title="直播流解析失败", message=f"多次尝试后无法获取直播流: {record_url}") + return None + except Exception as e: + console.log(f"[red]❌ [{record_url}] 初始化 aiohttp 会话失败: {e}[/red]") + DouyinToYouTubeStreamer()._send_ntfy_notification(title="会话初始化失败", message=f"初始化 aiohttp 会话失败: {e}") + return None + +def start_douyin_to_youtube(record_url, youtube_key, real_url: Optional[str] = None, room_name=None, proxy_addr=None, cookies='', bitrate=5000, enable_recording=False): + if real_url is None: + real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) + if not real_url: + console.log(f"[red]❌ [{record_url}] 无法获取真实直播流[/red]") + return None + console.log(f"[green]✅ [{record_url}] 真实直播流: {real_url}[/green]") + streamer = DouyinToYouTubeStreamer(bitrate=bitrate, enable_recording=enable_recording, max_minutes=None) + streamer.start_stream(real_url, youtube_key, room_name) + return streamer + +# ----------------- 自动调度 ----------------- +def load_config(): + config = configparser.ConfigParser() + try: + config.read("config.ini", encoding="utf-8") + youtube_key = config.get("youtube", "key", fallback=None) + if youtube_key: + console.log("[green]🔑 已加载 YouTube 密钥[/green]") + else: + console.log("[red]❌ config.ini 未找到 YouTube 密钥[/red]") + return youtube_key + except Exception as e: + console.log(f"[red]❌ 加载 config.ini 失败: {e}[/red]") + return None + +def load_list(): + urls = [] + seen = set() + try: + with open("list.ini", "r", encoding="utf-8") as f: + for line in f: + url = line.strip() + if url and not url.startswith("#") and url not in seen: + urls.append(url) + seen.add(url) + console.log(f"[cyan]📋 已加载 {len(urls)} 个直播间 URL[/cyan]") + return urls + except FileNotFoundError: + console.log("[red]❌ 未找到 list.ini 文件[/red]") + except Exception as e: + console.log(f"[red]❌ 加载 list.ini 失败: {e}[/red]") + return [] + +async def find_live_and_stream(youtube_key): + urls = load_list() + if not urls: + console.log("[red]❌ 无直播间 URL，退出检测[/red]") + return None + + table = Table(title="直播间检测状态", show_lines=True) + table.add_column("序号", style="cyan", width=6) + table.add_column("URL", style="magenta", width=40) + table.add_column("主播", style="yellow", width=20) + table.add_column("状态", style="green", width=10) + status_dict = {} + + with Progress(console=console, transient=True) as progress: + task = progress.add_task("[cyan]🔍 检测直播间...[/cyan]", total=len(urls)) + for index, url in enumerate(urls, 1): + console.log(f"[cyan]🔍 检查 [{index}/{len(urls)}] {url}[/cyan]") + real_url = await get_douyin_real_url(url) + status = "直播中" if real_url else "未开播" + anchor_name = "未知" + if not real_url: + try: + json_data = await spider.get_douyin_stream_data(url) + anchor_name = json_data.get('anchor_name', '未知') if json_data else "未知" + except Exception as e: + console.log(f"[yellow]⚠️ [{url}] 获取主播信息失败: {e}[/yellow]") + status_dict[url] = (index, url, anchor_name, status) + + # 更新表格 + table.rows = [] # 清空并重新添加以保持顺序 + for _, u, a, s in sorted(status_dict.values(), key=lambda x: x[0]): + table.add_row(str(_), u, a, s) + console.print(table) + progress.update(task, advance=1) + + if real_url: + console.log(f"[green]✅ [{url}] 发现直播流[/green]") + streamer = start_douyin_to_youtube(url, youtube_key, real_url=real_url, room_name=f"抖音_{url.split('/')[-1]}") + if streamer: + console.log(f"[green]🚀 [{url}] 推流已启动，停止后续检测[/green]") + return streamer + else: + console.log(f"[yellow]⚠️ [{url}] 启动推流失败，继续检测下一个[/yellow]") + else: + console.log(f"[yellow]⚠️ [{url}] 未开播，继续检测下一个[/yellow]") + + console.log("[yellow]⚠️ 所有直播间检测完成，未发现直播[/yellow]") + return None + +# ----------------- 主程序 ----------------- +if __name__ == "__main__": + app = typer.Typer() + + @app.command() + def start(): + youtube_key = load_config() + if not youtube_key: + console.log("[red]❌ 未找到 YouTube 密钥，退出[/red]") + sys.exit(1) + CONFIG.youtube_key = youtube_key + CONFIG.live_urls = load_list() + if not CONFIG.live_urls: + console.log("[red]❌ 无直播间 URL，退出[/red]") + sys.exit(1) + observer = start_file_watcher() + console.log("[green]✅ 程序启动，自动检测抖音直播[/green]") + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + while True: + try: + if CONFIG.config_changed: + console.log("[yellow]🛠️ 重新加载配置[/yellow]") + CONFIG.youtube_key = load_config() + CONFIG.live_urls = load_list() + CONFIG.config_changed = False + if not CONFIG.streamer_instance or CONFIG.streamer_instance.stop_flag.is_set(): + console.log("[cyan]🔄 推流中断或未启动，开始新一轮遍历检测...[/cyan]") + CONFIG.streamer_instance = loop.run_until_complete(find_live_and_stream(CONFIG.youtube_key)) + if not CONFIG.streamer_instance: + console.log("[yellow]⚠️ 未发现直播间，10秒后重试[/yellow]") + time.sleep(10) + time.sleep(5) + except Exception as e: + console.log(f"[red]❌ 主循环异常: {e}，5秒后重试[/red]") + time.sleep(5) + + @app.command() + def stop(): + if CONFIG.streamer_instance: + CONFIG.streamer_instance.stop() + console.log("[yellow]🛑 推流已停止[/yellow]") + else: + console.log("[yellow]⚠️ 无活跃推流[/yellow]") + + @app.command() + def status(): + if CONFIG.streamer_instance and not CONFIG.streamer_instance.stop_flag.is_set(): + console.log(f"[green]✅ 推流活跃: {CONFIG.streamer_instance.room_name}[/green]") + else: + console.log("[yellow]⚠️ 无活跃推流[/yellow]") + + def signal_handler(sig, frame): + if CONFIG.streamer_instance: + CONFIG.streamer_instance.stop() + console.log("[yellow]🛑 程序终止[/yellow]") + sys.exit(0) + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + app() \ No newline at end of file diff --git a/backup/bnack12 b/backup/bnack12 new file mode 100644 index 0000000..9cfb1fc --- /dev/null +++ b/backup/bnack12 @@ -0,0 +1,323 @@ +import os +import subprocess +import threading +import asyncio +import uuid +import time +import signal +import sys +import logging +import platform +import aiohttp +import spider # 你的 spider 模块 +import stream # 你的 stream 模块 +import configparser +from typing import Any, Optional + +# ----------------- Logging Setup ----------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S" +) + +# ----------------- Douyin to YouTube Streamer ----------------- +class DouyinToYouTubeStreamer: + def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', + fps=30, gop=60, max_minutes=None, enable_recording=False): + self.push_proc = None + self.record_proc = None + self.stop_flag = threading.Event() + self.bitrate = bitrate + self.audio_bitrate = audio_bitrate + self.resolution = resolution + self.fps = fps + self.gop = gop + self.start_time = None + self.room_name = None + self.pushing = False + self.out_file = None + self.max_minutes = max_minutes + self.enable_recording = enable_recording + + os.makedirs("records", exist_ok=True) + + def start_stream(self, real_url, youtube_key, room_name=None): + self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" + logging.info(f"[{self.room_name}] 启动低延迟推流，目标 YouTube Key: {youtube_key}") + threading.Thread(target=self._run_stream_loop, + args=(real_url, youtube_key), daemon=True).start() + threading.Thread(target=self._monitor_thread, daemon=True).start() + + def _build_push_cmd(self, real_url, youtube_key): + user_agent = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36") + headers = "Referer: https://www.douyin.com/\r\nOrigin: https://www.douyin.com\r\n" + cmd = [ + 'ffmpeg', '-y', + '-rw_timeout', '10000000', + '-analyzeduration', '1000000', + '-probesize', '5000000', + '-user_agent', user_agent, + '-headers', headers, + '-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto', + '-fflags', '+discardcorrupt+genpts', + '-flags', 'low_delay', + '-http_seekable', '0', + '-i', real_url, + '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-b:v', f'{self.bitrate}k', + '-maxrate', f'{self.bitrate}k', + '-bufsize', f'{self.bitrate * 2}k', + '-r', str(self.fps), + '-g', str(self.gop), + '-s', self.resolution, + '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', + '-f', 'flv', + f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}' + ] + return cmd + + def _build_record_cmd(self, real_url): + base_name = f"{self.room_name}_{int(time.time())}.mp4" + self.out_file = os.path.join("records", base_name).replace("\\", "/") + cmd = [ + 'ffmpeg', '-y', + '-i', real_url, + '-c:v', 'libx264', + '-preset', 'veryfast', + '-tune', 'zerolatency', + '-c:a', 'aac', + '-b:a', f'{self.audio_bitrate}k', + '-ar', '44100', + '-ac', '2', + '-movflags', '+faststart+frag_keyframe', + self.out_file + ] + return cmd + + def _run_stream_loop(self, real_url, youtube_key): + while not self.stop_flag.is_set(): + try: + self.start_time = time.time() + self.pushing = False + + push_cmd = self._build_push_cmd(real_url, youtube_key) + self.push_proc = subprocess.Popen( + push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + bufsize=1, universal_newlines=True + ) + logging.info(f"[{self.room_name}] 推流进程 PID: {self.push_proc.pid}") + + if self.enable_recording: + record_cmd = self._build_record_cmd(real_url) + self.record_proc = subprocess.Popen( + record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + bufsize=1, universal_newlines=True + ) + logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") + + def monitor_proc(proc, name): + try: + while not self.stop_flag.is_set() and proc and proc.poll() is None: + line = proc.stderr.readline() + if not line: + continue + if name == "push" and not self.pushing and "frame=" in line and "fps=" in line: + self.pushing = True + logging.info(f"[{self.room_name}] 推流到 YouTube 成功") + except Exception as e: + logging.error(f"[{self.room_name} {name} stderr异常: {e}") + + threading.Thread(target=monitor_proc, args=(self.push_proc, "push"), daemon=True).start() + if self.enable_recording: + threading.Thread(target=monitor_proc, args=(self.record_proc, "record"), daemon=True).start() + + while not self.stop_flag.is_set(): + elapsed = time.time() - self.start_time + if self.max_minutes and elapsed >= self.max_minutes * 60: + logging.warning(f"[{self.room_name}] 超过 {self.max_minutes} 分钟，自动停止推流和录制") + self.stop() + break + + if self.push_proc.poll() is not None: + logging.warning(f"[{self.room_name}] 推流进程退出，准备重新遍历 list.ini ...") + self.stop_flag.set() + break + + if self.enable_recording and self.record_proc and self.record_proc.poll() is not None: + logging.warning(f"[{self.room_name}] 录制进程退出，1秒后重启...") + self._terminate_procs() + break + + time.sleep(1) + + except Exception as e: + logging.error(f"[{self.room_name}] FFmpeg异常: {e}") + finally: + self._terminate_procs() + if not self.stop_flag.is_set(): + time.sleep(1) + + def _terminate_procs(self): + for proc in [self.push_proc, self.record_proc]: + if proc and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + proc.kill() + self.push_proc, self.record_proc = None, None + + def _monitor_thread(self): + while not self.stop_flag.is_set(): + uptime = int(time.time() - self.start_time) if self.start_time else 0 + m, s = divmod(uptime, 60) + status = "推流中" if self.pushing else "启动中" + print(f"\r[{self.room_name}] {status} | 运行时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k", end="") + time.sleep(1) + + def stop(self): + logging.info(f"[{self.room_name}] 停止推流和录制...") + self.stop_flag.set() + self._terminate_procs() + + +# ----------------- Douyin Stream Parsing ----------------- +async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=None, cookies=''): + headers = { + "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"), + "Referer": "https://www.douyin.com/", + "Origin": "https://www.douyin.com", + "Accept": "application/json, text/javascript, */*; q=0.01", + } + + session_args = {"headers": headers} + if proxy_addr: + session_args["proxy"] = proxy_addr + + try: + async with aiohttp.ClientSession(**session_args) as session: + for attempt in range(5): # 增加重试次数到5次 + try: + async with session.get(record_url, cookies=cookies, timeout=10) as resp: + text = await resp.text() + if not text.strip() or "captcha" in text.lower(): + logging.warning(f"第{attempt+1}次尝试获取流失败，返回内容为空或验证码，重试...") + await asyncio.sleep(2) # 增加重试间隔 + continue + + if 'v.douyin.com' not in record_url and '/user/' not in record_url: + json_data = await spider.get_douyin_stream_data( + url=record_url, proxy_addr=proxy_addr, cookies=cookies) + else: + json_data = await spider.get_douyin_app_stream_data( + url=record_url, proxy_addr=proxy_addr, cookies=cookies) + + if json_data is None: + logging.warning(f"第{attempt+1}次spider解析返回None，重试...") + await asyncio.sleep(2) + continue + + port_info = await stream.get_douyin_stream_url( + json_data, record_quality, proxy_addr) + + if port_info and isinstance(port_info, dict) and port_info.get("record_url"): + logging.info(f"成功解析直播流: {record_url}") + return port_info.get("record_url") + else: + logging.warning(f"第{attempt+1}次尝试解析失败，port_info无效: {port_info}") + await asyncio.sleep(2) + except json.JSONDecodeError as e: + logging.warning(f"第{attempt+1}次JSON解析失败: {e}，重试...") + await asyncio.sleep(2) + except aiohttp.ClientError as e: + logging.warning(f"第{attempt+1}次网络请求失败: {e}，重试...") + await asyncio.sleep(2) + except Exception as e: + logging.warning(f"第{attempt+1}次尝试抓取抖音流异常: {e}") + await asyncio.sleep(2) + logging.error(f"多次尝试后仍获取抖音真实流失败: {record_url}") + return None + except Exception as e: + logging.error(f"初始化 aiohttp session 失败: {e}") + return None + + +def start_douyin_to_youtube(record_url, youtube_key, real_url: Optional[str] = None, room_name=None, + proxy_addr=None, cookies='', bitrate=5000, enable_recording=False): + if real_url is None: + real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) + if not real_url: + logging.error(f"获取真实直播流失败: {record_url}") + return None + logging.info(f"实际直播流: {real_url}") + streamer = DouyinToYouTubeStreamer(bitrate=bitrate, enable_recording=enable_recording, max_minutes=None) + streamer.start_stream(real_url, youtube_key, room_name) + return streamer + + +# ----------------- Auto Scheduler ----------------- +def load_config(): + config = configparser.ConfigParser() + config.read("config.ini", encoding="utf-8") + youtube_key = config.get("youtube", "key", fallback=None) + return youtube_key + + +def load_list(): + urls = [] + seen = set() + with open("list.ini", "r", encoding="utf-8") as f: + for line in f: + url = line.strip() + if url and not url.startswith("#") and url not in seen: + urls.append(url) + seen.add(url) + logging.info(f"加载 {len(urls)} 个唯一直播间URL，按文件顺序排列") + return urls + + +def find_live_and_stream(youtube_key): + urls = load_list() + for i, url in enumerate(urls, 1): + logging.info(f"检查第 {i}/{len(urls)} 个URL: {url}") + real_url = asyncio.run(get_douyin_real_url(url)) + if real_url: + logging.info(f"找到可用直播流: {url}") + streamer = start_douyin_to_youtube(url, youtube_key, real_url=real_url, room_name=f"抖音_{url.split('/')[-1]}", enable_recording=False) + if streamer: + return streamer + else: + logging.warning(f"{url} 流检测成功但启动失败，继续下一个...") + else: + logging.info(f"{url} 暂无直播，继续检测...") + return None + + +# ----------------- Main Program ----------------- +if __name__ == "__main__": + youtube_key = load_config() + if not youtube_key: + logging.error("配置文件 config.ini 中未找到 YouTube Key") + sys.exit(1) + + streamer_instance = None + + def signal_handler(sig, frame): + if streamer_instance: + streamer_instance.stop() + sys.exit(0) + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + logging.info("程序已启动，自动检测抖音直播...") + while True: + if not streamer_instance or streamer_instance.stop_flag.is_set(): + streamer_instance = find_live_and_stream(youtube_key) + if not streamer_instance: + logging.info("未找到正在直播的房间，60秒后重试...") + time.sleep(60) + time.sleep(10) \ No newline at end of file diff --git a/list.ini b/list.ini index dff08a4..12c6e7c 100644 --- a/list.ini +++ b/list.ini @@ -1,3 +1,4 @@ +https://live.douyin.com/944705936194 https://live.douyin.com/767599969296 https://live.douyin.com/456576000931 https://live.douyin.com/745606325880 diff --git a/live.py b/live.py index 9cfb1fc..04e258b 100644 --- a/live.py +++ b/live.py @@ -7,24 +7,57 @@ import time import signal import sys import logging -import platform import aiohttp -import spider # 你的 spider 模块 -import stream # 你的 stream 模块 +import requests +import spider +import stream import configparser -from typing import Any, Optional +from typing import Any, Optional, Tuple +from rich.console import Console +from rich.progress import Progress +from rich.table import Table +import pyttsx3 +from watchdog.observers import Observer +from watchdog.events import FileSystemEventHandler +import typer +import json -# ----------------- Logging Setup ----------------- -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(message)s", - datefmt="%H:%M:%S" -) +# ----------------- 日志和 Rich 设置 ----------------- +console = Console() +logger = logging.getLogger() +logger.setLevel(logging.INFO) +handler = logging.StreamHandler() +handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) +logger.addHandler(handler) +logging.getLogger('comtypes').setLevel(logging.ERROR) -# ----------------- Douyin to YouTube Streamer ----------------- +# ----------------- 配置和状态 ----------------- +class Config: + youtube_key: Optional[str] = None + live_urls: list = [] + config_changed: bool = False + streamer_instance: Optional['DouyinToYouTubeStreamer'] = None + +CONFIG = Config() + +# ----------------- 文件监控 ----------------- +class ConfigFileHandler(FileSystemEventHandler): + def on_modified(self, event): + if event.src_path.endswith(("config.ini", "list.ini")): + console.log(f"[yellow]🛠️ 检测到 {event.src_path} 变更，重新加载配置[/yellow]") + CONFIG.config_changed = True + +def start_file_watcher() -> Observer: + observer = Observer() + observer.schedule(ConfigFileHandler(), path=".", recursive=False) + observer.start() + console.log("[cyan]📂 文件监控已启动，实时检测 config.ini 和 list.ini 变更[/cyan]") + return observer + +# ----------------- 抖音到 YouTube 推流类 ----------------- class DouyinToYouTubeStreamer: - def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', - fps=30, gop=60, max_minutes=None, enable_recording=False): + def __init__(self, bitrate: int = 5000, audio_bitrate: int = 128, resolution: str = '720x1280', + fps: int = 30, gop: int = 60, max_minutes: Optional[int] = None, enable_recording: bool = False): self.push_proc = None self.record_proc = None self.stop_flag = threading.Event() @@ -39,83 +72,61 @@ class DouyinToYouTubeStreamer: self.out_file = None self.max_minutes = max_minutes self.enable_recording = enable_recording - + try: + self.tts_engine = pyttsx3.init() + except Exception as e: + console.log(f"[red]❌ pyttsx3 初始化失败: {e}，语音播报不可用[/red]") + self.tts_engine = None os.makedirs("records", exist_ok=True) - def start_stream(self, real_url, youtube_key, room_name=None): + def start_stream(self, real_url: str, youtube_key: str, room_name: Optional[str] = None) -> None: self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" - logging.info(f"[{self.room_name}] 启动低延迟推流，目标 YouTube Key: {youtube_key}") - threading.Thread(target=self._run_stream_loop, - args=(real_url, youtube_key), daemon=True).start() + console.log(f"[green]🚀 [{self.room_name}] 启动推流到 YouTube[/green]") + if self.tts_engine: + self.tts_engine.say(f"正在为直播间 {self.room_name} 启动推流") + self.tts_engine.runAndWait() + threading.Thread(target=self._run_stream_loop, args=(real_url, youtube_key), daemon=True).start() threading.Thread(target=self._monitor_thread, daemon=True).start() - def _build_push_cmd(self, real_url, youtube_key): - user_agent = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36") + def _build_push_cmd(self, real_url: str, youtube_key: str, attempt: int = 0) -> list: + user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36" headers = "Referer: https://www.douyin.com/\r\nOrigin: https://www.douyin.com\r\n" - cmd = [ - 'ffmpeg', '-y', - '-rw_timeout', '10000000', - '-analyzeduration', '1000000', - '-probesize', '5000000', - '-user_agent', user_agent, - '-headers', headers, - '-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto', - '-fflags', '+discardcorrupt+genpts', - '-flags', 'low_delay', - '-http_seekable', '0', - '-i', real_url, + rtmp_servers = [f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}', f'rtmp://b.rtmp.youtube.com/live2/{youtube_key}'] + return [ + 'ffmpeg', '-y', '-rw_timeout', '10000000', '-analyzeduration', '1000000', '-probesize', '5000000', + '-user_agent', user_agent, '-headers', headers, '-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto', + '-fflags', '+discardcorrupt+genpts', '-flags', 'low_delay', '-http_seekable', '0', + '-reconnect', '1', '-reconnect_streamed', '1', '-i', real_url, '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', - '-b:v', f'{self.bitrate}k', - '-maxrate', f'{self.bitrate}k', - '-bufsize', f'{self.bitrate * 2}k', - '-r', str(self.fps), - '-g', str(self.gop), - '-s', self.resolution, - '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', - '-f', 'flv', - f'rtmp://a.rtmp.youtube.com/live2/{youtube_key}' + '-b:v', f'{self.bitrate}k', '-maxrate', f'{self.bitrate}k', '-bufsize', f'{self.bitrate * 2}k', + '-r', str(self.fps), '-g', str(self.gop), '-s', self.resolution, + '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-f', 'flv', + rtmp_servers[min(attempt, len(rtmp_servers)-1)] ] - return cmd - def _build_record_cmd(self, real_url): + def _build_record_cmd(self, real_url: str) -> list: base_name = f"{self.room_name}_{int(time.time())}.mp4" self.out_file = os.path.join("records", base_name).replace("\\", "/") - cmd = [ - 'ffmpeg', '-y', - '-i', real_url, - '-c:v', 'libx264', - '-preset', 'veryfast', - '-tune', 'zerolatency', - '-c:a', 'aac', - '-b:a', f'{self.audio_bitrate}k', - '-ar', '44100', - '-ac', '2', - '-movflags', '+faststart+frag_keyframe', + return [ + 'ffmpeg', '-y', '-i', real_url, '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-movflags', '+faststart+frag_keyframe', self.out_file ] - return cmd - def _run_stream_loop(self, real_url, youtube_key): - while not self.stop_flag.is_set(): + def _run_stream_loop(self, real_url: str, youtube_key: str) -> None: + attempt = 0 + max_attempts = 2 + while not self.stop_flag.is_set() and attempt < max_attempts: try: self.start_time = time.time() self.pushing = False - - push_cmd = self._build_push_cmd(real_url, youtube_key) - self.push_proc = subprocess.Popen( - push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, - bufsize=1, universal_newlines=True - ) - logging.info(f"[{self.room_name}] 推流进程 PID: {self.push_proc.pid}") - + push_cmd = self._build_push_cmd(real_url, youtube_key, attempt) + self.push_proc = subprocess.Popen(push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=1, universal_newlines=True) + console.log(f"[cyan]🔄 [{self.room_name}] 推流进程 PID: {self.push_proc.pid}[/cyan]") if self.enable_recording: record_cmd = self._build_record_cmd(real_url) - self.record_proc = subprocess.Popen( - record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, - bufsize=1, universal_newlines=True - ) - logging.info(f"[{self.room_name}] 本地录制开始: {self.out_file}") + self.record_proc = subprocess.Popen(record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=1, universal_newlines=True) + console.log(f"[cyan]📹 [{self.room_name}] 本地录制: {self.out_file}[/cyan]") def monitor_proc(proc, name): try: @@ -125,9 +136,13 @@ class DouyinToYouTubeStreamer: continue if name == "push" and not self.pushing and "frame=" in line and "fps=" in line: self.pushing = True - logging.info(f"[{self.room_name}] 推流到 YouTube 成功") + console.log(f"[green]✅ [{self.room_name}] 推流成功[/green]") + if self.tts_engine: + self.tts_engine.say(f"直播间 {self.room_name} 推流成功") + self.tts_engine.runAndWait() except Exception as e: - logging.error(f"[{self.room_name} {name} stderr异常: {e}") + console.log(f"[red]❌ [{self.room_name}] {name} stderr 异常: {e}[/red]") + self._send_ntfy_notification(title="推流错误", message=f"{self.room_name} {name} 异常: {str(e)}") threading.Thread(target=monitor_proc, args=(self.push_proc, "push"), daemon=True).start() if self.enable_recording: @@ -136,30 +151,32 @@ class DouyinToYouTubeStreamer: while not self.stop_flag.is_set(): elapsed = time.time() - self.start_time if self.max_minutes and elapsed >= self.max_minutes * 60: - logging.warning(f"[{self.room_name}] 超过 {self.max_minutes} 分钟，自动停止推流和录制") + console.log(f"[yellow]⏰ [{self.room_name}] 超过 {self.max_minutes} 分钟，停止推流[/yellow]") self.stop() break - if self.push_proc.poll() is not None: - logging.warning(f"[{self.room_name}] 推流进程退出，准备重新遍历 list.ini ...") - self.stop_flag.set() + console.log(f"[yellow]⚠️ [{self.room_name}] 推流进程退出，尝试重试...[/yellow]") + self._send_ntfy_notification(title="推流中断", message=f"{self.room_name} 推流进程退出") + attempt += 1 break - if self.enable_recording and self.record_proc and self.record_proc.poll() is not None: - logging.warning(f"[{self.room_name}] 录制进程退出，1秒后重启...") + console.log(f"[yellow]⚠️ [{self.room_name}] 录制进程退出，1秒后重启...[/yellow]") self._terminate_procs() break - time.sleep(1) - except Exception as e: - logging.error(f"[{self.room_name}] FFmpeg异常: {e}") + console.log(f"[red]❌ [{self.room_name}] FFmpeg 异常: {e}[/red]") + self._send_ntfy_notification(title="FFmpeg 错误", message=f"{self.room_name} FFmpeg 异常: {str(e)}") + attempt += 1 finally: self._terminate_procs() - if not self.stop_flag.is_set(): - time.sleep(1) + if not self.stop_flag.is_set() and attempt < max_attempts: + time.sleep(2) + elif attempt >= max_attempts: + console.log(f"[red]❌ [{self.room_name}] 达到最大重试次数，停止推流[/red]") + self.stop_flag.set() - def _terminate_procs(self): + def _terminate_procs(self) -> None: for proc in [self.push_proc, self.record_proc]: if proc and proc.poll() is None: proc.terminate() @@ -169,155 +186,250 @@ class DouyinToYouTubeStreamer: proc.kill() self.push_proc, self.record_proc = None, None - def _monitor_thread(self): - while not self.stop_flag.is_set(): - uptime = int(time.time() - self.start_time) if self.start_time else 0 - m, s = divmod(uptime, 60) - status = "推流中" if self.pushing else "启动中" - print(f"\r[{self.room_name}] {status} | 运行时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k", end="") - time.sleep(1) + def _send_ntfy_notification(self, title: str, message: str) -> None: + ntfy_url = "https://ntfy.sh/your_topic" + try: + requests.post(ntfy_url, data=message.encode('utf-8'), headers={"Title": title.encode('utf-8')}) + console.log(f"[cyan]📢 ntfy 通知已发送: {title}[/cyan]") + except Exception as e: + console.log(f"[red]❌ ntfy 通知发送失败: {e}[/red]") - def stop(self): - logging.info(f"[{self.room_name}] 停止推流和录制...") + def _monitor_thread(self) -> None: + with Progress(console=console) as progress: + task = progress.add_task(f"[cyan]📡 [{self.room_name}] 推流状态[/cyan]", total=None) + while not self.stop_flag.is_set(): + uptime = int(time.time() - self.start_time) if self.start_time else 0 + m, s = divmod(uptime, 60) + status = "推流中" if self.pushing else "启动中" + progress.update(task, description=f"[cyan]📡 [{self.room_name}] {status} | 时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k[/cyan]") + time.sleep(1) + + def stop(self) -> None: + console.log(f"[yellow]🛑 [{self.room_name}] 停止推流和录制[/yellow]") + if self.tts_engine: + self.tts_engine.say(f"直播间 {self.room_name} 停止推流") + self.tts_engine.runAndWait() self.stop_flag.set() self._terminate_procs() - -# ----------------- Douyin Stream Parsing ----------------- -async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=None, cookies=''): +# ----------------- 抖音流解析 ----------------- +async def get_douyin_real_url(record_url: str, record_quality: str = "原画", proxy_addr: Optional[str] = None, + cookies: str = '', max_retries: int = 2) -> Tuple[Optional[str], str]: headers = { - "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"), + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36", "Referer": "https://www.douyin.com/", "Origin": "https://www.douyin.com", "Accept": "application/json, text/javascript, */*; q=0.01", } - session_args = {"headers": headers} if proxy_addr: session_args["proxy"] = proxy_addr + if cookies: + session_args["cookies"] = cookies + anchor_name = "未知" try: async with aiohttp.ClientSession(**session_args) as session: - for attempt in range(5): # 增加重试次数到5次 + for attempt in range(max_retries): try: - async with session.get(record_url, cookies=cookies, timeout=10) as resp: + async with session.get(record_url, timeout=5) as resp: + console.log(f"[cyan]🔗 HTTP 请求: {record_url} -> 状态码 {resp.status}[/cyan]") text = await resp.text() if not text.strip() or "captcha" in text.lower(): - logging.warning(f"第{attempt+1}次尝试获取流失败，返回内容为空或验证码，重试...") - await asyncio.sleep(2) # 增加重试间隔 + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 验证码或空响应[/yellow]") + await asyncio.sleep(2) continue - + json_data = None if 'v.douyin.com' not in record_url and '/user/' not in record_url: - json_data = await spider.get_douyin_stream_data( - url=record_url, proxy_addr=proxy_addr, cookies=cookies) + try: + json_data = await spider.get_douyin_stream_data(url=record_url, proxy_addr=proxy_addr, cookies=cookies) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] spider.get_douyin_stream_data 异常: {e}[/yellow]") + await asyncio.sleep(2) + continue else: - json_data = await spider.get_douyin_app_stream_data( - url=record_url, proxy_addr=proxy_addr, cookies=cookies) - + try: + json_data = await spider.get_douyin_app_stream_data(url=record_url, proxy_addr=proxy_addr, cookies=cookies) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] spider.get_douyin_app_stream_data 异常: {e}[/yellow]") + await asyncio.sleep(2) + continue if json_data is None: - logging.warning(f"第{attempt+1}次spider解析返回None，重试...") + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 解析返回 None[/yellow]") await asyncio.sleep(2) continue - - port_info = await stream.get_douyin_stream_url( - json_data, record_quality, proxy_addr) - - if port_info and isinstance(port_info, dict) and port_info.get("record_url"): - logging.info(f"成功解析直播流: {record_url}") - return port_info.get("record_url") - else: - logging.warning(f"第{attempt+1}次尝试解析失败，port_info无效: {port_info}") + anchor_name = json_data.get('anchor_name', '未知') + try: + port_info = await stream.get_douyin_stream_url(json_data, record_quality, proxy_addr) + except Exception as e: + console.log(f"[yellow]⚠️ [{record_url}] stream.get_douyin_stream_url 异常: {e}[/yellow]") await asyncio.sleep(2) + continue + if port_info and isinstance(port_info, dict) and port_info.get("record_url"): + console.log(f"[green]✅ [{record_url}] 成功解析直播流，主播: {anchor_name}[/green]") + return port_info.get("record_url"), anchor_name + else: + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 主播 {anchor_name} 未开播[/yellow]") + await asyncio.sleep(2) + continue except json.JSONDecodeError as e: - logging.warning(f"第{attempt+1}次JSON解析失败: {e}，重试...") + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: JSON 解析失败 - {e}[/yellow]") await asyncio.sleep(2) except aiohttp.ClientError as e: - logging.warning(f"第{attempt+1}次网络请求失败: {e}，重试...") + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 网络请求失败 - {e}[/yellow]") await asyncio.sleep(2) except Exception as e: - logging.warning(f"第{attempt+1}次尝试抓取抖音流异常: {e}") + console.log(f"[yellow]⚠️ [{record_url}] 第 {attempt+1}/{max_retries} 次: 抓取异常 - {e}[/yellow]") await asyncio.sleep(2) - logging.error(f"多次尝试后仍获取抖音真实流失败: {record_url}") - return None + console.log(f"[red]❌ [{record_url}] 多次尝试后无法获取直播流，主播: {anchor_name}[/red]") + DouyinToYouTubeStreamer()._send_ntfy_notification(title="直播流解析失败", message=f"多次尝试后无法获取直播流: {record_url}") + return None, anchor_name except Exception as e: - logging.error(f"初始化 aiohttp session 失败: {e}") - return None + console.log(f"[red]❌ [{record_url}] 初始化 aiohttp 会话失败: {e}[/red]") + DouyinToYouTubeStreamer()._send_ntfy_notification(title="会话初始化失败", message=f"初始化 aiohttp 会话失败: {e}") + return None, anchor_name - -def start_douyin_to_youtube(record_url, youtube_key, real_url: Optional[str] = None, room_name=None, - proxy_addr=None, cookies='', bitrate=5000, enable_recording=False): +def start_douyin_to_youtube(record_url: str, youtube_key: str, real_url: Optional[str] = None, + room_name: Optional[str] = None, proxy_addr: Optional[str] = None, + cookies: str = '', bitrate: int = 5000, enable_recording: bool = False) -> Optional['DouyinToYouTubeStreamer']: if real_url is None: - real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) - if not real_url: - logging.error(f"获取真实直播流失败: {record_url}") - return None - logging.info(f"实际直播流: {real_url}") + real_url, anchor_name = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) + if not real_url: + console.log(f"[red]❌ [{record_url}] 无法获取真实直播流，主播: {anchor_name}[/red]") + return None + console.log(f"[green]✅ [{record_url}] 真实直播流: {real_url}[/green]") streamer = DouyinToYouTubeStreamer(bitrate=bitrate, enable_recording=enable_recording, max_minutes=None) - streamer.start_stream(real_url, youtube_key, room_name) + streamer.start_stream(real_url, youtube_key, room_name or f"抖音_{anchor_name or record_url.split('/')[-1]}") return streamer - -# ----------------- Auto Scheduler ----------------- -def load_config(): +# ----------------- 自动调度 ----------------- +def load_config() -> Optional[str]: config = configparser.ConfigParser() - config.read("config.ini", encoding="utf-8") - youtube_key = config.get("youtube", "key", fallback=None) - return youtube_key + try: + config.read("config.ini", encoding="utf-8") + youtube_key = config.get("youtube", "key", fallback=None) + console.log(f"[green]🔑 {'已加载 YouTube 密钥' if youtube_key else '❌ config.ini 未找到 YouTube 密钥'}[/green]") + return youtube_key + except Exception as e: + console.log(f"[red]❌ 加载 config.ini 失败: {e}[/red]") + return None - -def load_list(): +def load_list() -> list: urls = [] seen = set() - with open("list.ini", "r", encoding="utf-8") as f: - for line in f: - url = line.strip() - if url and not url.startswith("#") and url not in seen: - urls.append(url) - seen.add(url) - logging.info(f"加载 {len(urls)} 个唯一直播间URL，按文件顺序排列") - return urls + try: + with open("list.ini", "r", encoding="utf-8") as f: + for line in f: + url = line.strip() + if url and not url.startswith("#") and url not in seen: + urls.append(url) + seen.add(url) + console.log(f"[cyan]📋 已加载 {len(urls)} 个直播间 URL[/cyan]") + return urls + except FileNotFoundError: + console.log("[red]❌ 未找到 list.ini 文件[/red]") + except Exception as e: + console.log(f"[red]❌ 加载 list.ini 失败: {e}[/red]") + return [] - -def find_live_and_stream(youtube_key): +async def find_live_and_stream(youtube_key: str) -> Optional['DouyinToYouTubeStreamer']: urls = load_list() - for i, url in enumerate(urls, 1): - logging.info(f"检查第 {i}/{len(urls)} 个URL: {url}") - real_url = asyncio.run(get_douyin_real_url(url)) - if real_url: - logging.info(f"找到可用直播流: {url}") - streamer = start_douyin_to_youtube(url, youtube_key, real_url=real_url, room_name=f"抖音_{url.split('/')[-1]}", enable_recording=False) - if streamer: - return streamer + if not urls: + console.log("[red]❌ 无直播间 URL，退出检测[/red]") + return None + + table = Table(title="直播间检测状态", show_lines=True) + table.add_column("序号", style="cyan", width=6, justify="center") + table.add_column("URL", style="magenta", width=40) + table.add_column("主播", style="yellow", width=20) + table.add_column("状态", style="green", width=10, justify="center") + status_dict = {} + + with Progress(console=console, transient=True) as progress: + task = progress.add_task("[cyan]🔍 检测直播间...[/cyan]", total=len(urls)) + for index, url in enumerate(urls, 1): + console.log(f"[cyan]🔍 检查 [{index}/{len(urls)}] {url}[/cyan]") + real_url, anchor_name = await get_douyin_real_url(url) + status = "直播中" if real_url else "未开播" + status_dict[url] = (index, url, anchor_name, status) + + table.rows = [] + for _, u, a, s in sorted(status_dict.values(), key=lambda x: x[0]): + table.add_row(str(_), u, a, s) + console.print(table) + progress.update(task, advance=1) + + if real_url: + console.log(f"[green]✅ [{url}] 发现直播流，主播: {anchor_name}[/green]") + streamer = start_douyin_to_youtube(url, youtube_key, real_url=real_url, room_name=f"抖音_{anchor_name or url.split('/')[-1]}") + if streamer: + console.log(f"[green]🚀 [{url}] 推流已启动，停止后续检测[/green]") + return streamer + console.log(f"[yellow]⚠️ [{url}] 启动推流失败，继续检测下一个[/yellow]") else: - logging.warning(f"{url} 流检测成功但启动失败，继续下一个...") - else: - logging.info(f"{url} 暂无直播，继续检测...") + console.log(f"[yellow]⚠️ [{url}] 主播 {anchor_name} 未开播，继续检测下一个[/yellow]") + + console.log("[yellow]⚠️ 所有直播间检测完成，未发现直播[/yellow]") return None - -# ----------------- Main Program ----------------- +# ----------------- 主程序 ----------------- if __name__ == "__main__": - youtube_key = load_config() - if not youtube_key: - logging.error("配置文件 config.ini 中未找到 YouTube Key") - sys.exit(1) + app = typer.Typer() - streamer_instance = None + @app.command() + def start(): + youtube_key = load_config() + if not youtube_key: + console.log("[red]❌ 未找到 YouTube 密钥，退出[/red]") + sys.exit(1) + CONFIG.youtube_key = youtube_key + CONFIG.live_urls = load_list() + if not CONFIG.live_urls: + console.log("[red]❌ 无直播间 URL，退出[/red]") + sys.exit(1) + observer = start_file_watcher() + console.log("[green]✅ 程序启动，开始自动检测抖音直播[/green]") + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + while True: + try: + if CONFIG.config_changed: + console.log("[yellow]🛠️ 检测到配置变更，重新加载...[/yellow]") + CONFIG.youtube_key = load_config() + CONFIG.live_urls = load_list() + CONFIG.config_changed = False + if not CONFIG.streamer_instance or CONFIG.streamer_instance.stop_flag.is_set(): + console.log("[cyan]🔄 推流中断或未启动，开始新一轮检测...[/cyan]") + CONFIG.streamer_instance = loop.run_until_complete(find_live_and_stream(CONFIG.youtube_key)) + if not CONFIG.streamer_instance: + console.log("[yellow]⚠️ 未发现直播间，10秒后重试[/yellow]") + time.sleep(10) + time.sleep(5) + except Exception as e: + console.log(f"[red]❌ 主循环异常: {e}，5秒后重试[/red]") + time.sleep(5) + + @app.command() + def stop(): + if CONFIG.streamer_instance: + CONFIG.streamer_instance.stop() + console.log("[yellow]🛑 推流已停止[/yellow]") + else: + console.log("[yellow]⚠️ 无活跃推流[/yellow]") + + @app.command() + def status(): + if CONFIG.streamer_instance and not CONFIG.streamer_instance.stop_flag.is_set(): + console.log(f"[green]✅ 推流活跃: {CONFIG.streamer_instance.room_name}[/green]") + else: + console.log("[yellow]⚠️ 无活跃推流[/yellow]") def signal_handler(sig, frame): - if streamer_instance: - streamer_instance.stop() + if CONFIG.streamer_instance: + CONFIG.streamer_instance.stop() + console.log("[yellow]🛑 程序终止[/yellow]") sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) - - logging.info("程序已启动，自动检测抖音直播...") - while True: - if not streamer_instance or streamer_instance.stop_flag.is_set(): - streamer_instance = find_live_and_stream(youtube_key) - if not streamer_instance: - logging.info("未找到正在直播的房间，60秒后重试...") - time.sleep(60) - time.sleep(10) \ No newline at end of file + app() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..99dd4fa --- /dev/null +++ b/requirements.txt @@ -0,0 +1,15 @@ +requests>=2.31.0 +loguru>=0.7.3 +pycryptodome>=3.20.0 +distro>=1.9.0 +tqdm>=4.67.1 +httpx[http2]>=0.28.1 +PyExecJS>=1.5.1 +rich +pyttsx3 +watchdog +celery +redis +python-ntfy +aiohttp +typer>=0.12.5 \ No newline at end of file