import json
import secrets
import asyncio
import time
import ntpath
from datetime import datetime, timedelta
from pathlib import Path

from fastapi import APIRouter, HTTPException, Depends, Request
from fastapi.responses import FileResponse, StreamingResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, Field
from typing import Optional
from database import get_db
from routes.auth import require_auth
from services.stream_service import StreamService
from services.media_service import media_service
from services.load_balancer import lb
from config import settings
from models.audit import ControlCommand, ControlSession, ViewerSession

router = APIRouter(prefix="/api/streams", tags=["streams"])


class RenditionInput(BaseModel):
    label: str
    enabled: bool = True
    width: int
    height: int
    fps: int = 30
    codec: str = "libx264"
    bitrate: str = "2500k"
    crf: int = 23
    preset: str = "veryfast"
    audio_bitrate: str = "128k"
    keyframe_seconds: int = 2


class StreamCreate(BaseModel):
    title: str
    description: Optional[str] = None
    cover_url: Optional[str] = None
    scheduled_at: Optional[str] = None
    replay_file: Optional[str] = None
    replay_url: Optional[str] = None
    chapters: list[dict] = Field(default_factory=list)
    transcode_profile: str = "standard"
    renditions: Optional[list[RenditionInput]] = None
    homepage_visible: bool = True
    remote_control_enabled: bool = False


class StreamUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    cover_url: Optional[str] = None
    scheduled_at: Optional[str] = None
    replay_file: Optional[str] = None
    replay_url: Optional[str] = None
    chapters: Optional[list[dict]] = None
    transcode_profile: Optional[str] = None
    renditions: Optional[list[RenditionInput]] = None
    homepage_visible: Optional[bool] = None
    remote_control_enabled: Optional[bool] = None


class RemoteSessionCreate(BaseModel):
    expires_in: int = 300
    operator: str = "administrator"


class RemoteCommand(BaseModel):
    token: str
    action: str
    value: str | None = None
    targets: list[str] = Field(default_factory=list)


class ViewerHeartbeat(BaseModel):
    session_id: str
    node_id: str | None = None
    node_name: str | None = None
    quality: str | None = None
    playback_state: str = "idle"
    buffered_seconds: float | None = None
    error: str | None = None


class CommandResult(BaseModel):
    result: str = "applied"


VIEWER_ONLINE_SECONDS = 45


def _media_path(filename: str | None) -> Path | None:
    if not filename:
        return None
    root = Path(settings.replay_media_dir).resolve()
    supplied = Path(filename)
    windows_drive, _ = ntpath.splitdrive(filename)
    if supplied.is_absolute() or supplied.drive or windows_drive or ".." in supplied.parts:
        raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
    candidate = (root / supplied).resolve()
    if candidate != root and root not in candidate.parents:
        raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
    if not candidate.is_file():
        raise HTTPException(status_code=422, detail="回放文件不在允许的媒体目录中")
    return candidate


def _valid_replay_url(url: str | None) -> str | None:
    if not url:
        return None
    url = url.strip()
    if not (url.startswith("http://") or url.startswith("https://")):
        raise HTTPException(status_code=422, detail="回放 URL 必须是 http(s) 地址")
    if " " in url or "\n" in url or "\r" in url:
        raise HTTPException(status_code=422, detail="回放 URL 格式无效")
    return url


def _status_payload(stream) -> dict:
    return {
        "stream_id": stream.id,
        "status": stream.status.value,
        "replay_local_url": f"/api/streams/{stream.id}/replay" if stream.replay_file else None,
        "replay_url": stream.replay_url,
        "chapters": json.loads(stream.chapters_json or "[]"),
    }


def _normalize_chapters(chapters: list[dict]) -> list:
    normalized = []
    for chapter in chapters:
        title = str(chapter.get("title", "")).strip()
        try:
            chapter_time = float(chapter.get("time", 0))
        except (TypeError, ValueError) as exc:
            raise HTTPException(status_code=422, detail="章节时间必须是非负数字") from exc
        if not title or chapter_time < 0:
            raise HTTPException(status_code=422, detail="章节标题不能为空，时间必须是非负数字")
        normalized.append({"title": title, "time": chapter_time})
    return sorted(normalized, key=lambda chapter: chapter["time"])


def _normalize_renditions(renditions: list[dict]) -> list:
    codecs = {"libx264", "libx265"}
    presets = {"ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow"}
    normalized = []
    labels = set()
    for rendition in renditions:
        label = str(rendition.get("label", "")).strip()
        if not label or not label.replace("-", "").replace("_", "").isalnum() or label in labels:
            raise HTTPException(status_code=422, detail="清晰度标签无效或重复")
        labels.add(label)
        if not rendition.get("enabled", True):
            continue
        width, height = int(rendition["width"]), int(rendition["height"])
        fps, crf = int(rendition.get("fps", 30)), int(rendition.get("crf", 23))
        keyframe_seconds = int(rendition.get("keyframe_seconds", 2))
        codec, preset = rendition.get("codec", "libx264"), rendition.get("preset", "veryfast")
        bitrate, audio_bitrate = str(rendition.get("bitrate", "")), str(rendition.get("audio_bitrate", ""))
        if not (160 <= width <= 7680 and 90 <= height <= 4320 and 1 <= fps <= 120 and 0 <= crf <= 51 and 1 <= keyframe_seconds <= 10):
            raise HTTPException(status_code=422, detail="清晰度参数超出允许范围")
        if codec not in codecs or preset not in presets:
            raise HTTPException(status_code=422, detail="编码器或预设不受支持")
        if not bitrate[:-1].isdigit() or bitrate[-1:] not in {"k", "M"} or not audio_bitrate[:-1].isdigit() or audio_bitrate[-1:] not in {"k", "M"}:
            raise HTTPException(status_code=422, detail="码率格式必须类似 2500k 或 5M")
        normalized.append({"label": label, "enabled": True, "width": width, "height": height, "fps": fps, "codec": codec, "bitrate": bitrate, "crf": crf, "preset": preset, "audio_bitrate": audio_bitrate, "keyframe_seconds": keyframe_seconds})
    if not normalized:
        raise HTTPException(status_code=422, detail="至少启用一个清晰度")
    return normalized


def _serialize(s, include_private: bool = True):
    data = {
        "id": s.id, "title": s.title, "description": s.description,
        "cover_url": s.cover_url, "status": s.status.value,
        "scheduled_at": str(s.scheduled_at) if s.scheduled_at else None,
        "started_at": str(s.started_at) if s.started_at else None,
        "ended_at": str(s.ended_at) if s.ended_at else None,
        "multicast_enabled": s.multicast_enabled,
        "replay_file": s.replay_file,
        "replay_url": s.replay_url,
        "chapters": json.loads(s.chapters_json or "[]"),
        "transcode_profile": s.transcode_profile,
        "renditions": json.loads(getattr(s, "renditions_json", None)) if getattr(s, "renditions_json", None) else None,
        "homepage_visible": s.homepage_visible,
        "remote_control_enabled": s.remote_control_enabled,
        "created_at": str(s.created_at),
    }
    if include_private:
        data["stream_key"] = s.stream_key
    return data


def _get_client_ip(request: Request) -> str | None:
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    client = request.headers.get("X-Real-IP")
    if client:
        return client
    if request.client:
        return request.client.host
    return None


@router.get("")
async def list_streams(admin: bool = False, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    streams = await svc.get_all_streams() if admin else await svc.get_public_streams()
    return [_serialize(s, include_private=admin) for s in streams]


@router.get("/catalog/events")
async def catalog_events(request: Request, db: AsyncSession = Depends(get_db)):
    async def events():
        previous = None
        while not await request.is_disconnected():
            db.expire_all()
            streams = await StreamService(db).get_public_streams()
            payload = json.dumps(
                [_serialize(stream, include_private=False) for stream in streams],
                ensure_ascii=False,
            )
            if payload != previous:
                yield f"event: catalog\ndata: {payload}\n\n"
                previous = payload
            else:
                yield ": keep-alive\n\n"
            await asyncio.sleep(2)

    return StreamingResponse(events(), media_type="text/event-stream", headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
    })


@router.post("/{stream_id}/viewer-heartbeat")
async def viewer_heartbeat(stream_id: int, body: ViewerHeartbeat,
                           db: AsyncSession = Depends(get_db)):
    stream = await StreamService(db).get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="节目不存在")
    session_id = body.session_id.strip()
    if not session_id or len(session_id) > 64:
        raise HTTPException(status_code=422, detail="观众会话 ID 无效")
    viewer = await db.get(ViewerSession, session_id)
    if viewer and viewer.stream_id != stream_id:
        raise HTTPException(status_code=409, detail="观众会话已用于其他节目")
    if not viewer:
        viewer = ViewerSession(session_id=session_id, stream_id=stream_id)
        db.add(viewer)
    viewer.node_id = body.node_id
    viewer.node_name = body.node_name
    viewer.quality = body.quality
    viewer.playback_state = body.playback_state[:32]
    viewer.buffered_seconds = body.buffered_seconds
    viewer.error = body.error[:1000] if body.error else None
    viewer.last_heartbeat = datetime.utcnow()
    await db.commit()
    return {"ok": True, "remote_adjusted": viewer.remote_adjusted}


@router.get("/{stream_id}/viewer-sessions")
async def list_viewer_sessions(stream_id: int, db: AsyncSession = Depends(get_db),
                               _: bool = Depends(require_auth)):
    cutoff = datetime.utcnow() - timedelta(seconds=VIEWER_ONLINE_SECONDS)
    viewers = (await db.execute(select(ViewerSession).where(
        ViewerSession.stream_id == stream_id,
        ViewerSession.last_heartbeat >= cutoff,
    ).order_by(ViewerSession.last_heartbeat.desc()))).scalars().all()
    return [{
        "session_id": viewer.session_id,
        "node_id": viewer.node_id,
        "node_name": viewer.node_name,
        "quality": viewer.quality,
        "playback_state": viewer.playback_state,
        "buffered_seconds": viewer.buffered_seconds,
        "error": viewer.error,
        "remote_adjusted": viewer.remote_adjusted,
        "last_heartbeat": viewer.last_heartbeat.isoformat(),
    } for viewer in viewers]


@router.get("/{stream_id}/control-audit")
async def control_audit(stream_id: int, db: AsyncSession = Depends(get_db),
                        _: bool = Depends(require_auth)):
    commands = (await db.execute(select(ControlCommand).where(
        ControlCommand.stream_id == stream_id,
    ).order_by(ControlCommand.created_at.desc()).limit(100))).scalars().all()
    return [{
        "id": command.id,
        "operator": command.operator,
        "target_session_id": command.target_session_id,
        "action": command.action,
        "value": command.value,
        "status": command.status,
        "result": command.result,
        "created_at": command.created_at.isoformat(),
        "claimed_at": command.claimed_at.isoformat() if command.claimed_at else None,
    } for command in commands]


@router.post("/{stream_id}/remote-command/{command_id}/result")
async def report_command_result(stream_id: int, command_id: int, body: CommandResult,
                                session_id: str, db: AsyncSession = Depends(get_db)):
    command = await db.get(ControlCommand, command_id)
    if not command or command.stream_id != stream_id or command.target_session_id != session_id:
        raise HTTPException(status_code=404, detail="控制命令不存在")
    command.status = "completed"
    command.result = body.result[:1000]
    await db.commit()
    return {"ok": True}


@router.get("/media-files")
async def list_media_files(_: bool = Depends(require_auth)):
    root = Path(settings.replay_media_dir)
    root.mkdir(parents=True, exist_ok=True)
    extensions = {".mp4", ".m4v", ".mov", ".webm", ".mkv"}
    return [p.name for p in root.iterdir() if p.is_file() and p.suffix.lower() in extensions]


@router.post("")
async def create_stream(body: StreamCreate, db: AsyncSession = Depends(get_db),
                         _: bool = Depends(require_auth)):
    title = body.title.strip()
    if not title:
        raise HTTPException(status_code=422, detail="节目标题不能为空")
    _media_path(body.replay_file)
    _valid_replay_url(body.replay_url)
    if body.transcode_profile not in settings.transcode_profiles:
        raise HTTPException(status_code=422, detail="转码配置不存在")
    renditions = _normalize_renditions([item.model_dump() for item in body.renditions]) if body.renditions is not None else None
    svc = StreamService(db)
    try:
        stream = await svc.create_stream(
            title=title,
            description=body.description,
            cover_url=body.cover_url,
            scheduled_at=body.scheduled_at,
            replay_file=body.replay_file,
            replay_url=body.replay_url,
            chapters=body.chapters,
            transcode_profile=body.transcode_profile,
            renditions_json=json.dumps(renditions, ensure_ascii=False) if renditions is not None else None,
            homepage_visible=body.homepage_visible,
            remote_control_enabled=body.remote_control_enabled,
        )
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except Exception as exc:
        await db.rollback()
        raise HTTPException(status_code=500, detail="创建节目失败，请检查日期格式和数据库状态") from exc
    rtmp_url = media_service.get_rtmp_url(stream.stream_key)
    return {
        "id": stream.id,
        "title": stream.title,
        "stream_key": stream.stream_key,
        "rtmp_url": rtmp_url,
        "status": stream.status.value,
    }


@router.get("/{stream_id}")
async def get_stream(stream_id: int, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    s = await svc.get_stream(stream_id)
    if not s:
        raise HTTPException(status_code=404, detail="Stream not found")
    return _serialize(s)


@router.put("/{stream_id}")
async def update_stream(stream_id: int, body: StreamUpdate, db: AsyncSession = Depends(get_db),
                        _: bool = Depends(require_auth)):
    svc = StreamService(db)
    existing = await svc.get_stream(stream_id)
    if not existing:
        raise HTTPException(status_code=404, detail="Stream not found")
    updates = body.model_dump(exclude_unset=True)
    replay_fields = {"replay_file", "replay_url", "chapters"}
    if replay_fields.intersection(updates) and existing.status.value != "ended":
        raise HTTPException(status_code=409, detail="仅已结束的节目可以配置回放和章节")

    if updates.get("replay_file"):
        _media_path(updates["replay_file"])
    if "replay_url" in updates and updates["replay_url"] is not None:
        updates["replay_url"] = _valid_replay_url(updates["replay_url"])

    final_replay_file = updates.get("replay_file", existing.replay_file)
    final_replay_url = updates.get("replay_url", existing.replay_url)
    if final_replay_file and final_replay_url:
        raise HTTPException(status_code=409, detail="本地回放文件和远端回放地址只能选择一种")

    if "transcode_profile" in updates and updates["transcode_profile"] not in settings.transcode_profiles:
        raise HTTPException(status_code=422, detail="转码配置不存在")
    if "renditions" in updates:
        renditions = updates.pop("renditions")
        updates["renditions_json"] = (
            json.dumps(_normalize_renditions(renditions), ensure_ascii=False)
            if renditions is not None else None
        )
    if "chapters" in updates:
        chapters = updates.pop("chapters")
        if chapters and not (final_replay_file or final_replay_url):
            raise HTTPException(status_code=409, detail="请先配置回放来源再编辑章节")
        updates["chapters_json"] = json.dumps(_normalize_chapters(chapters or []), ensure_ascii=False)
    if updates:
        await svc.update_stream(stream_id, **updates)
    return {"ok": True}


@router.delete("/{stream_id}")
async def delete_stream(stream_id: int, db: AsyncSession = Depends(get_db),
                        _: bool = Depends(require_auth)):
    svc = StreamService(db)
    existing = await svc.get_stream(stream_id)
    if not existing:
        raise HTTPException(status_code=404, detail="Stream not found")
    await svc.delete_stream(stream_id)
    return {"ok": True}


@router.get("/{stream_id}/watch")
async def get_watch_urls(stream_id: int, request: Request, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    stream = await svc.get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Stream not found")
    client_ip = _get_client_ip(request)
    urls = await svc.get_watch_urls(stream.stream_key, client_ip)
    return {
        "stream_id": stream.id,
        "title": stream.title,
        "status": stream.status.value,
        "multicast_available": urls["multicast_available"],
        "selected_node": urls.get("selected_node"),
        "lines": urls["lines"],
        "qualities": urls["qualities"],
        "description": stream.description,
        "scheduled_at": str(stream.scheduled_at) if stream.scheduled_at else None,
        "replay_local_url": f"/api/streams/{stream.id}/replay" if stream.replay_file else None,
        "replay_url": stream.replay_url,
        "chapters": json.loads(stream.chapters_json or "[]"),
        "remote_control_enabled": stream.remote_control_enabled,
    }


@router.get("/{stream_id}/events")
async def stream_events(stream_id: int, request: Request, db: AsyncSession = Depends(get_db)):
    if not await StreamService(db).get_stream(stream_id):
        raise HTTPException(status_code=404, detail="Stream not found")

    async def events():
        previous = None
        while not await request.is_disconnected():
            stream = await StreamService(db).get_stream(stream_id)
            if not stream:
                yield "event: deleted\ndata: {}\n\n"
                return
            payload = json.dumps(_status_payload(stream), ensure_ascii=False)
            if payload != previous:
                yield f"event: status\ndata: {payload}\n\n"
                previous = payload
            else:
                yield ": keep-alive\n\n"
            await asyncio.sleep(2)

    return StreamingResponse(events(), media_type="text/event-stream", headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
    })


@router.get("/{stream_id}/replay")
async def serve_replay(stream_id: int, preview: str | None = None,
                       db: AsyncSession = Depends(get_db)):
    stream = await StreamService(db).get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Program not found")
    path = _media_path(preview if preview is not None else stream.replay_file)
    if not path:
        raise HTTPException(status_code=404, detail="Replay not available")
    return FileResponse(path)


@router.post("/{stream_id}/remote-session")
async def create_remote_session(stream_id: int, body: RemoteSessionCreate,
                                db: AsyncSession = Depends(get_db),
                                _: bool = Depends(require_auth)):
    stream = await StreamService(db).get_stream(stream_id)
    if not stream or not stream.remote_control_enabled:
        raise HTTPException(status_code=409, detail="节目未启用远程控制")
    expires_in = min(max(body.expires_in, 30), 3600)
    session = ControlSession(
        token=secrets.token_urlsafe(24),
        stream_id=stream_id,
        operator=body.operator.strip() or "administrator",
        expires_at=datetime.utcnow() + timedelta(seconds=expires_in),
    )
    db.add(session)
    await db.commit()
    return {"token": session.token, "expires_in": expires_in}


@router.post("/{stream_id}/remote-command")
async def send_remote_command(stream_id: int, body: RemoteCommand,
                              db: AsyncSession = Depends(get_db),
                              _: bool = Depends(require_auth)):
    session = (await db.execute(select(ControlSession).where(
        ControlSession.token == body.token,
        ControlSession.stream_id == stream_id,
    ))).scalar_one_or_none()
    if not session or session.expires_at < datetime.utcnow():
        raise HTTPException(status_code=403, detail="远程控制会话无效或已过期")
    stream = await StreamService(db).get_stream(stream_id)
    if not stream or not stream.remote_control_enabled:
        raise HTTPException(status_code=409, detail="节目未启用远程控制")
    if body.action not in {"switch_node", "switch_quality", "restore_auto"}:
        raise HTTPException(status_code=422, detail="不支持的远程命令")
    if not body.targets:
        raise HTTPException(status_code=422, detail="至少选择一个观众会话")
    viewers = (await db.execute(select(ViewerSession).where(
        ViewerSession.stream_id == stream_id,
        ViewerSession.session_id.in_(body.targets),
    ))).scalars().all()
    found = {viewer.session_id for viewer in viewers}
    if found != set(body.targets):
        raise HTTPException(status_code=404, detail="部分观众会话不存在")
    commands = []
    for target in body.targets:
        command = ControlCommand(
            control_session_id=session.id,
            stream_id=stream_id,
            target_session_id=target,
            operator=session.operator,
            action=body.action,
            value=body.value,
        )
        db.add(command)
        commands.append(command)
    await db.commit()
    return {"ok": True, "count": len(commands), "command_ids": [command.id for command in commands]}


@router.get("/{stream_id}/remote-command")
async def poll_remote_command(stream_id: int, session_id: str,
                              db: AsyncSession = Depends(get_db)):
    viewer = await db.get(ViewerSession, session_id)
    if not viewer or viewer.stream_id != stream_id:
        raise HTTPException(status_code=404, detail="观众会话不存在")
    command = (await db.execute(select(ControlCommand).where(
        ControlCommand.stream_id == stream_id,
        ControlCommand.target_session_id == session_id,
        ControlCommand.status == "pending",
    ).order_by(ControlCommand.created_at).limit(1))).scalar_one_or_none()
    if not command:
        return None
    command.status = "claimed"
    command.claimed_at = datetime.utcnow()
    viewer.remote_adjusted = command.action != "restore_auto"
    await db.commit()
    return {"id": command.id, "action": command.action, "value": command.value,
            "created_at": command.created_at.isoformat()}


@router.post("/{stream_id}/go-live")
async def stream_go_live(stream_id: int, db: AsyncSession = Depends(get_db),
                         _: bool = Depends(require_auth)):
    svc = StreamService(db)
    stream = await svc.get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Stream not found")
    await svc.set_stream_live(stream.stream_key)
    return {"ok": True, "status": "live"}


@router.post("/{stream_id}/end")
async def stream_end(stream_id: int, db: AsyncSession = Depends(get_db),
                     _: bool = Depends(require_auth)):
    svc = StreamService(db)
    stream = await svc.get_stream(stream_id)
    if not stream:
        raise HTTPException(status_code=404, detail="Stream not found")
    await svc.set_stream_ended(stream.stream_key)
    return {"ok": True, "status": "ended"}


@router.get("/detail/{stream_key}")
async def get_stream_by_key(stream_key: str, request: Request, db: AsyncSession = Depends(get_db)):
    svc = StreamService(db)
    s = await svc.get_stream_by_key(stream_key)
    if not s:
        raise HTTPException(status_code=404, detail="Stream not found")
    client_ip = _get_client_ip(request)
    urls = await svc.get_watch_urls(stream_key, client_ip)
    return {
        "stream_id": s.id,
        "title": s.title,
        "status": s.status.value,
        "multicast_available": urls["multicast_available"],
        "selected_node": urls.get("selected_node"),
        "lines": urls["lines"],
        "qualities": urls["qualities"],
    }