from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import datetime from urllib.parse import urlencode from ..security import admin_csrf_token from ..settings import DdnsSettings from ..store import DdnsStore from .html_utils import SafeHtml, attr, escape, trusted from .layout import PageOptions, _page, _top_nav from .templates_admin_tables import ( _account_row, _admin_table_pagination, _admin_table_toolbar, _event_row, ) @dataclass(frozen=True) class _CleanupSummary: last_run: object next_run: object status: object last_error: object interval_seconds: object accounts_removed: object @dataclass(frozen=True) class AdminPageContext: accounts: Sequence[Mapping[str, object]] cleanup_runs: Sequence[Mapping[str, object]] scheduled_tasks: Sequence[Mapping[str, object]] update_events: Sequence[Mapping[str, object]] service_metrics: dict[str, float] table_counts: dict[str, int] table_offsets: dict[str, int] table_limit: int notice_html: SafeHtml | None def _admin_page_context( store: DdnsStore, table_offsets: dict[str, int], table_limit: int, notice_html: SafeHtml | None, ) -> AdminPageContext: return AdminPageContext( accounts=store.list_accounts(limit=table_limit, offset=table_offsets.get("accounts", 0)), cleanup_runs=store.list_cleanup_runs(limit=table_limit, offset=table_offsets.get("cleanup_runs", 0)), scheduled_tasks=store.list_scheduled_tasks(), update_events=store.list_update_events(table_limit, offset=table_offsets.get("update_events", 0)), service_metrics=store.get_service_metrics(), table_counts={ "accounts": store.count_accounts(), "cleanup_runs": store.count_cleanup_runs(), "update_events": store.count_update_events(), }, table_offsets=table_offsets, table_limit=table_limit, notice_html=notice_html, ) def _admin_secret_panel(title: str, rows: list[SafeHtml]) -> SafeHtml: return trusted(f"""

Admin action

{escape(title)}

Copy these values now. Secret values are only shown once.

{"".join(rows)}
""") def _render_admin_page( settings: DdnsSettings, context: AdminPageContext, ) -> str: csrf = attr(admin_csrf_token(settings)) admin_path = settings.admin_path safe_admin_path = attr(admin_path) rows = _account_rows(context.accounts, csrf=csrf, admin_path=admin_path) event_rows = _event_rows(context.update_events) accounts_count = escape(context.table_counts.get("accounts", len(context.accounts))) event_count = escape(context.table_counts.get("update_events", len(context.update_events))) metrics_html = _metrics_panel(context.service_metrics) account_pager = _server_page_links( admin_path, "accounts_offset", context.table_offsets.get("accounts", 0), context.table_limit, context.table_counts.get("accounts", 0), ) event_pager = _server_page_links( admin_path, "update_events_offset", context.table_offsets.get("update_events", 0), context.table_limit, context.table_counts.get("update_events", 0), ) cleanup_panel = _cleanup_panel(settings, context.cleanup_runs, context.scheduled_tasks, safe_admin_path, csrf) totals_html = _totals_strip(context.table_counts, context.service_metrics) title_section = _admin_title_section(totals_html) status_section = _admin_status_section(cleanup_panel, metrics_html) hostnames_section = _admin_hostnames_section(context, rows, accounts_count, account_pager) events_section = _admin_events_section(context, event_rows, event_count, event_pager) return _page( "KarlDNS Admin", trusted(f"""
{_top_nav("Operator")} {title_section} {context.notice_html or ""} {status_section} {hostnames_section} {events_section}
"""), PageOptions( extra_styles=trusted(''), extra_scripts=trusted(''), ), ) def _admin_title_section(totals_html: SafeHtml) -> SafeHtml: return trusted(f"""

KarlDNS

Operator console

{totals_html}
""") def _admin_status_section(cleanup_panel: SafeHtml, metrics_html: SafeHtml) -> SafeHtml: return trusted(f"""
{cleanup_panel} {metrics_html}
""") def _admin_hostnames_section( context: AdminPageContext, rows: SafeHtml, accounts_count: str, account_pager: SafeHtml, ) -> SafeHtml: return trusted(f"""

Hostnames

{escape(len(context.accounts))} of {accounts_count} shown.

{_admin_table_toolbar("hostnames", "Filter visible hostnames")}
{rows}
Hostname Status IPv4 IPv6 Updated Links Actions
{_admin_table_pagination()} {account_pager}
""") def _admin_events_section( context: AdminPageContext, event_rows: SafeHtml, event_count: str, event_pager: SafeHtml, ) -> SafeHtml: return trusted(f"""

Update events

{escape(len(context.update_events))} of {event_count} shown. Unchanged checks are grouped.

{_admin_table_toolbar("update-events", "Filter visible update events")}
{event_rows}
Time Hostname Status IPv4 IPv6 Count Detail
{_admin_table_pagination()} {event_pager}
""") def _account_rows(accounts: Sequence[Mapping[str, object]], *, csrf: str, admin_path: str) -> SafeHtml: return trusted("\n".join(_account_row(account, csrf=csrf, admin_path=admin_path) for account in accounts) or """
No hostnames yet.
Generated hostnames appear here after users create or register them.
""") def _event_rows(events: Sequence[Mapping[str, object]]) -> SafeHtml: return trusted("\n".join(_event_row(event) for event in events) or """
No update events yet.
Router update activity and DNS publish results will appear here.
""") def _totals_strip(table_counts: dict[str, int], metrics: dict[str, float]) -> SafeHtml: items = [ ("Hostnames", table_counts.get("accounts", 0)), ("Events", table_counts.get("update_events", 0)), ("DNS failures", int(metrics.get("dns_failures_total", 0))), ("Blocked", int(metrics.get("rate_limit_hits_total", 0))), ] return trusted( '
' + "".join(f"{escape(value)}{escape(label)}" for label, value in items) + "
" ) def _admin_notice(kind: str, message: str) -> SafeHtml: alert_class = "alert-success" if kind == "success" else "alert-info" return trusted(f"""
{escape(message)}
""") def _server_page_links(admin_path: str, offset_name: str, offset: int, limit: int, total: int) -> SafeHtml: if total <= limit: return trusted("") previous_offset = max(0, offset - limit) next_offset = offset + limit previous_disabled = " disabled" if offset <= 0 else "" next_disabled = " disabled" if next_offset >= total else "" previous_href = attr(f"{admin_path}?{urlencode({offset_name: previous_offset})}") next_href = attr(f"{admin_path}?{urlencode({offset_name: next_offset})}") start = min(offset + 1, total) end = min(offset + limit, total) return trusted(f""" """) def _cleanup_panel( settings: DdnsSettings, cleanup_runs: Sequence[Mapping[str, object]], scheduled_tasks: Sequence[Mapping[str, object]], safe_admin_path: str, csrf: str, ) -> SafeHtml: summary = _cleanup_summary( cleanup_runs=cleanup_runs, scheduled_tasks=scheduled_tasks, fallback_interval_seconds=settings.cleanup_interval_seconds, ) error_html = ( trusted(f'

Last error: {escape(summary.last_error)}

') if summary.last_error else trusted("") ) last_run_title = attr(str(summary.last_run or "Never")) next_run_title = attr(str(summary.next_run or "Pending")) last_run_display = escape(_format_timestamp(summary.last_run, "Never")) next_run_display = escape(_format_timestamp(summary.next_run, "Pending")) return trusted(f"""

Cleanup

Every {escape(_format_interval(summary.interval_seconds))}
Last run
{last_run_display}
Next run
{next_run_display}
Status
{escape(str(summary.status))}
Removed
{escape(summary.accounts_removed)}
{error_html}
""") def _cleanup_task(scheduled_tasks: Sequence[Mapping[str, object]]) -> Mapping[str, object]: return next( (item for item in scheduled_tasks if item.get("name") == "cleanup"), scheduled_tasks[0] if scheduled_tasks else {}, ) def _cleanup_summary( *, cleanup_runs: Sequence[Mapping[str, object]], scheduled_tasks: Sequence[Mapping[str, object]], fallback_interval_seconds: int, ) -> _CleanupSummary: task = _cleanup_task(scheduled_tasks) last_cleanup = cleanup_runs[0] if cleanup_runs else {} return _CleanupSummary( last_run=_cleanup_last_run(task, last_cleanup), next_run=task.get("next_run_at"), status=task.get("last_status") or "pending", last_error=task.get("last_error") or "", interval_seconds=task.get("interval_seconds") or fallback_interval_seconds, accounts_removed=_cleanup_accounts_removed(last_cleanup), ) def _cleanup_last_run(task: Mapping[str, object], last_cleanup: Mapping[str, object]) -> object: return task.get("last_finished_at") or last_cleanup.get("created_at") def _cleanup_accounts_removed(last_cleanup: Mapping[str, object]) -> object: return last_cleanup.get("unused_accounts_deleted", 0) def _metrics_panel(metrics: dict[str, float]) -> SafeHtml: samples = metrics.get("update_latency_samples_total", 0) average_latency = metrics.get("update_latency_ms_total", 0) / samples if samples else 0 items = [ ("Updates", _metric_int(metrics.get("update_requests_total", 0))), ("Updated", _metric_int(metrics.get("update_success_total", 0))), ("DNS failures", _metric_int(metrics.get("dns_failures_total", 0))), ("DNS check failures", _metric_int(metrics.get("dns_reconcile_failures_total", 0))), ("Blocked requests", _metric_int(metrics.get("rate_limit_hits_total", 0))), ("Average update time", f"{average_latency:.1f} ms"), ("Last update time", f"{metrics.get('last_update_latency_ms', 0):.1f} ms"), ( "Cleanup ok/errors", ( f"{_metric_int(metrics.get('cleanup_success_total', 0))}/" f"{_metric_int(metrics.get('cleanup_errors_total', 0))}" ), ), ] cards = "\n".join(f""" {escape(value)}{escape(label)} """ for label, value in items) return trusted(f"""
{cards}
""") def _metric_int(value: object) -> str: try: return str(int(float(str(value)))) except ValueError: return "0" def _format_interval(seconds: object) -> str: try: value = int(str(seconds)) except (TypeError, ValueError): return str(seconds) if value % 3600 == 0: hours = value // 3600 return f"{hours}h" if value % 60 == 0: minutes = value // 60 return f"{minutes}m" return f"{value}s" def _format_timestamp(value: object, fallback: str = "-") -> str: if not value: return fallback raw = str(value) try: parsed = datetime.fromisoformat(raw) except ValueError: return raw return parsed.strftime("%Y-%m-%d %H:%M")