Blame for postgres_dashboard.py

"""Public statistics from a single, safely filtered aerial.videos catalog."""

from __future__ import annotations

from datetime import date
import html
import math
from typing import Callable, Mapping, Sequence

import pandas as pd
import plotly.graph_objects as go
import streamlit as st

from live_inventory import VideoCatalog, VideoRecord, filter_records, summarize_records
from city_labels import city_display_name, is_city_label
from dashboard_charts import DATA_BLUE, style_figure


_UNKNOWN = "未知"
_FILTER_KEYS = ("pg_cities", "pg_date_range", "pg_include_unknown_dates",
                "pg_codecs", "pg_periods")
_EXPORT_COLUMNS = (
    "city", "collection_month", "period", "codec", "resolution", "fps",
    "duration_range", "videos", "duration_hours", "size_tb",
    "duration_known_videos", "size_known_videos",
)


def selected_records(catalog: VideoCatalog, state: Mapping) -> tuple[VideoRecord, ...]:
    """Use the same inclusive filters for KPIs, charts, maps and export."""
    # Other pages and unavailable renders have no filter widgets. Their shared
    # facts must still use the last selection, while current widgets override it.
    state = {**state.get("_pg_filter_snapshot", {}),
             **{key: state[key] for key in _FILTER_KEYS if key in state}}
    dates = state.get("pg_date_range", ())
    if isinstance(dates, date):
        dates = (dates, dates)
    start = dates[0] if dates else None
    end = dates[-1] if dates else None
    return filter_records(
        catalog.records,
        cities=state.get("pg_cities", ()),
        start_date=start, end_date=end,
        include_unknown_dates=state.get("pg_include_unknown_dates", True),
        codecs=state.get("pg_codecs", ()),
        periods=state.get("pg_periods", ()),
    )


def _resolution(record: VideoRecord) -> str:
    return f"{record.width} × {record.height}" if record.width and record.height else _UNKNOWN


def _fps(record: VideoRecord) -> str:
    return f"{float(record.fps):g}" if record.fps is not None else _UNKNOWN


def _duration_range(record: VideoRecord) -> str:
    value = record.duration_seconds
    if value is None:
        return _UNKNOWN
    if value < 60:
        return "< 1 min"
    if value < 300:
        return "1–5 min"
    if value < 900:
        return "5–15 min"
    if value < 1800:
        return "15–30 min"
    if value < 3600:
        return "30–60 min"
    return "≥ 60 min"


def aggregate_rows(records: Sequence[VideoRecord]) -> pd.DataFrame:
    """Only coarse, aggregate metadata can be downloaded from the dashboard."""
    groups: dict[tuple, list] = {}
    for record in records:
        key = (record.city, record.collected_on.strftime("%Y-%m") if record.collected_on else _UNKNOWN,
               record.period, record.codec or _UNKNOWN, _resolution(record),
               _fps(record), _duration_range(record))
        totals = groups.setdefault(key, [0, 0.0, 0.0, 0, 0])
        totals[0] += 1
        if record.duration_seconds is not None:
            totals[1] += float(record.duration_seconds) / 3600
            totals[3] += 1
        if record.size_bytes is not None:
            totals[2] += float(record.size_bytes) / 1_000_000_000_000
            totals[4] += 1
    return pd.DataFrame(
        [(city_display_name(key[0]), *key[1:], totals[0], totals[1] if totals[3] else None,
          totals[2] if totals[4] else None, totals[3], totals[4])
         for key, totals in sorted(groups.items())],
        columns=_EXPORT_COLUMNS,
    )



def duration_hours_by(records: Sequence[VideoRecord], key: Callable[[VideoRecord], str]) -> dict[str, float | None]:
    """Sum actual known durations per category; unknown-only is not measured zero."""
    totals = {}
    for record in records:
        bucket = totals.setdefault(key(record), [0.0, False])
        seconds = record.duration_seconds
        if seconds is not None and not isinstance(seconds, bool) and math.isfinite(float(seconds)) and seconds >= 0:
            bucket[0] += float(seconds) / 3600
            bucket[1] = True
    return {value: hours if known else None for value, (hours, known) in totals.items()}


def public_aggregate_rows(records: Sequence[VideoRecord]) -> pd.DataFrame:
    """Keep internal count fields for validation, expose hours and quality states."""
    rows = aggregate_rows(records)
    for kind in ('duration', 'size'):
        known = rows[f'{kind}_known_videos']
        rows[f'{kind}_status'] = ['unknown' if n == 0 else 'partial' if n < total else 'known'
                                  for n, total in zip(known, rows.videos)]
    return rows.drop(columns=['videos', 'duration_known_videos', 'size_known_videos'])

def render_dashboard(load_catalog: Callable, clear_catalog: Callable,
                     translate: Callable[[str], str], *, render_annotations: Callable | None = None) -> None:
    """Refresh the whole dashboard together, never a separate population."""
    def tr(zh: str, en: str | None = None) -> str:
        translated = translate(zh)
        if st.session_state.get("lang") == "en" and translated == zh and en:
            return en
        return translated

    def category(value: str) -> str:
        if is_city_label(value):
            return city_display_name(value)
        labels = {
            _UNKNOWN: "Unknown", "缺少坐标": "Missing coordinates",
            "边界外或坐标异常": "Unassigned city",
            "已归属城市": "Assigned city", "早高峰 (07-09)": "Morning peak (07-09)",
            "日常规": "Daytime", "晚高峰 (17-19)": "Evening peak (17-19)",
            "夜间 (19-07)": "Night (19-07)",
            "未知采集时间": "Unknown collection date", "known": "Known", "partial": "Partly unknown",
            "unknown": "Unknown", "< 1 min": "< 0.017 h", "1–5 min": "0.017–0.083 h",
            "5–15 min": "0.083–0.25 h", "15–30 min": "0.25–0.5 h", "30–60 min": "0.5–1 h", "≥ 60 min": "≥ 1 h",
        }
        if value in {"< 1 min", "1–5 min", "5–15 min", "15–30 min", "30–60 min", "≥ 60 min"}:
            return labels[value]
        if value in {'known', 'partial', 'unknown'}:
            return tr({'known':'已知','partial':'部分未知','unknown':'未知'}[value], labels[value])
        return tr(value, labels.get(value))

    def heading(anchor: str, zh: str, en: str, description: str = "", english: str = "") -> None:
        st.markdown(
            f"<div id='{anchor}' class='dash-section'><h2 class='dash-section-title'>"
            f"{html.escape(tr(zh, en))}</h2></div>", unsafe_allow_html=True,
        )
        if description:
            st.caption(tr(description, english))

    def bar(counts: Mapping[str, float | None], title: str, english: str,
            *, order: Sequence[str] | None = None) -> None:
        keys = [key for key in order if key in counts] if order else sorted(counts, key=lambda key: (counts[key] is None, -(counts[key] or 0), key))
        if order:
            keys.extend(key for key in counts if key not in keys)
        figure = go.Figure(go.Bar(x=[category(key) for key in keys],
                                 y=[counts[key] for key in keys],
                                 marker_color=DATA_BLUE,
                                 hovertemplate="%{x}: %{y:.3f} h<extra></extra>"))
        style_figure(figure, title=tr(title, english), height=320)
        figure.update_yaxes(title='h', rangemode="tozero", tickformat=",.2f")
        st.plotly_chart(figure, width="stretch")

    def unavailable_annotations() -> None:
        if render_annotations is not None:
            render_annotations()
            return
        st.info(tr(
            "当前视频表没有道路、天气、光照、动态目标、审核关键帧或经审定地点标签，相关标注分布与覆盖率暂不可统计。",
            "The video table does not contain road, weather, lighting, dynamic-object, reviewed-keyframe or approved-location labels. Their distributions and annotation coverage are unavailable.",
        ))

    @st.fragment(run_every="60s")
    def render_current_catalog() -> None:
        heading("nav-scenario", "实时视频统计", "Live video statistics")
        if st.button(tr("刷新数据", "Refresh data"), key="pg_refresh"):
            clear_catalog()
        catalog, status = load_catalog()
        if status == "not_configured":
            st.info(tr("尚未配置 PostgreSQL 数据源。", "PostgreSQL data source is not configured."))
            return
        if status != "ok" or catalog is None:
            st.warning(tr("PostgreSQL 数据暂不可用，请稍后刷新。", "PostgreSQL data is unavailable. Please refresh later."))
            return

        # Streamlit removes widget state when an unavailable page does not
        # render its controls. Keep a nonwidget snapshot so recovery preserves
        # the user's filters instead of silently showing the entire inventory.
        saved_filters = st.session_state.get("_pg_filter_snapshot", {})
        for key, value in saved_filters.items():
            if key not in st.session_state:
                st.session_state[key] = list(value) if isinstance(value, list) else value

        updated = str(catalog.latest_updated_at)[:19] if catalog.latest_updated_at else "—"
        st.caption(f"{tr('数据源', 'Source')}: aerial_db · aerial.videos  |  "
                   f"{tr('库内最近更新', 'Latest database update')}: {updated}  |  "
                   f"{tr('每 60 秒自动刷新', 'Refreshes every 60 seconds')}")
        if not catalog.records:
            st.info(tr("当前数据库没有有效视频记录。", "The database has no active video records."))

        with st.expander(tr("筛选条件", "Filters"), expanded=True):
            # Retain removed selections as available options. Clearing them would
            # turn a deliberately narrow filter into the entire new population.
            city_options = sorted({record.city for record in catalog.records} |
                                  set(st.session_state.get("pg_cities", ())))
            codec_options = sorted({record.codec or _UNKNOWN for record in catalog.records} |
                                   set(st.session_state.get("pg_codecs", ())))
            period_options = sorted({record.period for record in catalog.records} |
                                    set(st.session_state.get("pg_periods", ())))
            left, right = st.columns(2)
            with left:
                st.multiselect(tr("城市", "City"), city_options, key="pg_cities", format_func=category)
                dates = [record.collected_on for record in catalog.records if record.collected_on]
                if dates:
                    bounds = (min(dates), max(dates))
                    previous_bounds = st.session_state.get("_pg_date_bounds")
                    selected_dates = st.session_state.get("pg_date_range")
                    if selected_dates is None or tuple(selected_dates) == previous_bounds:
                        st.session_state["pg_date_range"] = bounds
                    st.session_state["_pg_date_bounds"] = bounds
                if dates or "pg_date_range" in st.session_state:
                    # The default 'today' reads an existing range from state;
                    # this avoids competing widget-default and state values.
                    st.date_input(tr("采集日期范围", "Collection date range"),
                                  key="pg_date_range")
                else:
                    st.caption(tr("采集日期均未知。", "All collection dates are unknown."))
                if "pg_include_unknown_dates" not in st.session_state:
                    st.session_state["pg_include_unknown_dates"] = True
                st.checkbox(tr("包含未知采集日期", "Include unknown collection dates"),
                            key="pg_include_unknown_dates")
            with right:
                st.multiselect(tr("编码格式", "Codec"), codec_options, key="pg_codecs", format_func=category)
                st.multiselect(tr("采集时段", "Collection period"), period_options,
                               key="pg_periods", format_func=category)
                st.caption(tr("时段按北京时间分组；每个图表与导出均应用以上筛选。",
                              "Periods use China Standard Time. These filters apply to every chart and export."))

        st.session_state["_pg_filter_snapshot"] = {
            key: list(st.session_state[key]) if isinstance(st.session_state[key], list)
            else st.session_state[key]
            for key in _FILTER_KEYS if key in st.session_state
        }
        records = selected_records(catalog, st.session_state)
        snapshot = summarize_records(records, catalog.latest_updated_at)
        city_count = len({record.city for record in records if record.gps_status == "已归属城市"})
        duration_known = sum(record.duration_seconds is not None for record in records)
        size_known = sum(record.size_bytes is not None for record in records)
        duration_display = f"{snapshot.duration_hours:,.1f}" if not records or duration_known else "—"
        size_display = f"{snapshot.size_tb:,.2f}" if not records or size_known else "—"
        gps_records = tuple(record for record in records if record.gps_status != "缺少坐标")
        gps_hours = sum(record.duration_seconds / 3600 for record in gps_records if record.duration_seconds is not None)
        gps_display = f"{gps_hours:,.2f}" if not gps_records or any(record.duration_seconds is not None for record in gps_records) else '—'
        metrics = (
            (duration_display, tr("已知时长合计（h）", "Known duration total (h)")),
            (size_display, tr("已知容量合计（TB）", "Known size total (TB)")),
            (str(city_count), tr("覆盖城市", "Cities")),
            (gps_display, tr("含坐标视频时长（h）", "Geolocated video duration (h)")),
        )
        for index, (column, (value, label)) in enumerate(zip(st.columns(4), metrics)):
            with column:
                style = ("kpi-blue", "kpi-indigo", "kpi-teal", "kpi-purple")[index % 4]
                st.markdown(f"<div class='kpi-card {style}'><div class='kpi-num'>{value}</div>"
                            f"<div class='kpi-lbl'>{html.escape(label)}</div></div>", unsafe_allow_html=True)
        if not records:
            st.info(tr("当前筛选没有匹配的视频，请调整条件。", "No videos match these filters. Adjust the selections."))
            heading("nav-odd", "元数据与标注状态", "Metadata and annotation status")
            unavailable_annotations()
            return

        heading("nav-city", "城市与采集分布", "Cities and collection distribution",
                "城市按采集坐标与行政区边界归属；地图点为城市行政区代表点，不表示视频采集位置。",
                "Cities are assigned from collection GPS and administrative boundaries. Map markers represent city administrative areas, not video collection positions.")
        city_counts = duration_hours_by(records, lambda record: record.city)
        centers = {center.city: center for center in catalog.city_centers}
        mapped_cities = sorted(set(city_counts) & set(centers))
        left, right = st.columns(2)
        with left:
            if mapped_cities:
                figure = go.Figure(go.Scattermap(
                    lon=[centers[city].longitude for city in mapped_cities],
                    lat=[centers[city].latitude for city in mapped_cities],
                    text=[f"{category(city)} · " + (f"{city_counts[city]:.3f} h" if city_counts[city] is not None else tr("时长未知", "Duration unknown")) for city in mapped_cities],
                    customdata=[city_counts[city] for city in mapped_cities],
                    mode="markers", marker=dict(size=14, color=DATA_BLUE),
                    hovertemplate="%{text}<extra></extra>",
                ))
                style_figure(figure, height=340)
                figure.update_layout(map=dict(style="carto-positron", center=dict(lat=36, lon=113), zoom=3),
                                     margin=dict(l=0, r=0, t=0, b=0), height=340)
                st.plotly_chart(figure, width="stretch")
            else:
                st.info(tr("当前筛选没有可归属城市的记录。", "No selected videos can be assigned to a city."))
        with right:
            bar(city_counts, "城市视频时长（h）", "Video duration by city (h)")
        left, right = st.columns(2)
        with left:
            months = duration_hours_by(records, lambda record: record.collected_on.strftime("%Y-%m") if record.collected_on else _UNKNOWN)
            bar(months, "采集月份时长（h）", "Duration by collection month (h)", order=sorted(months))
        with right:
            bar(duration_hours_by(records, lambda record: record.period), "采集时段分布（h）", "Collection period (h)",
                order=("早高峰 (07-09)", "日常规", "晚高峰 (17-19)", "夜间 (19-07)", _UNKNOWN))

        heading("nav-params", "视频参数分布", "Video parameters")
        for left_spec, right_spec in (
            ((lambda record: record.codec or _UNKNOWN, "编码格式分布", "Codec distribution"),
             (_resolution, "分辨率分布", "Resolution distribution")),
            ((_fps, "帧率分布", "Frame rate distribution"),
             (_duration_range, "时长区间分布", "Duration distribution")),
        ):
            for column, (key, zh, en) in zip(st.columns(2), (left_spec, right_spec)):
                with column:
                    order = ("< 1 min", "1–5 min", "5–15 min", "15–30 min", "30–60 min", "≥ 60 min", _UNKNOWN) if key is _duration_range else None
                    bar(duration_hours_by(records, key), zh + "（h）", en + " (h)", order=order)

        heading("nav-odd", "元数据与标注状态", "Metadata and annotation status",
                "所有分布按已知视频时长（h）汇总；缺失字段独立显示，未知时长不推算，空值与测得零分开。",
                "Distributions sum known video duration in hours. Missing metadata remains separate; unknown durations are not estimated and blank is distinct from measured zero.")
        quality_specs = (
            ("采集日期", "Collection date", lambda record: record.collected_on is not None),
            ("城市归属", "City attribution", lambda record: record.gps_status == "已归属城市"),
            ("编码", "Codec", lambda record: record.codec not in (None, "", _UNKNOWN)),
            ("分辨率", "Resolution", lambda record: bool(record.width and record.height)),
            ("帧率", "Frame rate", lambda record: record.fps is not None),
            ("时长", "Duration", lambda record: record.duration_seconds is not None),
            ("容量", "Size", lambda record: record.size_bytes is not None),
            ("入库日期", "Ingestion date", lambda record: record.added_on is not None),
        )
        labels, present, absent = [], [], []
        for zh, en, predicate in quality_specs:
            labels.append(tr(zh, en))
            hours = duration_hours_by(records, lambda record: 'known' if predicate(record) else 'unknown')
            present.append(hours.get('known', 0.0))
            absent.append(hours.get('unknown', 0.0))
        figure = go.Figure((
            go.Bar(name=tr("已知 / 有效", "Known / valid"), x=labels, y=present, marker_color=DATA_BLUE),
            go.Bar(name=tr("未知 / 无效", "Unknown / invalid"), x=labels,
                   y=absent, marker_color="#dce5ec"),
        ))
        style_figure(figure, height=320)
        figure.update_layout(barmode="stack", margin=dict(t=50))
        figure.update_yaxes(title='h', rangemode="tozero", tickformat=",.2f")
        figure.update_traces(hovertemplate='%{x}: %{y:.3f} h<extra>%{fullData.name}</extra>')
        st.plotly_chart(figure, width="stretch")
        left, right = st.columns(2)
        with left:
            bar(duration_hours_by(records, lambda record: record.gps_status), "坐标归属时长（h）", "GPS attribution duration (h)")
        with right:
            additions = duration_hours_by(records, lambda record: record.added_on.strftime("%Y-%m") if record.added_on else _UNKNOWN)
            bar(additions, "入库月份时长（h）", "Duration by ingestion month (h)", order=sorted(additions))
        unavailable_annotations()

        st.subheader(tr("当前筛选汇总", "Filtered aggregate data"))
        st.caption(tr("时长单位 h；空值表示该组时长未知，部分未知状态表示仅合计已知部分；测得零保留为零。",
                      "Duration is in h. Blank means unknown; partial means the known portion only. A measured zero remains zero."))
        exported = public_aggregate_rows(records)
        exported['duration_range'] = exported.duration_range.map(category)
        exported['duration_status'] = exported.duration_status.map(category)
        exported['size_status'] = exported.size_status.map(category)
        st.dataframe(exported, hide_index=True, width="stretch", column_config={
            "city": tr("城市", "City"), "collection_month": tr("采集月份", "Collection month"),
            "period": tr("时段", "Period"), "codec": tr("编码", "Codec"),
            "resolution": tr("分辨率", "Resolution"), "fps": tr("帧率", "FPS"),
            "duration_range": tr("时长区间", "Duration range"),
            "duration_hours": tr("时长（h）", "Duration (h)"), "size_tb": tr("容量（TB）", "Size (TB)"),
            "duration_status": tr("时长完整度", "Duration completeness"),
            "size_status": tr("容量完整度", "Size completeness"),
        })
        st.download_button(tr("下载汇总 CSV", "Download aggregate CSV"),
                           data=exported.to_csv(index=False).encode("utf-8-sig"),
                           file_name="aerial-videos-statistics.csv", mime="text/csv", key="pg_download")

    render_current_catalog()