"""Server-rendered pages for the central Blueforce panel (Python stdlib only). Rendering happens on the server through ``string.Template`` from the standard library. No template engine is imported: ``jinja2`` is not installable on an air-gapped fleet (PRODUCT.md, "Stack"). Every page is plain HTML; the palette lives in the local stylesheets and this module never writes a colour, a webfont or a URL into the markup -- ``tests/test_web_static.py`` enforces that. Layout contract (docs/DESIGN.md, FIRST VIEWPORT): the dark header band carries the product name, the target identity and the privilege level on the right; then the data-source banner, the summary strip with the one large figure, the fleet list (a state badge per row, non-ready first, filter bar, selected row in the accent-soft fill) and the selected device's detail region with the channel verification list. Colour is never the only carrier of meaning: every badge carries its Turkish word, so the page survives greyscale. """ from __future__ import annotations import html from dataclasses import dataclass from pathlib import Path from string import Template from urllib.parse import urlencode from typing import Any, Mapping, Sequence from . import PANEL_NAME, PANEL_SURFACE, PANEL_VERSION from . import fleet_data from .fleet_data import ( CHANNELS, CHANNEL_STATUS_WORDS, STATE_VOCABULARY, WAVES, Device, FleetFilter, FleetSnapshot, Wave, ) TEMPLATES_DIR = Path(__file__).resolve().parent / "templates" STATIC_DIR = Path(__file__).resolve().parent / "static" FLEET_PATH = "/" ABOUT_PATH = "/hakkinda" # Short Turkish note per state, for the quiet part of the summary strip. The # long meaning lives on the about page and in the badge tooltip. STATE_SHORT_NOTES: dict[str, str] = { "KURULDU": "kuruldu, kayıt bekliyor", "KAYITLI": "kanallar eksik", "HAZIR": "sevk edilebilir", "HATA": "bir insan gerekiyor", "YOK": "hiç görülmedi", } # The panel's own rules, shown to the operator instead of being a comment. PANEL_RULES: tuple[str, ...] = ( "Panel yalnız okuma yapar: hiçbir uç nokta komut çalıştırmaz, dosya yazmaz ve cihazda değişiklik yapmaz.", "Dış kaynak yok: yazı tipi, CDN veya internet erişimi kullanılmaz; stil dosyaları yereldir (air-gapped filo).", "Durum uydurulmaz: «HAZIR» iddiası dört kanal doğrulanmadıkça reddedilir ve cihaz HATA olur (docs/27).", "Kanal denetimi başarısız olduysa cihaz HATA sayılır: cihaz bir insan bekliyor demektir.", "Panel hiçbir durumu yükseltmez; HAZIR kararı merkezindir (KAYITLI -> HAZIR).", "Kimliği kurulamayan kayıt sessizce düşürülmez; «Geçersiz kayıtlar» altında nedeniyle görünür (docs/02).", "Envanter dosyası okunamazsa panel demo veriye düşmez; «VERİ YOK» der ve filo durumunu bilinmiyor olarak bırakır.", ) _TEMPLATE_CACHE: dict[str, Template] = {} def template(name: str) -> Template: """Load and cache one local template by file name.""" cached = _TEMPLATE_CACHE.get(name) if cached is None: cached = Template((TEMPLATES_DIR / name).read_text(encoding="utf-8")) _TEMPLATE_CACHE[name] = cached return cached def render(name: str, **context: Any) -> str: """Fill one template. A missing placeholder raises instead of rendering empty.""" return template(name).substitute(**context) def esc(value: Any) -> str: """Escape anything that reaches the markup; inventory text is operator input.""" return html.escape(str(value), quote=True) # -------------------------------------------------------------------------- # Privilege (docs/DESIGN.md "Privilege") # -------------------------------------------------------------------------- @dataclass(frozen=True) class Privilege: """The panel's privilege level: a state on screen, not a buried sentence.""" level: str word: str reason: str @property def is_read_only(self) -> bool: return self.level == PRIVILEGE_READ_ONLY_LEVEL PRIVILEGE_READ_ONLY_LEVEL = "read-only" PRIVILEGE_FULL_LEVEL = "full" PRIVILEGES: dict[str, Privilege] = { PRIVILEGE_READ_ONLY_LEVEL: Privilege( PRIVILEGE_READ_ONLY_LEVEL, "Salt-okunur", "Yazma reddedilir: bu panel yalnız okuma yapar (GET, HEAD).", ), PRIVILEGE_FULL_LEVEL: Privilege( PRIVILEGE_FULL_LEVEL, "Tam yetki", "Yazma yetkisi açık, ancak bu sürümde yazma uç noktası uygulanmadı; hiçbir uç nokta cihazda değişiklik yapmaz.", ), } def privilege_from_level(level: str) -> Privilege: try: return PRIVILEGES[level] except KeyError: # pragma: no cover - argparse restricts the value raise ValueError(f"bilinmeyen yetki düzeyi: {level!r}") from None @dataclass(frozen=True) class ServerInfo: """Facts about the running panel, for the about/durum page.""" host: str port: int privilege: Privilege python_version: str started_at: str data_path: str | None = None @property def bind_address(self) -> str: return f"{self.host}:{self.port}" # -------------------------------------------------------------------------- # Query (server-side filtering: no JavaScript ships with this panel) # -------------------------------------------------------------------------- @dataclass(frozen=True) class PanelQuery: """The operator's request: a filter, an optional explicit device.""" criteria: FleetFilter device: str = "" warnings: tuple[str, ...] = () @classmethod def parse(cls, params: Mapping[str, Any]) -> "PanelQuery": criteria, warnings = fleet_data.parse_filter(params) device = fleet_data.first_param(params, "cihaz") or fleet_data.first_param(params, "device") return cls(criteria=criteria, device=device, warnings=warnings) @property def has_filter(self) -> bool: return self.criteria.is_active def _params(self, device: str | None) -> dict[str, str]: params: dict[str, str] = {} if self.criteria.query: params["q"] = self.criteria.query if self.criteria.state: params["durum"] = self.criteria.state if self.criteria.wave: params["dalga"] = self.criteria.wave if self.criteria.only_not_ready: params["hazir_degil"] = "1" if device: params["cihaz"] = device return params def href(self, path: str = FLEET_PATH, *, device: str | None = None) -> str: """Link that keeps the current filter, optionally selecting a device.""" query = urlencode(self._params(device)) return f"{path}?{query}" if query else path @property def reload_href(self) -> str: """The primary action: re-read the inventory, keep the filter.""" return self.href(FLEET_PATH) # -------------------------------------------------------------------------- # Fragments # -------------------------------------------------------------------------- def _route_link(href: str) -> str: return esc(href) def _summary(snapshot: FleetSnapshot, shown: Sequence[Device]) -> dict[str, str]: counts = fleet_data.state_counts(shown) ready = counts["HAZIR"] cells = "".join( render( "summary_cell.html", count=str(counts[spec.word]), word=spec.word, note=STATE_SHORT_NOTES[spec.word], ) for spec in STATE_VOCABULARY ) return { "lead_figure": str(ready), "lead_label": "HAZIR — tek sevk edilebilir durum", "lead_note": ( f"Gösterilen {len(shown)} cihaz içinde: hazır {ready}, hazır değil " f"{len(shown) - ready}." ), "summary_cells": cells, } def _device_rows(shown: Sequence[Device], selected_id: str, query: PanelQuery) -> str: rows: list[str] = [] for device in shown: is_selected = device.device_id == selected_id rows.append( render( "device_row.html", row_selected_class=" fleet-row--selected" if is_selected else "", row_selected_attr=' aria-current="true"' if is_selected else "", state_token_suffix=device.state_token_suffix, state_meaning=esc(f"{device.state_label}: {device.state_meaning}"), state_label=esc(device.state_label), device_id=esc(device.device_id), hostname=esc(device.hostname), wave_code=esc(device.wave_code), wave_label=esc(device.wave_label), wave_name=esc(device.wave.label), last_seen_text=esc(device.last_seen_text), select_href=esc(query.href(FLEET_PATH, device=device.device_id)), ) ) return "\n".join(rows) def _channel_rows(device: Device) -> str: rows: list[str] = [] for check in device.channels: rows.append( render( "channel_row.html", label=esc(check.label), tone=check.tone, status_word=esc(check.status_word), checked_at_text=esc(check.checked_at_text), hint=esc(check.hint), ) ) return "\n".join(rows) def _detail( snapshot: FleetSnapshot, shown: Sequence[Device], selection: Device | None, query: PanelQuery, case: str, ) -> str: if case == "unknown": return render( "detail_unknown.html", device_id=esc(query.device), reset_href=esc(FLEET_PATH), ) if case == "empty" or selection is None: if snapshot.total == 0: empty = ( "Envanterde cihaz kaydı yok. Veri kaynağı sayfanın üstünde ve " "«Hakkında ve durum» sayfasında yazılı." ) else: empty = ( "Bu filtreye uyan cihaz yok · uygulanan filtre: " f"{esc(query.criteria.description)}." ) return render("detail_empty.html", empty_text=empty) if snapshot.is_demo: record_source = "sentetik demo kaydı (gerçek cihaz değil)" elif snapshot.source_path: record_source = f"envanter dosyası: {esc(snapshot.source_path)}" else: record_source = "envanter dosyası" selection_note_block = "" if selection not in shown: selection_note_block = render( "selection_note.html", text=( "Seçili cihaz geçerli filtreye uymuyor; ayrıntısı yine de " "gösteriliyor." ), ) reason_block = "" if selection.reason: reason = selection.reason if selection.claimed_state: reason += f" Bildirilen durum: {selection.claimed_state}." reason_block = render("reason_block.html", text=esc(reason)) note_block = "" if selection.note: note_block = render("note_block.html", text=esc(selection.note)) notes_block = "" if selection.notes: items = "\n".join( render("notes_item.html", text=esc(note)) for note in selection.notes ) notes_block = render("notes_list.html", items=items) return render( "detail.html", state_token_suffix=selection.state_token_suffix, state_meaning=esc(f"{selection.state_label}: {selection.state_meaning}"), state_label=esc(selection.state_label), selection_note_block=selection_note_block, reason_block=reason_block, note_block=note_block, notes_block=notes_block, device_id=esc(selection.device_id), hostname=esc(selection.hostname), dealer_id=esc(selection.dealer_id), wave_code=esc(selection.wave_code), wave_label=esc(selection.wave_label), wave_name=esc(selection.wave.label), wave_target=str(selection.wave.target), last_seen_text=esc(selection.last_seen_text), record_source=record_source, ready_gate_note=esc(selection.ready_gate_note), channel_rows=_channel_rows(selection), channel_count_note=( f"Doğrulanan kanal: {selection.verified_channel_count}/{len(CHANNELS)}. " "HAZIR yalnız dördü de «doğrulandı» olduğunda verilir (docs/27)." ), ) def _filter_bar(snapshot: FleetSnapshot, shown: Sequence[Device], query: PanelQuery) -> str: state_options = [render("option.html", value="", selected="", label="Tüm durumlar")] for spec in STATE_VOCABULARY: state_options.append( render( "option.html", value=esc(spec.state), selected=' selected' if query.criteria.state == spec.state else "", label=esc(f"{spec.word} — {STATE_SHORT_NOTES[spec.word]}"), ) ) wave_options = [render("option.html", value="", selected="", label="Tüm dalgalar")] for wave in WAVES: wave_options.append( render( "option.html", value=esc(wave.code), selected=' selected' if query.criteria.wave == wave.code else "", label=esc(f"{wave.full_label} — hedef {wave.target} cihaz"), ) ) return render( "filter_bar.html", form_action=FLEET_PATH, query=esc(query.criteria.query), state_options="\n".join(state_options), wave_options="\n".join(wave_options), only_not_ready_checked=' checked' if query.criteria.only_not_ready else "", reset_href=esc(FLEET_PATH), shown_count=str(len(shown)), total_count=str(snapshot.total), filter_description=esc(query.criteria.description), ) def _wave_strip(snapshot: FleetSnapshot) -> str: counts = fleet_data.wave_counts(snapshot.devices) items = " · ".join( render( "wave_item.html", code=esc(wave.code), count=str(counts[wave.code]), target=str(wave.target), ) for wave in WAVES ) return render("wave_strip.html", wave_items=items) def _warnings_block(warnings: Sequence[str]) -> str: if not warnings: return "" items = "\n".join(render("warning_item.html", text=esc(text)) for text in warnings) return render("warnings.html", warning_items=items) def _invalid_block(snapshot: FleetSnapshot) -> str: if not snapshot.invalid: return "" items = "\n".join( render("invalid_item.html", record_ref=esc(record.record_ref), reason=esc(record.reason)) for record in snapshot.invalid ) return render( "invalid_block.html", invalid_count=str(len(snapshot.invalid)), invalid_items=items, ) def _empty_note(snapshot: FleetSnapshot, shown: Sequence[Device], query: PanelQuery) -> str: if shown: return "" if snapshot.total == 0: return render( "empty_note.html", text="Envanterde gösterilecek cihaz yok.", ) return render( "empty_note.html", text=f"Bu filtreye uyan cihaz yok · uygulanan filtre: {esc(query.criteria.description)}.", ) # -------------------------------------------------------------------------- # Selections # -------------------------------------------------------------------------- def resolve_selection( snapshot: FleetSnapshot, shown: Sequence[Device], query: PanelQuery ) -> tuple[Device | None, str]: """-> (device, case) with case in ``selected`` / ``unknown`` / ``empty``. No explicit device means the first row of the list is the selection, so the first viewport is never half empty; an explicit id the inventory does not know is reported as ``unknown`` rather than invented. """ if query.device: wanted = fleet_data.fold(query.device) for device in snapshot.devices: if wanted in ( fleet_data.fold(device.device_id), fleet_data.fold(device.hostname), device.dealer_id, ): return device, "selected" return None, "unknown" if shown: return shown[0], "selected" return None, "empty" def target_identity( snapshot: FleetSnapshot, shown: Sequence[Device], selection: Device | None, query: PanelQuery, ) -> str: """The header band's target identity: what this page is about.""" if query.device: if selection is not None: return f"{selection.device_id} · {selection.hostname}" return f"{query.device} (envanterde yok)" if query.criteria.wave: wave: Wave = fleet_data.WAVE_BY_CODE[query.criteria.wave] return f"Dalga {wave.full_label} · hedef {wave.target} cihaz" if query.criteria.state: spec = fleet_data.STATE_BY_NAME[query.criteria.state] return f"Durum {spec.word} · gösterilen {len(shown)} cihaz" if query.criteria.is_active: return f"Filtre · gösterilen {len(shown)} / {snapshot.total} cihaz" return f"Tüm filo · {snapshot.total} cihaz" # -------------------------------------------------------------------------- # Shell # -------------------------------------------------------------------------- def _shell( *, page_title: str, content: str, snapshot: FleetSnapshot, privilege: Privilege, server_info: ServerInfo, target: str, nav_section: str, primary_href: str, ) -> str: footer_source = f"Veri kaynağı: {snapshot.source_label}" footer_meta = ( f"{PANEL_SURFACE} {PANEL_VERSION} · veri yüklemesi: {snapshot.loaded_at} · " f"yetki: {privilege.word} · Python {server_info.python_version}" ) return render( "base.html", page_title=esc(page_title), panel_name=esc(PANEL_NAME), panel_surface=esc(PANEL_SURFACE), target_identity=esc(target), privilege_level=privilege.level, privilege_word=esc(privilege.word), privilege_reason=esc(privilege.reason), primary_action=render("primary_action.html", href=esc(primary_href)), gated_action=render("gated_action.html", reason=esc(privilege.reason)), nav_fleet_attr=' aria-current="page"' if nav_section == "fleet" else "", nav_about_attr=' aria-current="page"' if nav_section == "about" else "", content=content, footer_source=esc(footer_source), footer_meta=esc(footer_meta), ) # -------------------------------------------------------------------------- # Pages # -------------------------------------------------------------------------- def render_fleet_page( snapshot: FleetSnapshot, params: Mapping[str, Any] | None = None, *, privilege: Privilege | None = None, server_info: ServerInfo | None = None, ) -> str: """The first screen: which devices are installed, and in which wave.""" query = PanelQuery.parse(params or {}) shown = fleet_data.sort_devices( fleet_data.filter_devices(snapshot.devices, query.criteria) ) selection, case = resolve_selection(snapshot, shown, query) selected_id = selection.device_id if selection is not None and selection in shown else "" summary = _summary(snapshot, shown) content = render( "fleet.html", data_banner_tone=snapshot.banner_tone, data_banner_word=esc(snapshot.banner_word), data_banner_text=esc(snapshot.banner_text), warnings_block=_warnings_block(query.warnings), list_count_text=f"{len(shown)} / {snapshot.total} cihaz", filter_bar=_filter_bar(snapshot, shown, query), wave_strip=_wave_strip(snapshot), invalid_block=_invalid_block(snapshot), device_rows=_device_rows(shown, selected_id, query), empty_note=_empty_note(snapshot, shown, query), device_detail=_detail(snapshot, shown, selection, query, case), **summary, ) return _shell( page_title=f"{PANEL_NAME} — {PANEL_SURFACE} · Filo", content=content, snapshot=snapshot, privilege=privilege or PRIVILEGES[PRIVILEGE_READ_ONLY_LEVEL], server_info=server_info or _fallback_server_info(privilege), target=target_identity(snapshot, shown, selection, query), nav_section="fleet", primary_href=query.reload_href, ) def render_about_page( snapshot: FleetSnapshot, *, privilege: Privilege | None = None, server_info: ServerInfo | None = None, ) -> str: """The panel's own facts: version, data source and the rules it applies.""" privilege = privilege or PRIVILEGES[PRIVILEGE_READ_ONLY_LEVEL] server_info = server_info or _fallback_server_info(privilege) vocabulary_rows = "\n".join( render( "about_vocabulary_row.html", state_token_suffix=spec.token_suffix, word=esc(spec.word), token=esc(spec.token), meaning=esc(spec.meaning), ) for spec in STATE_VOCABULARY ) counts = fleet_data.wave_counts(snapshot.devices) wave_rows = "\n".join( render( "about_wave_row.html", code=esc(wave.code), label=esc(wave.label), target=str(wave.target), in_panel=str(counts[wave.code]), ) for wave in WAVES ) channel_rows = "\n".join( render( "about_channel_row.html", label=esc(spec.label), key=esc(spec.key), hint=esc(spec.hint), ) for spec in CHANNELS ) rule_items = "\n".join(render("rule_item.html", text=esc(rule)) for rule in PANEL_RULES) content = render( "about.html", panel_name=esc(PANEL_NAME), panel_surface=esc(PANEL_SURFACE), panel_version=esc(PANEL_VERSION), python_version=esc(server_info.python_version), bind_address=esc(server_info.bind_address), privilege_level=privilege.level, privilege_word=esc(privilege.word), privilege_reason=esc(privilege.reason), started_at=esc(server_info.started_at), data_banner_tone=snapshot.banner_tone, data_banner_word=esc(snapshot.banner_word), data_banner_text=esc(snapshot.banner_text), source_kind_word=esc(snapshot.source_kind_word), source_path_text=esc(snapshot.source_path or "kullanılmadı (demo veri)"), schema_version_text=esc(str(snapshot.schema_version) if snapshot.schema_version else "yok"), generated_at_text=esc(snapshot.generated_at or "bildirilmemiş"), loaded_at=esc(snapshot.loaded_at), total_text=str(snapshot.total), ready_text=f"{snapshot.ready_count} hazır · {snapshot.not_ready_count} hazır değil", invalid_count_text=str(len(snapshot.invalid)), error_text=esc(snapshot.error or "yok"), vocabulary_rows=vocabulary_rows, wave_rows=wave_rows, channel_rows=channel_rows, rule_items=rule_items, ) return _shell( page_title=f"{PANEL_NAME} — {PANEL_SURFACE} · Hakkında ve durum", content=content, snapshot=snapshot, privilege=privilege, server_info=server_info, target=f"Panel · sürüm {PANEL_VERSION}", nav_section="about", primary_href=ABOUT_PATH, ) def render_refused_page( method: str, *, snapshot: FleetSnapshot, privilege: Privilege, server_info: ServerInfo, ) -> str: content = render( "refused_body.html", method=esc(method), privilege_word=esc(privilege.word), ) return _shell( page_title=f"{PANEL_NAME} — Yöntem reddedildi", content=content, snapshot=snapshot, privilege=privilege, server_info=server_info, target=f"Reddedilen istek · {method}", nav_section="", primary_href=FLEET_PATH, ) def render_not_found_page( request_path: str, *, snapshot: FleetSnapshot, privilege: Privilege, server_info: ServerInfo, ) -> str: content = render("not_found_body.html", request_path=esc(request_path)) return _shell( page_title=f"{PANEL_NAME} — Sayfa yok", content=content, snapshot=snapshot, privilege=privilege, server_info=server_info, target=f"Tanımsız adres · {request_path}", nav_section="", primary_href=FLEET_PATH, ) def _fallback_server_info(privilege: Privilege | None) -> ServerInfo: """Used when a page is rendered outside the server (tests, --check).""" import sys level = privilege.level if privilege is not None else PRIVILEGE_READ_ONLY_LEVEL return ServerInfo( host="127.0.0.1", port=0, privilege=PRIVILEGES[level], python_version=f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", started_at=fleet_data.timestamp_now(), )