"""All active collection batches and database ingestion updates."""
from __future__ import annotations

from collections import Counter
from datetime import date, datetime, timedelta
import html
import math
from typing import Callable, Iterable
from zoneinfo import ZoneInfo

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

from batch_updates import BatchUpdateCatalog, BatchUpdateRecord
from city_labels import city_display_name
from dashboard_charts import DATA_BLUE, style_figure

_BEIJING = ZoneInfo('Asia/Shanghai')


def ingestion_totals(catalog: BatchUpdateCatalog, today: date) -> tuple[int, int, int]:
    """Today and inclusive last 7/30 Beijing days, excluding future entries."""
    counts = Counter()
    for record in catalog.records:
        for item in record.ingestion_days:
            counts[item.day] += item.videos
    return tuple(sum(count for day, count in counts.items()
                     if today - timedelta(days=window-1) <= day <= today)
                 for window in (1, 7, 30))


def ingestion_timeline(catalog: BatchUpdateCatalog, today: date,
                       window: int | None = 30) -> list[tuple[date, int]]:
    """Fill Beijing calendar days through today; future rows never add activity."""
    counts = Counter()
    for record in catalog.records:
        for item in record.ingestion_days:
            if item.day <= today:
                counts[item.day] += item.videos
    if window is None:
        if not counts:
            return []
        start = min(counts)
    else:
        start = today - timedelta(days=window-1)
    return [(start + timedelta(days=offset), counts[start + timedelta(days=offset)])
            for offset in range((today-start).days+1)]


def _hour_days(catalog: BatchUpdateCatalog, today: date) -> dict[date, tuple[float, int, int]] | None:
    if not catalog.includes_durations:
        return None
    buckets = {}
    for record in catalog.records:
        for item in record.ingestion_days:
            if item.day <= today:
                bucket = buckets.setdefault(item.day, [[], 0, 0])
                bucket[0].append(item.known_seconds)
                bucket[1] += item.known_videos
                bucket[2] += item.unknown_videos
    return {day: (math.fsum(values[0]), values[1], values[2]) for day, values in buckets.items()}


def ingestion_hours(catalog: BatchUpdateCatalog, today: date) -> tuple[float | None, ...]:
    """Known active-video hours received in Beijing day windows, never frame estimates."""
    days = _hour_days(catalog, today)
    if days is None:
        return None, None, None
    totals = []
    for window in (1, 7, 30):
        selected = [values for day, values in days.items()
                    if today - timedelta(days=window-1) <= day <= today]
        known, unknown = sum(row[1] for row in selected), sum(row[2] for row in selected)
        totals.append(math.fsum(row[0] for row in selected) / 3600 if known or not unknown else None)
    return tuple(totals)


def ingestion_hour_timeline(catalog: BatchUpdateCatalog, today: date,
                            window: int | None = 30) -> list[tuple[date, float | None]] | None:
    """No arrivals is measured zero; all-unknown arrivals leave an explicit gap."""
    days = _hour_days(catalog, today)
    if days is None:
        return None
    if window is None and not days:
        return []
    start = min(days) if window is None else today - timedelta(days=window-1)
    rows = []
    for offset in range((today-start).days+1):
        day = start + timedelta(days=offset)
        seconds, known, unknown = days.get(day, (0, 0, 0))
        rows.append((day, seconds / 3600 if known or not unknown else None))
    return rows


def batch_table(records: Iterable[BatchUpdateRecord], translate: Callable) -> pd.DataFrame:
    """Export only curated batch fields, with public timestamps in Beijing time."""
    def stamp(value):
        return value.astimezone(_BEIJING).strftime('%Y-%m-%d %H:%M:%S') if value else translate('未知')
    records = sorted(records, key=lambda record: (record.last_ingested_at is None,
                     -record.last_ingested_at.timestamp() if record.last_ingested_at else 0, record.code))
    columns = tuple(map(translate, ('批次编号','城市','已知时长 (h)','首次入库','最近入库',
                                    '源数据更新','标注完成','可用标签')))
    return pd.DataFrame([
        (record.code, city_display_name(record.city), record.duration_hours, stamp(record.first_ingested_at),
         stamp(record.last_ingested_at), stamp(record.source_updated_at),
         stamp(record.annotation_completed_at), translate('有' if record.annotations_available else '暂无'))
        for record in records
    ], columns=columns)


def render_batch_updates(load_updates: Callable, translate: Callable) -> None:
    def tr(zh, en):
        translated = translate(zh)
        return en if st.session_state.get('lang') == 'en' and translated == zh else translated

    names = {'批次编号':'Batch code','城市':'City','已知时长 (h)':'Known duration (h)',
             '首次入库':'First ingested','最近入库':'Last ingested','源数据更新':'Source updated',
             '标注完成':'Annotation completed','可用标签':'Usable labels','有':'Available',
             '暂无':'Unavailable','未知':'Unknown'}
    def table_translate(value):
        return tr(value, names.get(value, value))

    st.markdown("<div id='nav-updates' class='dash-section'><h2 class='dash-section-title'>"
                + html.escape(tr('数据更新', 'Dataset updates')) + '</h2></div>', unsafe_allow_html=True)
    st.caption(tr(
        '这里展示全部活跃采集批次，独立于城市、采集日期与 ODD 筛选。入库时间是数据库接收视频的时间，'
        '不等于相机采集时间；新增量按当前活跃视频的首次入库日计算，不是包含删除记录的历史审计。'
        '时间统一为北京时间；标注完成取实际处理结束时间，源数据更新取视频修改时间。',
        'All active collection batches are shown independently of city, collection-date and ODD filters. '
        'Ingestion means database receipt, not camera collection. Additions use the creation dates of currently '
        'active videos; this is not a deletion audit. Times use Beijing time. Annotation completion uses the '
        'actual processing finish time; source updates use video modification time.'))
    catalog, status = load_updates()
    if status != 'ok' or catalog is None:
        st.warning(tr('数据更新暂不可用，请稍后刷新。', 'Dataset updates are unavailable. Please refresh later.'))
        return
    today = datetime.now(_BEIJING).date()
    today_hours, seven_hours, month_hours = ingestion_hours(catalog, today)
    with st.container(key='batch_updates_kpis'):
        for column, label, english, value in zip(st.columns(4),
                ('活跃采集批次','今日新增 (h)','近 7 日新增 (h)','近 30 日新增 (h)'),
                ('Active batches','Added today (h)','Added in 7 days (h)','Added in 30 days (h)'),
                (len(catalog.records), today_hours, seven_hours, month_hours)):
            column.metric(tr(label, english), (f'{value:,}' if label == '活跃采集批次'
                                               else f'{value:,.3f}' if value is not None else '—'))
    if not catalog.includes_durations:
        st.info(tr('批次及每日入库时长尚未发布；暂不显示时长趋势，不以视频数量换算。',
                   'Batch and daily ingestion durations have not been published. The duration trend is unavailable.'))
    elif any(record.unknown_videos for record in catalog.records):
        st.caption(tr('含未知时长数据，仅合计实际已知时长；全未知显示空值，测得零时长保留为 0 h。',
                      'Some durations are unknown. Only measured duration is summed; all-unknown values are blank, measured zero remains 0 h.'))
    if any(sum(day.videos for day in record.ingestion_days) < record.videos for record in catalog.records):
        st.caption(tr('存在入库日期缺失的数据；每日趋势和近期新增指标仅统计有入库日期的已知时长。',
                      'Some ingestion dates are missing. Daily and recent additions include only measured duration with an ingestion date.'))
    if catalog.unlinked_videos:
        hours = catalog.unlinked_duration_hours
        value = f'{hours:,.3f} h' if hours is not None else tr('时长未知或尚未发布', 'duration unknown or unpublished')
        st.info(tr(f'另有未关联采集批次的数据（已知部分：{value}），未计入批次新增量与下表。',
                   f'Unlinked data (known portion: {value}) is excluded from batch additions and the table.'))
    if not catalog.records:
        st.info(tr('暂无活跃采集批次。', 'No active collection batches are available.'))
        return

    ranges = (tr('近 7 日', 'Last 7 days'), tr('近 30 日', 'Last 30 days'), tr('全部', 'All time'))
    selected_range = st.selectbox(tr('入库时间范围', 'Ingestion time range'), ranges, index=1,
                                  key='batch_updates_range')
    window = {ranges[0]:7, ranges[1]:30, ranges[2]:None}[selected_range]
    timeline = ingestion_hour_timeline(catalog, today, window)
    if timeline:
        figure = go.Figure(go.Bar(x=[day.isoformat() for day, _ in timeline],
                                 y=[count for _, count in timeline], marker_color=DATA_BLUE,
                                 hovertemplate='%{x}: %{y:.3f} h<extra></extra>'))
        style_figure(figure, title=tr('每日新增时长 (h) · 当前活跃数据', 'Daily ingested duration (h) · currently active data'),
                     height=300)
        figure.update_xaxes(title=tr('首次入库日（北京时间）', 'First ingestion day (Beijing)'), type='date')
        figure.update_yaxes(title='h', rangemode='tozero', tickformat='.3~f')
        st.plotly_chart(figure, width='stretch', key='batch_updates_timeline')
    elif timeline is not None:
        st.info(tr('当前批次尚无可用入库日期。', 'No ingestion dates are available for these batches.'))

    left, right = st.columns(2)
    query = left.text_input(tr('搜索批次编号或城市', 'Search batch code or city'), key='batch_updates_search').strip().casefold()
    cities = right.multiselect(tr('更新表城市', 'Batch table cities'), sorted({r.city for r in catalog.records}),
                               key='batch_updates_cities', format_func=city_display_name)
    selected = tuple(record for record in catalog.records
                     if (not cities or record.city in cities)
                     and (not query or query in record.code.casefold() or query in record.city.casefold()
                          or query in city_display_name(record.city).casefold()))
    st.caption(tr(f'显示 {len(selected):,} / {len(catalog.records):,} 个批次；CSV 始终包含全部活跃批次。',
                  f'Showing {len(selected):,} / {len(catalog.records):,} batches; CSV always includes all active batches.'))
    st.dataframe(batch_table(selected, table_translate), hide_index=True, width='stretch')
    st.download_button(tr('下载全部批次 CSV', 'Download all batches CSV'),
                       batch_table(catalog.records, table_translate).to_csv(index=False).encode('utf-8-sig'),
                       file_name='aerial-batch-updates.csv', mime='text/csv', key='batch_updates_csv')