This commit is contained in:
eric
2025-09-24 14:55:54 +08:00
parent a49143093c
commit 8f496f2109
5 changed files with 1077 additions and 183 deletions

443
backup/back13 Normal file
View File

@@ -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()

323
backup/bnack12 Normal file
View File

@@ -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)

View File

@@ -1,3 +1,4 @@
https://live.douyin.com/944705936194
https://live.douyin.com/767599969296
https://live.douyin.com/456576000931
https://live.douyin.com/745606325880

440
live.py
View File

@@ -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):
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 _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 "启动中"
print(f"\r[{self.room_name}] {status} | 运行时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k", end="")
progress.update(task, description=f"[cyan]📡 [{self.room_name}] {status} | 时间: {m:02d}:{s:02d} | 码率: {self.bitrate}k[/cyan]")
time.sleep(1)
def stop(self):
logging.info(f"[{self.room_name}] 停止推流和录制...")
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)
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重试...")
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
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}")
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:
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))
real_url, anchor_name = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies))
if not real_url:
logging.error(f"获取真实直播流失败: {record_url}")
console.log(f"[red]❌ [{record_url}] 无法获取真实直播流,主播: {anchor_name}[/red]")
return None
logging.info(f"直播流: {real_url}")
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()
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()
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)
logging.info(f"加载 {len(urls)}唯一直播间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
else:
logging.warning(f"{url} 流检测成功但启动失败,继续下一个...")
else:
logging.info(f"{url} 暂无直播,继续检测...")
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 = {}
# ----------------- Main Program -----------------
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__":
app = typer.Typer()
@app.command()
def start():
youtube_key = load_config()
if not youtube_key:
logging.error("配置文件 config.ini 中未找到 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)
streamer_instance = None
@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)
app()

15
requirements.txt Normal file
View File

@@ -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