src/ui/admin_tab.py

"""Admin status tab (spec section 12). Callbacks re-check the username; hiding is not enough."""

from __future__ import annotations

import logging

import gradio as gr

from src.services.access import is_admin
from src.services.flags import RuntimeFlags
from src.services.limits import LimitService
from src.services.session import VisitorSession
from src.ui.components import ApiVisibility, Localizer

REFRESH_SECONDS = 5
logger = logging.getLogger(__name__)


def build_admin_tab(
    *,
    demo: gr.Blocks,
    limits: LimitService,
    flags: RuntimeFlags,
    photo_studio: bool,
    high_quality: bool,
    loc: Localizer,
    session: gr.State,
    api_visibility: ApiVisibility,
) -> gr.Tab:
    t = loc.t
    visibility = {"api_visibility": api_visibility, "show_progress": "hidden"}

    with loc.make(
        gr.Tab, lambda lang: {"label": t(lang, "tab.admin")}, id="admin", visible=False
    ) as tab:
        summary = gr.Markdown()
        paused = loc.make(gr.Checkbox, lambda lang: {"label": t(lang, "admin.pause")})
        photo_open = (
            loc.make(gr.Checkbox, lambda lang: {"label": t(lang, "admin.photo")})
            if photo_studio
            else None
        )
        quality_on = (
            loc.make(gr.Checkbox, lambda lang: {"label": t(lang, "admin.quality")})
            if high_quality
            else None
        )
        ask_reset = loc.make(gr.Button, lambda lang: {"value": t(lang, "admin.reset")})
        with gr.Column(visible=False) as confirm:
            loc.make(gr.Markdown, lambda lang: {"value": t(lang, "admin.reset_confirm")})
            with gr.Row():
                yes = loc.make(
                    gr.Button, lambda lang: {"value": t(lang, "admin.reset_yes")}, variant="stop"
                )
                no = loc.make(gr.Button, lambda lang: {"value": t(lang, "admin.reset_cancel")})
    timer = gr.Timer(REFRESH_SECONDS, active=False)

    def render(lang: str) -> str:
        snap = limits.snapshot()
        latency = ", ".join(f"{m} {s:.1f} s" for m, s in sorted(snap.p50_seconds.items()))
        return "\n\n".join(
            [
                t(lang, "admin.paused" if snap.paused else "admin.running"),
                t(
                    lang,
                    "admin.usage",
                    used=snap.images_used,
                    max_images=snap.max_images,
                    spend=f"{snap.spend_usd:.2f}",
                    max_cost=f"{snap.max_cost_usd:.2f}",
                ),
                t(lang, "admin.activity", in_flight=snap.in_flight, **snap.recent),
                t(lang, "admin.latency", details=latency or t(lang, "admin.latency_none")),
            ]
        )

    def flag_state() -> list:
        state = []
        if photo_open is not None:
            state.append(flags.photo_studio)
        if quality_on is not None:
            state.append(flags.high_quality)
        return state

    def on_load(current: VisitorSession, request: gr.Request):
        if not is_admin(request.username):
            return [gr.skip()] * (4 + len(flag_state()))
        return (
            gr.update(visible=True),
            gr.Timer(active=True),
            render(current.lang),
            limits.paused,
            *flag_state(),
        )

    def on_refresh(current: VisitorSession, request: gr.Request):
        if not is_admin(request.username):
            return [gr.skip()] * (2 + len(flag_state()))
        return render(current.lang), limits.paused, *flag_state()

    def on_photo(value: bool, request: gr.Request):
        if is_admin(request.username):
            flags.set_photo_studio(bool(value))
            logger.info("admin action: photo studio %s", "opened" if value else "closed")

    def on_quality(value: bool, request: gr.Request):
        if is_admin(request.username):
            flags.set_high_quality(bool(value))
            logger.info("admin action: high-quality model %s", "on" if value else "off")

    def on_pause(value: bool, current: VisitorSession, request: gr.Request):
        if not is_admin(request.username):
            return gr.skip()
        limits.set_paused(bool(value))
        return render(current.lang)

    def on_confirm_reset(current: VisitorSession, request: gr.Request):
        if is_admin(request.username):
            limits.reset_counters()
            return gr.update(visible=False), render(current.lang)
        return gr.skip(), gr.skip()

    extra = [box for box in (photo_open, quality_on) if box is not None]
    demo.load(on_load, [session], [tab, timer, summary, paused, *extra], **visibility)
    timer.tick(on_refresh, [session], [summary, paused, *extra], **visibility)
    paused.input(on_pause, [paused, session], [summary], **visibility)
    if photo_open is not None:
        photo_open.input(on_photo, [photo_open], None, **visibility)
    if quality_on is not None:
        quality_on.input(on_quality, [quality_on], None, **visibility)
    ask_reset.click(lambda: gr.update(visible=True), None, [confirm], **visibility)
    no.click(lambda: gr.update(visible=False), None, [confirm], **visibility)
    yes.click(on_confirm_reset, [session], [confirm, summary], **visibility)
    return tab