"""Dashboard and resource-tab UI."""

from __future__ import annotations

import time
from typing import Any

import pandas as pd
import streamlit as st

from media_library_viewer.clients.resources import (
    disk_space,
    read_resource_metrics,
    resource_collector_debug_info,
    resource_collector_status,
    restart_resource_collector,
    start_resource_collector,
    stop_resource_collector,
)
from media_library_viewer.utils import human_size


def format_rate_bytes(bytes_per_second: float | int | None) -> str:
    if bytes_per_second is None:
        return ""
    return f"{human_size(bytes_per_second)}/s"


def rate_scale(max_value: float | int | None) -> tuple[float, str]:
    value = abs(float(max_value or 0))
    units = [(1_000_000_000_000, "TB/s"), (1_000_000_000, "GB/s"), (1_000_000, "MB/s"), (1_000, "KB/s"), (1, "B/s")]
    for divisor, suffix in units:
        if value >= divisor or divisor == 1:
            return float(divisor), suffix
    return 1.0, units[-1][1]


def scaled_rate_chart_df(chart_df: pd.DataFrame, columns: list[str], labels: list[str]) -> tuple[pd.DataFrame, str]:
    max_value = chart_df[columns].max(numeric_only=True).max()
    divisor, suffix = rate_scale(max_value)
    scaled = chart_df[columns].copy() / divisor
    scaled.columns = [f"{label} ({suffix})" for label in labels]
    return scaled, suffix


def format_elapsed(seconds: float | int | None) -> str:
    if seconds is None:
        return ""
    seconds = float(seconds)
    if seconds < 60:
        return f"{seconds:.1f}s"
    minutes = int(seconds // 60)
    remainder = seconds % 60
    if minutes < 60:
        return f"{minutes}m {remainder:.0f}s"
    hours = minutes // 60
    minutes = minutes % 60
    return f"{hours}h {minutes}m"


def render_media_overview(cached_media_counts, cached_library_counts, base_url: str, api_key: str,
                          user_id: str) -> None:
    """Render dashboard counts for movies/series/episodes and per-library breakdown."""
    st.subheader("Media library overview")
    try:
        counts = cached_media_counts(base_url, api_key, user_id)
    except Exception as exc:
        st.warning(f"Could not load Jellyfin media counts: {exc}")
        return

    # Top-level totals
    total_items = counts.get("movies", 0) + counts.get("series", 0) + counts.get("episodes", 0)
    top_cols = st.columns(4)
    top_cols[0].metric("Total items", f"{total_items:,}")
    top_cols[1].metric("Movies", f"{counts.get('movies', 0):,}")
    top_cols[2].metric("Series", f"{counts.get('series', 0):,}")
    top_cols[3].metric("Episodes", f"{counts.get('episodes', 0):,}")

    # Per-library breakdown
    try:
        lib_counts = cached_library_counts(base_url, api_key, user_id)
    except Exception as exc:
        st.caption(f"Could not load per-library counts: {exc}")
        return

    if not lib_counts:
        return

    st.markdown("**Libraries**")

    movie_libs = [e for e in lib_counts if e.get("type") == "movies"]
    tv_libs = [e for e in lib_counts if e.get("type") == "tvshows"]

    if movie_libs and tv_libs:
        left_col, right_col = st.columns(2)
    elif movie_libs:
        left_col = st.container()
        right_col = None
    elif tv_libs:
        left_col = None
        right_col = st.container()
    else:
        return

    if movie_libs and left_col:
        with left_col:
            st.caption("Movie libraries")
            for entry in movie_libs:
                with st.container(border=True):
                    st.markdown(f"**{entry['library']}**")
                    m_cols = st.columns(2)
                    m_cols[0].metric("Movies", f"{entry['movies']:,}")

    if tv_libs and right_col:
        with right_col:
            st.caption("TV libraries")
            for entry in tv_libs:
                with st.container(border=True):
                    st.markdown(f"**{entry['library']}**")
                    m_cols = st.columns(2)
                    m_cols[0].metric("Series", f"{entry['series']:,}")
                    m_cols[1].metric("Episodes", f"{entry['total']:,}")


def render_now_playing(cached_active_sessions, base_url: str, api_key: str) -> None:
    """Render currently playing users/items and transcoding state."""
    st.subheader("Now playing")
    try:
        sessions = cached_active_sessions(base_url, api_key)
    except Exception as exc:
        st.warning(f"Could not load active sessions: {exc}")
        return

    if not sessions:
        st.caption("No active playback sessions right now.")
        return

    rows = []
    for session in sessions:
        item = session.get("NowPlayingItem") or {}
        session_id = session.get("Id") or ""
        user_name = session.get("UserName") or "Unknown"
        device = session.get("DeviceName") or session.get("Client") or ""
        play_state = session.get("PlayState") or {}
        paused = bool(play_state.get("IsPaused"))
        state_label = "paused" if paused else "playing"

        series = item.get("SeriesName") or ""
        if series:
            title = f"{series} - {item.get('Name', '')}"
        else:
            title = item.get("Name") or "Unknown"

        transcoding = session.get("TranscodingInfo") or {}
        is_transcoding = bool(transcoding)
        transcode_type = []
        if is_transcoding:
            if transcoding.get("IsVideoDirect") is False:
                transcode_type.append("video")
            if transcoding.get("IsAudioDirect") is False:
                transcode_type.append("audio")
            if not transcode_type:
                transcode_type.append("active")

        rows.append(
            {
                "user": user_name,
                "title": title,
                "type": item.get("Type", ""),
                "state": state_label,
                "transcoding": "yes" if is_transcoding else "no",
                "transcoding_type": ", ".join(transcode_type),
                "device": device,
                "session_id": session_id,
            }
        )

    st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)


def render_resource_dashboard(get_ssh_client, ssh_args: tuple, media_root: str, detailed: bool = False) -> None:
    """Render server monitoring summary or detailed charts."""
    st.subheader("Server monitoring details" if detailed else "Server overview")
    host, username, port, key_filename, password = ssh_args
    ssh = get_ssh_client(host, username, port, key_filename, password)

    try:
        status = resource_collector_status(ssh)
    except Exception as exc:
        st.error(f"Could not check resource collector status: {exc}")
        return

    if detailed:
        control_col, start_col, restart_col, stop_col, refresh_col = st.columns([2.3, 1, 1, 1, 1])
        control_col.caption(f"Collector: `{status}` | sample interval: 10s | retention: 7 days / 70k samples")
        if start_col.button("Start metrics", key="monitoring_start_metrics", use_container_width=True):
            try:
                st.success(start_resource_collector(ssh))
            except Exception as exc:
                st.error(str(exc))
        if restart_col.button("Restart", key="monitoring_restart_metrics", use_container_width=True):
            try:
                st.success(restart_resource_collector(ssh))
            except Exception as exc:
                st.error(str(exc))
        if stop_col.button("Stop metrics", key="monitoring_stop_metrics", use_container_width=True):
            try:
                st.info(stop_resource_collector(ssh))
            except Exception as exc:
                st.error(str(exc))
        if refresh_col.button("Refresh", key="monitoring_refresh", use_container_width=True):
            st.rerun()
    else:
        st.caption(f"Collector: `{status}`. Open the Monitoring tab for controls, diagnostics, and detailed charts.")

    try:
        rows = read_resource_metrics(ssh, max_lines=1000)
    except Exception as exc:
        st.error(f"Could not read resource metrics: {exc}")
        rows = []

    disk_path = media_root or "/"
    try:
        space = disk_space(ssh, disk_path)
        used_pct_value = float(str(space.get("used_pct", "0")).rstrip("%") or 0)
        disk_cols = st.columns(4)
        disk_cols[0].metric("Disk used", human_size(space.get("used")))
        disk_cols[1].metric("Disk available", human_size(space.get("available")))
        disk_cols[2].metric("Disk total", human_size(space.get("size")))
        disk_cols[3].metric("Used percent", f"{used_pct_value:.0f}%")
        st.progress(min(max(used_pct_value / 100, 0), 1),
                    text=f"{space.get('mount', disk_path)} on {space.get('filesystem', '')}")
    except Exception as exc:
        st.warning(f"Could not read disk space for {disk_path}: {exc}")

    if not rows:
        if detailed:
            st.info(
                "No monitoring history yet. Click 'Start metrics' and wait at least 10 seconds for the first sample. If this stays empty, use Restart to install the latest collector script.")
            with st.expander("Collector diagnostics"):
                try:
                    st.code(resource_collector_debug_info(ssh))
                except Exception as exc:
                    st.error(f"Could not read collector diagnostics: {exc}")
        else:
            st.info("No monitoring history yet. Open the Monitoring tab to start the collector.")
        return

    df = pd.DataFrame(rows)
    numeric_columns = [
        "ts", "cpu_pct", "iowait_pct", "mem_pct", "net_rx_bytes_per_sec", "net_tx_bytes_per_sec", "disk_read_bps",
        "disk_write_bps"
    ]
    for column in numeric_columns:
        if column in df.columns:
            df[column] = pd.to_numeric(df[column], errors="coerce")
    df = df.dropna(subset=["ts"])
    df["time"] = pd.to_datetime(df["ts"], unit="s", utc=True).dt.tz_convert(None)

    cutoff_ts = time.time() - 3600
    all_sample_count = len(df)
    df = df[df["ts"] >= cutoff_ts]
    if df.empty:
        st.info("No samples in the last hour yet.")
        if detailed:
            with st.expander("Resource sample diagnostics"):
                if all_sample_count:
                    raw_df = pd.DataFrame(rows)
                    st.write(f"Parsed samples: {all_sample_count}")
                    st.write(
                        f"Newest remote sample age: {time.time() - float(raw_df['ts'].astype(float).max()):.0f} seconds")
                    st.dataframe(raw_df.tail(10), use_container_width=True, hide_index=True)
                else:
                    st.write("No parseable samples found.")
        return

    df = df.sort_values("ts")
    latest = df.iloc[-1]
    avg_cpu = df["cpu_pct"].mean()
    max_cpu = df["cpu_pct"].max()
    avg_iowait = df["iowait_pct"].mean() if "iowait_pct" in df.columns else 0
    max_iowait = df["iowait_pct"].max() if "iowait_pct" in df.columns else 0
    avg_mem = df["mem_pct"].mean()
    max_mem = df["mem_pct"].max()
    avg_net_down = df["net_rx_bytes_per_sec"].mean()
    max_net_down = df["net_rx_bytes_per_sec"].max()
    avg_net_up = df["net_tx_bytes_per_sec"].mean()
    max_net_up = df["net_tx_bytes_per_sec"].max()
    avg_disk_read = df["disk_read_bps"].mean()
    max_disk_read = df["disk_read_bps"].max()
    avg_disk_write = df["disk_write_bps"].mean()
    max_disk_write = df["disk_write_bps"].max()

    metric_cols = st.columns(7)
    metric_cols[0].metric("CPU now", f"{latest['cpu_pct']:.1f}%")
    metric_cols[0].caption(f"avg {avg_cpu:.1f}%  \npeak {max_cpu:.1f}%")
    metric_cols[1].metric("IO wait", f"{latest.get('iowait_pct', 0):.1f}%")
    metric_cols[1].caption(f"avg {avg_iowait:.1f}%  \npeak {max_iowait:.1f}%")
    metric_cols[2].metric("RAM now", f"{latest['mem_pct']:.1f}%")
    metric_cols[2].caption(f"avg {avg_mem:.1f}%  \npeak {max_mem:.1f}%")
    metric_cols[3].metric("Network down", format_rate_bytes(latest["net_rx_bytes_per_sec"]))
    metric_cols[3].caption(f"avg {format_rate_bytes(avg_net_down)}  \npeak {format_rate_bytes(max_net_down)}")
    metric_cols[4].metric("Network up", format_rate_bytes(latest["net_tx_bytes_per_sec"]))
    metric_cols[4].caption(f"avg {format_rate_bytes(avg_net_up)}  \npeak {format_rate_bytes(max_net_up)}")
    metric_cols[5].metric("Disk read", format_rate_bytes(latest["disk_read_bps"]))
    metric_cols[5].caption(f"avg {format_rate_bytes(avg_disk_read)}  \npeak {format_rate_bytes(max_disk_read)}")
    metric_cols[6].metric("Disk write", format_rate_bytes(latest["disk_write_bps"]))
    metric_cols[6].caption(f"avg {format_rate_bytes(avg_disk_write)}  \npeak {format_rate_bytes(max_disk_write)}")

    chart_df = df.set_index("time")
    if detailed:
        st.markdown("**CPU, IO wait, and RAM - last hour**")
        chart_cols = ["cpu_pct", "mem_pct"]
        if "iowait_pct" in chart_df.columns:
            chart_cols = ["cpu_pct", "iowait_pct", "mem_pct"]
        st.line_chart(chart_df[chart_cols], use_container_width=True)
    else:
        st.caption(
            "Detailed CPU/IO wait/RAM charts, network, disk I/O, raw samples, and collector controls are available in the Monitoring tab.")
        return

    net_down_df, net_down_suffix = scaled_rate_chart_df(chart_df, ["net_rx_bytes_per_sec"], ["download"])
    net_up_df, net_up_suffix = scaled_rate_chart_df(chart_df, ["net_tx_bytes_per_sec"], ["upload"])
    net_down_col, net_up_col = st.columns(2)
    with net_down_col:
        st.markdown(f"**Network down - last hour ({net_down_suffix})**")
        st.line_chart(net_down_df, use_container_width=True)
    with net_up_col:
        st.markdown(f"**Network up - last hour ({net_up_suffix})**")
        st.line_chart(net_up_df, use_container_width=True)

    disk_read_df, disk_read_suffix = scaled_rate_chart_df(chart_df, ["disk_read_bps"], ["read"])
    disk_write_df, disk_write_suffix = scaled_rate_chart_df(chart_df, ["disk_write_bps"], ["write"])
    disk_read_col, disk_write_col = st.columns(2)
    with disk_read_col:
        st.markdown(f"**Disk read - last hour ({disk_read_suffix})**")
        st.line_chart(disk_read_df, use_container_width=True)
    with disk_write_col:
        st.markdown(f"**Disk write - last hour ({disk_write_suffix})**")
        st.line_chart(disk_write_df, use_container_width=True)

    with st.expander("Raw monitoring samples"):
        st.dataframe(df.sort_values("time", ascending=False), use_container_width=True, hide_index=True)
