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