This commit is contained in:
eric
2025-09-25 11:52:19 +08:00
parent 8f496f2109
commit 6d3f50e1e3
9 changed files with 124449 additions and 346 deletions

View File

@@ -1,2 +1,2 @@
[youtube] [youtube]
key = ue78-1c3e-mr9g-14mz-9r4z key = qxvb-r47b-r5ju-6ud3-6k7z

View File

@@ -1,28 +1,27 @@
https://live.douyin.com/944705936194 https://live.douyin.com/43665279648
https://live.douyin.com/767599969296 https://live.douyin.com/359511214397
https://live.douyin.com/72212955083
https://live.douyin.com/621118100231
https://live.douyin.com/347310028364
https://live.douyin.com/280422307519
https://live.douyin.com/651293616593
https://live.douyin.com/114111936890
https://live.douyin.com/122383435921
https://live.douyin.com/92876979328
https://live.douyin.com/136996589868
https://live.douyin.com/517715534931
https://live.douyin.com/460525712926
https://live.douyin.com/456576000931 https://live.douyin.com/456576000931
https://live.douyin.com/745606325880 https://live.douyin.com/745606325880
https://live.douyin.com/511878781385 https://live.douyin.com/749687541944
https://live.douyin.com/416879949175
https://live.douyin.com/675343045974 https://live.douyin.com/675343045974
https://live.douyin.com/673565298571 https://live.douyin.com/673565298571
https://live.douyin.com/835571459859 https://live.douyin.com/835571459859
https://live.douyin.com/152358755212 https://live.douyin.com/152358755212
https://live.douyin.com/990825651731 https://live.douyin.com/990825651731
https://live.douyin.com/591442402624 https://live.douyin.com/591442402624
https://live.douyin.com/483160615952 https://live.douyin.com/967381081829
https://live.douyin.com/673565298571 https://live.douyin.com/15652099787
https://live.douyin.com/525514386431 https://live.douyin.com/78012762575
https://live.douyin.com/8687122573 https://live.douyin.com/80248655914
https://live.douyin.com/642534242822
https://live.douyin.com/589037028237
https://live.douyin.com/743565594721
https://live.douyin.com/249578288248
https://live.douyin.com/472140253414
https://live.douyin.com/700846653732
https://live.douyin.com/81849868631
https://live.douyin.com/690114366322
https://live.douyin.com/642534242822
https://live.douyin.com/469980190666
https://live.douyin.com/388066418744
https://live.douyin.com/122383435921

557
live.py
View File

@@ -8,56 +8,67 @@ import signal
import sys import sys
import logging import logging
import aiohttp import aiohttp
import requests import spider # 您的 spider 模块
import spider import stream # 您的 stream 模块
import stream
import configparser import configparser
from typing import Any, Optional, Tuple from typing import Optional, Tuple
from rich.console import Console from colorama import init, Fore, Style
from rich.progress import Progress from logging.handlers import RotatingFileHandler
from rich.table import Table
import pyttsx3
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import typer
import json
# ----------------- 日志和 Rich 设置 ----------------- # ----------------- 日志设置 -----------------
console = Console() # 初始化 colorama
logger = logging.getLogger() init(autoreset=True)
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)
# ----------------- 配置和状态 ----------------- # 创建 logs 目录
class Config: LOG_DIR = "logs"
youtube_key: Optional[str] = None os.makedirs(LOG_DIR, exist_ok=True)
live_urls: list = [] LOG_FILE = os.path.join(LOG_DIR, "app.log")
config_changed: bool = False
streamer_instance: Optional['DouyinToYouTubeStreamer'] = None
CONFIG = Config() # 自定义彩色日志 Formatter
class ColorFormatter(logging.Formatter):
COLORS = {
logging.DEBUG: Fore.BLUE,
logging.INFO: Fore.CYAN,
logging.WARNING: Fore.YELLOW,
logging.ERROR: Fore.RED,
logging.CRITICAL: Fore.RED + Style.BRIGHT,
}
# ----------------- 文件监控 ----------------- def format(self, record):
class ConfigFileHandler(FileSystemEventHandler): color = self.COLORS.get(record.levelno, "")
def on_modified(self, event): message = super().format(record)
if event.src_path.endswith(("config.ini", "list.ini")): return f"{color}{message}{Style.RESET_ALL}"
console.log(f"[yellow]🛠️ 检测到 {event.src_path} 变更,重新加载配置[/yellow]")
CONFIG.config_changed = True
def start_file_watcher() -> Observer: # 控制台日志(INFO 以上,带颜色)
observer = Observer() console_handler = logging.StreamHandler(sys.stdout)
observer.schedule(ConfigFileHandler(), path=".", recursive=False) console_handler.setLevel(logging.INFO)
observer.start() console_formatter = ColorFormatter("%(asctime)s [%(levelname)s] %(message)s", "%H:%M:%S")
console.log("[cyan]📂 文件监控已启动,实时检测 config.ini 和 list.ini 变更[/cyan]") console_handler.setFormatter(console_formatter)
return observer
# ----------------- 抖音到 YouTube 推流类 ----------------- # 文件日志(DEBUG 以上,自动滚动,5MB*5份)
file_handler = RotatingFileHandler(LOG_FILE, maxBytes=5*1024*1024, backupCount=5, encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s", "%Y-%m-%d %H:%M:%S")
file_handler.setFormatter(file_formatter)
# 根 logger
logging.basicConfig(level=logging.DEBUG, handlers=[console_handler, file_handler])
logger = logging.getLogger("main")
# 抑制不必要的三方日志
logging.getLogger("aiohttp").setLevel(logging.CRITICAL)
logging.getLogger("aiohttp.client").setLevel(logging.CRITICAL)
logging.getLogger("httpcore").setLevel(logging.CRITICAL)
logging.getLogger("urllib3").setLevel(logging.CRITICAL)
logging.getLogger("asyncio").setLevel(logging.CRITICAL)
if hasattr(spider, 'logger'):
spider.logger.setLevel(logging.CRITICAL)
if hasattr(stream, 'logger'):
stream.logger.setLevel(logging.CRITICAL)
# ----------------- 主逻辑 -----------------
class DouyinToYouTubeStreamer: class DouyinToYouTubeStreamer:
def __init__(self, bitrate: int = 5000, audio_bitrate: int = 128, resolution: str = '720x1280', def __init__(self, bitrate=5000, audio_bitrate=128, resolution='720x1280', fps=30, gop=60, enable_recording=False):
fps: int = 30, gop: int = 60, max_minutes: Optional[int] = None, enable_recording: bool = False):
self.push_proc = None self.push_proc = None
self.record_proc = None self.record_proc = None
self.stop_flag = threading.Event() self.stop_flag = threading.Event()
@@ -70,113 +81,102 @@ class DouyinToYouTubeStreamer:
self.room_name = None self.room_name = None
self.pushing = False self.pushing = False
self.out_file = None self.out_file = None
self.max_minutes = max_minutes
self.enable_recording = enable_recording 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) os.makedirs("records", exist_ok=True)
self.logger = logging.getLogger(self.__class__.__name__)
def start_stream(self, real_url: str, youtube_key: str, room_name: Optional[str] = None) -> None: def start_stream(self, real_url, youtube_key, room_name=None):
self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}"
console.log(f"[green]🚀 [{self.room_name}] 启动推流到 YouTube[/green]") self.logger.info(f"{Fore.GREEN}[{self.room_name}] 启动推流到 YouTube{Style.RESET_ALL}")
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._run_stream_loop, args=(real_url, youtube_key), daemon=True).start()
threading.Thread(target=self._monitor_thread, daemon=True).start() threading.Thread(target=self._monitor_thread, daemon=True).start()
def _build_push_cmd(self, real_url: str, youtube_key: str, attempt: int = 0) -> list: 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" 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" 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}']
return [ return [
'ffmpeg', '-y', '-rw_timeout', '10000000', '-analyzeduration', '1000000', '-probesize', '5000000', 'ffmpeg', '-y',
'-user_agent', user_agent, '-headers', headers, '-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto', '-rw_timeout', '10000000',
'-fflags', '+discardcorrupt+genpts', '-flags', 'low_delay', '-http_seekable', '0', '-user_agent', user_agent,
'-reconnect', '1', '-reconnect_streamed', '1', '-i', real_url, '-headers', headers,
'-protocol_whitelist', 'file,pipe,http,https,tcp,tls,udp,rtp,crypto',
'-fflags', '+discardcorrupt+genpts',
'-flags', 'low_delay',
'-i', real_url,
'-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency',
'-b:v', f'{self.bitrate}k', '-maxrate', f'{self.bitrate}k', '-bufsize', f'{self.bitrate * 2}k', '-b:v', f'{self.bitrate}k',
'-r', str(self.fps), '-g', str(self.gop), '-s', self.resolution, '-maxrate', f'{self.bitrate}k',
'-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-f', 'flv', '-bufsize', f'{self.bitrate * 2}k',
rtmp_servers[min(attempt, len(rtmp_servers)-1)] '-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}'
] ]
def _build_record_cmd(self, real_url: str) -> list: def _build_record_cmd(self, real_url):
base_name = f"{self.room_name}_{int(time.time())}.mp4" base_name = f"{self.room_name}_{int(time.time())}.mp4"
self.out_file = os.path.join("records", base_name).replace("\\", "/") self.out_file = os.path.join("records", base_name).replace("\\", "/")
return [ return [
'ffmpeg', '-y', '-i', real_url, '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', 'ffmpeg', '-y',
'-c:a', 'aac', '-b:a', f'{self.audio_bitrate}k', '-ar', '44100', '-ac', '2', '-movflags', '+faststart+frag_keyframe', '-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',
self.out_file self.out_file
] ]
def _run_stream_loop(self, real_url: str, youtube_key: str) -> None: def _run_stream_loop(self, real_url, youtube_key):
attempt = 0 while not self.stop_flag.is_set():
max_attempts = 2
while not self.stop_flag.is_set() and attempt < max_attempts:
try: try:
self.start_time = time.time() self.start_time = time.time()
self.pushing = False self.pushing = False
push_cmd = self._build_push_cmd(real_url, youtube_key, attempt) 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) self.push_proc = subprocess.Popen(
console.log(f"[cyan]🔄 [{self.room_name}] 推流进程 PID: {self.push_proc.pid}[/cyan]") push_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
text=True, bufsize=1
)
self.logger.info(f"{Fore.CYAN}[{self.room_name}] 推流进程 PID: {self.push_proc.pid}{Style.RESET_ALL}")
if self.enable_recording: if self.enable_recording:
record_cmd = self._build_record_cmd(real_url) 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) self.record_proc = subprocess.Popen(
console.log(f"[cyan]📹 [{self.room_name}] 本地录制: {self.out_file}[/cyan]") record_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
text=True, bufsize=1
)
self.logger.info(f"{Fore.CYAN}[{self.room_name}] 录制到: {self.out_file}{Style.RESET_ALL}")
def monitor_proc(proc, name): def monitor_proc(proc, name):
try: while not self.stop_flag.is_set() and proc and proc.poll() is None:
while not self.stop_flag.is_set() and proc and proc.poll() is None: line = proc.stderr.readline()
line = proc.stderr.readline() if line and name == "push" and not self.pushing and "frame=" in line:
if not line: self.pushing = True
continue self.logger.info(f"{Fore.GREEN}[{self.room_name}] 成功推流到 YouTube{Style.RESET_ALL}")
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() threading.Thread(target=monitor_proc, args=(self.push_proc, "push"), daemon=True).start()
if self.enable_recording: if self.enable_recording:
threading.Thread(target=monitor_proc, args=(self.record_proc, "record"), daemon=True).start() threading.Thread(target=monitor_proc, args=(self.record_proc, "record"), daemon=True).start()
while not self.stop_flag.is_set(): 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: if self.push_proc.poll() is not None:
console.log(f"[yellow]⚠️ [{self.room_name}] 推流进程退出,尝试重试...[/yellow]") self.logger.warning(f"{Fore.YELLOW}[{self.room_name}] 推流进程退出{Style.RESET_ALL}")
self._send_ntfy_notification(title="推流中断", message=f"{self.room_name} 推流进程退出") self.stop_flag.set()
attempt += 1
break break
if self.enable_recording and self.record_proc and self.record_proc.poll() is not None: 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.logger.warning(f"{Fore.YELLOW}[{self.room_name}] 录制进程退出,重启中...{Style.RESET_ALL}")
self._terminate_procs() self._terminate_procs()
break break
time.sleep(1) time.sleep(1)
except Exception as e: except Exception as e:
console.log(f"[red]❌ [{self.room_name}] FFmpeg 异常: {e}[/red]") self.logger.error(f"{Fore.RED}[{self.room_name}] FFmpeg 错误: {e}{Style.RESET_ALL}")
self._send_ntfy_notification(title="FFmpeg 错误", message=f"{self.room_name} FFmpeg 异常: {str(e)}")
attempt += 1
finally: finally:
self._terminate_procs() self._terminate_procs()
if not self.stop_flag.is_set() and attempt < max_attempts: if not self.stop_flag.is_set():
time.sleep(2) time.sleep(1)
elif attempt >= max_attempts:
console.log(f"[red]❌ [{self.room_name}] 达到最大重试次数,停止推流[/red]")
self.stop_flag.set()
def _terminate_procs(self) -> None: def _terminate_procs(self):
for proc in [self.push_proc, self.record_proc]: for proc in [self.push_proc, self.record_proc]:
if proc and proc.poll() is None: if proc and proc.poll() is None:
proc.terminate() proc.terminate()
@@ -186,250 +186,157 @@ class DouyinToYouTubeStreamer:
proc.kill() proc.kill()
self.push_proc, self.record_proc = None, None self.push_proc, self.record_proc = None, None
def _send_ntfy_notification(self, title: str, message: str) -> None: def _monitor_thread(self):
ntfy_url = "https://ntfy.sh/your_topic" while not self.stop_flag.is_set():
try: uptime = int(time.time() - self.start_time) if self.start_time else 0
requests.post(ntfy_url, data=message.encode('utf-8'), headers={"Title": title.encode('utf-8')}) m, s = divmod(uptime, 60)
console.log(f"[cyan]📢 ntfy 通知已发送: {title}[/cyan]") status = f"{Fore.GREEN}推流中{Style.RESET_ALL}" if self.pushing else f"{Fore.YELLOW}启动中{Style.RESET_ALL}"
except Exception as e: sys.stdout.write(f"\r[{self.room_name}] {status} | 时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k")
console.log(f"[red]❌ ntfy 通知发送失败: {e}[/red]") sys.stdout.flush()
time.sleep(1)
def _monitor_thread(self) -> None: def stop(self):
with Progress(console=console) as progress: self.logger.info(f"{Fore.YELLOW}[{self.room_name}] 停止推流和录制{Style.RESET_ALL}")
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.stop_flag.set()
self._terminate_procs() self._terminate_procs()
# ----------------- 抖音流解析 ----------------- # ----------------- 其他函数保持不变 -----------------
async def get_douyin_real_url(record_url: str, record_quality: str = "原画", proxy_addr: Optional[str] = None, async def get_douyin_real_url(record_url: str, index: int, total: int, record_quality="原画", proxy_addr=None, cookies='') -> Tuple[Optional[str], Optional[str]]:
cookies: str = '', max_retries: int = 2) -> Tuple[Optional[str], str]: def update_status(status: str, nickname: str = "未知主播"):
display_url = record_url[:40] + "..." if len(record_url) > 40 else record_url
display_nickname = nickname[:20] + "..." if len(nickname) > 20 else nickname
sys.stdout.write(f"\r{index}/{total} 检查: {display_url} [{display_nickname}] [{status}]")
sys.stdout.flush()
headers = { 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/", "Referer": "https://www.douyin.com/",
"Origin": "https://www.douyin.com", "Origin": "https://www.douyin.com"
"Accept": "application/json, text/javascript, */*; q=0.01",
} }
session_args = {"headers": headers} session_args = {"headers": headers}
if proxy_addr: if proxy_addr:
session_args["proxy"] = proxy_addr session_args["proxy"] = proxy_addr
if cookies:
session_args["cookies"] = cookies
anchor_name = "未知" update_status(f"{Fore.YELLOW}检测中...{Style.RESET_ALL}")
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
json_data = None
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
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:
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}] 多次尝试后无法获取直播流,主播: {anchor_name}[/red]")
DouyinToYouTubeStreamer()._send_ntfy_notification(title="直播流解析失败", message=f"多次尝试后无法获取直播流: {record_url}")
return None, anchor_name
except Exception as e:
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: str, youtube_key: str, real_url: Optional[str] = None, async with aiohttp.ClientSession(**session_args) as session:
room_name: Optional[str] = None, proxy_addr: Optional[str] = None, try:
cookies: str = '', bitrate: int = 5000, enable_recording: bool = False) -> Optional['DouyinToYouTubeStreamer']: async with session.get(record_url, cookies=cookies, timeout=10) as resp:
if real_url is None: text = await resp.text()
real_url, anchor_name = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) if not text.strip() or "captcha" in text.lower():
if not real_url: update_status(f"{Fore.RED}失败: 验证码{Style.RESET_ALL}")
console.log(f"[red]❌ [{record_url}] 无法获取真实直播流,主播: {anchor_name}[/red]") return None, None
return None
console.log(f"[green]✅ [{record_url}] 真实直播流: {real_url}[/green]") json_data = await (spider.get_douyin_stream_data if 'v.douyin.com' not in record_url and '/user/' not in record_url
streamer = DouyinToYouTubeStreamer(bitrate=bitrate, enable_recording=enable_recording, max_minutes=None) else spider.get_douyin_app_stream_data)(url=record_url, proxy_addr=proxy_addr, cookies=cookies)
streamer.start_stream(real_url, youtube_key, room_name or f"抖音_{anchor_name or record_url.split('/')[-1]}")
if not json_data or not isinstance(json_data, dict):
update_status(f"{Fore.RED}失败: 无流{Style.RESET_ALL}")
return None, None
# 尝试多种字段获取主播名
nickname = (
json_data.get('room_title') or
json_data.get('title') or
json_data.get('user', {}).get('nickname') or
json_data.get('owner', {}).get('nickname') or
json_data.get('stream', {}).get('owner', {}).get('nickname') or
"未知主播"
)
update_status(f"{Fore.YELLOW}解析中...{Style.RESET_ALL}", nickname)
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"):
update_status(f"{Fore.GREEN}成功{Style.RESET_ALL}", nickname)
return port_info.get("record_url"), nickname
update_status(f"{Fore.RED}失败: 无效流{Style.RESET_ALL}", nickname)
return None, None
except Exception:
update_status(f"{Fore.RED}失败: 错误{Style.RESET_ALL}")
return None, None
def start_douyin_to_youtube(record_url: str, youtube_key: str, real_url: Optional[str] = None, nickname: Optional[str] = None,
proxy_addr=None, cookies='', bitrate=5000, enable_recording=False):
if real_url is None or nickname is None:
real_url, nickname = asyncio.run(get_douyin_real_url(record_url, 0, 0, proxy_addr=proxy_addr, cookies=cookies))
if not real_url:
logger.error(f"{Fore.RED}无法获取 {record_url[:50]}... 的流{Style.RESET_ALL}")
return None
logger.info(f"{Fore.GREEN}推流自: {real_url[:50]}... [{nickname}]{Style.RESET_ALL}")
streamer = DouyinToYouTubeStreamer(bitrate=bitrate, enable_recording=enable_recording)
streamer.start_stream(real_url, youtube_key, room_name=nickname)
return streamer return streamer
# ----------------- 自动调度 ----------------- def load_config():
def load_config() -> Optional[str]:
config = configparser.ConfigParser() config = configparser.ConfigParser()
try: config.read("config.ini", encoding="utf-8")
config.read("config.ini", encoding="utf-8") return config.get("youtube", "key", fallback=None)
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() -> list: def load_list(last_mtime=0):
urls = []
seen = set()
try: try:
mtime = os.path.getmtime("list.ini")
if mtime <= last_mtime:
return None, last_mtime
with open("list.ini", "r", encoding="utf-8") as f: with open("list.ini", "r", encoding="utf-8") as f:
for line in f: urls = [line.strip() for line in f if line.strip() and not line.startswith("#")]
url = line.strip() logger.info(f"{Fore.CYAN}加载 {len(urls)} 个唯一 URL{Style.RESET_ALL}")
if url and not url.startswith("#") and url not in seen: return urls, mtime
urls.append(url)
seen.add(url)
console.log(f"[cyan]📋 已加载 {len(urls)} 个直播间 URL[/cyan]")
return urls
except FileNotFoundError: except FileNotFoundError:
console.log("[red]❌ 未找到 list.ini 文件[/red]") logger.error(f"{Fore.RED}未找到 list.ini 文件{Style.RESET_ALL}")
except Exception as e: return [], last_mtime
console.log(f"[red]❌ 加载 list.ini 失败: {e}[/red]")
return []
async def find_live_and_stream(youtube_key: str) -> Optional['DouyinToYouTubeStreamer']:
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, 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:
console.log(f"[yellow]⚠️ [{url}] 主播 {anchor_name} 未开播,继续检测下一个[/yellow]")
console.log("[yellow]⚠️ 所有直播间检测完成,未发现直播[/yellow]")
return None
# ----------------- 主程序 -----------------
if __name__ == "__main__": if __name__ == "__main__":
app = typer.Typer() youtube_key = load_config()
if not youtube_key:
logger.error(f"{Fore.RED}config.ini 中无 YouTube Key{Style.RESET_ALL}")
sys.exit(1)
@app.command() streamer_instance = None
def start(): last_mtime = 0
youtube_key = load_config() urls = []
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): def signal_handler(sig, frame):
if CONFIG.streamer_instance: if streamer_instance:
CONFIG.streamer_instance.stop() streamer_instance.stop()
console.log("[yellow]🛑 程序终止[/yellow]") sys.stdout.write("\r" + " " * 80 + "\r")
sys.stdout.flush()
sys.exit(0) sys.exit(0)
signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGTERM, signal_handler)
app()
logger.info(f"{Fore.GREEN}启动抖音直播检测{Style.RESET_ALL}")
while True:
urls_result, new_mtime = load_list(last_mtime)
if urls_result is not None:
urls = urls_result
last_mtime = new_mtime
if not streamer_instance or (hasattr(streamer_instance, 'stop_flag') and streamer_instance.stop_flag.is_set()):
if streamer_instance:
logger.info(f"{Fore.YELLOW}直播中断,重新遍历 URL 列表{Style.RESET_ALL}")
sys.stdout.write("\r" + " " * 80 + "\r")
sys.stdout.flush()
time.sleep(5) # 中断后短暂延迟
streamer_instance = None
if not urls:
sys.stdout.write(f"\r{Fore.YELLOW}无URL列表,10秒后重试...{Style.RESET_ALL}")
sys.stdout.flush()
time.sleep(10)
continue
for i, url in enumerate(urls, 1):
real_url, nickname = asyncio.run(get_douyin_real_url(url, i, len(urls)))
sys.stdout.write("\r" + " " * 80 + "\r")
sys.stdout.flush()
if real_url and nickname:
logger.info(f"{Fore.GREEN}发现直播: {url[:50]}... [{nickname}]{Style.RESET_ALL}")
streamer_instance = start_douyin_to_youtube(url, youtube_key, real_url=real_url, nickname=nickname)
break
if not streamer_instance:
sys.stdout.write(f"\r{Fore.YELLOW}未找到直播,60秒后重试...{Style.RESET_ALL}")
sys.stdout.flush()
time.sleep(60)
else:
time.sleep(10)

15633
logs/app.log Normal file

File diff suppressed because one or more lines are too long

21685
logs/app.log.1 Normal file

File diff suppressed because one or more lines are too long

21722
logs/app.log.2 Normal file

File diff suppressed because one or more lines are too long

21793
logs/app.log.3 Normal file

File diff suppressed because one or more lines are too long

21645
logs/app.log.4 Normal file

File diff suppressed because one or more lines are too long

21719
logs/app.log.5 Normal file

File diff suppressed because one or more lines are too long