Refine built-in web app experience

app/core/discovery.py

@@ -60,9 +60,7 @@ class DiscoveryService:
 
     def _background_interval_seconds(self) -> int:
         return int(
-            os.getenv(
-                "DISCOVERY_INTERVAL_SECONDS", DEFAULT_DISCOVERY_INTERVAL_SECONDS
-            )
+            os.getenv("DISCOVERY_INTERVAL_SECONDS", DEFAULT_DISCOVERY_INTERVAL_SECONDS)
         )
 
     def _background_missing_threshold(self) -> int:
@@ -161,7 +159,9 @@ class DiscoveryService:
         if not candidates:
             return []
 
-        private_candidates = [candidate for candidate in candidates if candidate.address.is_private]
+        private_candidates = [
+            candidate for candidate in candidates if candidate.address.is_private
+        ]
         usable_candidates = private_candidates or candidates
         preferred_candidates = [
             candidate
@@ -202,9 +202,7 @@ class DiscoveryService:
             f"{local_ip}/{self._auto_min_prefix_len()}",
             strict=False,
         )
-        logger.info(
-            "Авто-discovery fallback: использую локальный сегмент %s", network
-        )
+        logger.info("Авто-discovery fallback: использую локальный сегмент %s", network)
         return [str(network)]
 
     def _get_target_subnets(self) -> List[str]:
@@ -298,6 +296,7 @@ class DiscoveryService:
                 remove_missing=remove_missing,
                 missing_threshold=missing_threshold,
             )
+            state_manager.record_discovery(mode, result)
             logger.info(
                 "Discovery (%s): found=%s added=%s updated=%s removed=%s pending_removal=%s online=%s",
                 mode,
@@ -337,7 +336,9 @@ class DiscoveryService:
             timeout=timeout,
         )
 
-    async def start_background_discovery(self, state_manager, interval: int | None = None):
+    async def start_background_discovery(
+        self, state_manager, interval: int | None = None
+    ):
         interval_seconds = interval or self._background_interval_seconds()
         while True:
             await asyncio.sleep(interval_seconds)

app/core/server_info.py

@@ -7,6 +7,7 @@ from pathlib import Path
 import socket
 
 from app.api.deps import get_master_key
+from app.core.state import state_manager
 
 APP_NAME = "Ignis Core"
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -58,11 +59,7 @@ def _read_version_file() -> str | None:
 
 
 def get_app_version() -> str:
-    return (
-        _clean_env("IGNIS_BUILD_VERSION")
-        or _read_version_file()
-        or "1.0.0"
-    )
+    return _clean_env("IGNIS_BUILD_VERSION") or _read_version_file() or "1.0.0"
 
 
 def _resolve_git_ref(git_dir: Path, ref_name: str) -> str | None:
@@ -144,7 +141,9 @@ def get_configuration_status(build_info: ServerBuildInfo) -> ServerConfiguration
     master_key_configured = get_master_key() is not None
     public_base_url_configured = _clean_env("IGNIS_PUBLIC_BASE_URL") is not None
     scan_network_configured = _clean_env("SCAN_NETWORK") is not None
-    build_metadata_complete = bool(build_info.version and build_info.git_sha and build_info.build_date)
+    build_metadata_complete = bool(
+        build_info.version and build_info.git_sha and build_info.build_date
+    )
     return ServerConfigurationStatus(
         configured=master_key_configured,
         master_key_configured=master_key_configured,
@@ -166,14 +165,21 @@ def build_server_info(
 ) -> dict:
     payload = {
         "app_name": APP_NAME,
+        "instance_name": get_instance_name(),
         "uptime_seconds": get_uptime_seconds(),
         "diagnostics_visible": include_diagnostics,
     }
     if not include_diagnostics:
+        discovery_snapshot = state_manager.get_discovery_snapshot()
+        if discovery_snapshot:
+            payload["discovery"] = {
+                "last_scan_at": discovery_snapshot["last_scan_at"],
+                "last_scan_mode": discovery_snapshot["last_scan_mode"],
+                "online": discovery_snapshot["summary"].get("online"),
+            }
         return payload
 
     build_info = get_build_info()
-    payload["instance_name"] = get_instance_name()
     payload["timezone"] = os.getenv("APP_TIMEZONE", "Asia/Novosibirsk")
     payload["started_at"] = SERVER_STARTED_AT.isoformat()
     payload["build"] = {
@@ -181,6 +187,13 @@ def build_server_info(
         "git_sha": build_info.git_sha,
         "build_date": build_info.build_date,
     }
+    discovery_snapshot = state_manager.get_discovery_snapshot()
+    if discovery_snapshot:
+        payload["discovery"] = {
+            "last_scan_at": discovery_snapshot["last_scan_at"],
+            "last_scan_mode": discovery_snapshot["last_scan_mode"],
+            **discovery_snapshot["summary"],
+        }
     payload["urls"] = asdict(get_server_urls(observed_base_url))
     payload["configuration"] = asdict(get_configuration_status(build_info))
     return payload

app/core/state.py

@@ -1,4 +1,5 @@
 from dataclasses import asdict, dataclass
+from datetime import datetime
 import logging
 from typing import Dict, List
 
@@ -21,6 +22,18 @@ class DiscoveryApplyResult:
         return asdict(self)
 
 
+@dataclass(frozen=True)
+class DiscoverySnapshot:
+    last_scan_at: str
+    last_scan_mode: str
+    summary: DiscoveryApplyResult
+
+    def to_dict(self) -> dict:
+        payload = asdict(self)
+        payload["summary"] = self.summary.to_dict()
+        return payload
+
+
 class StateManager:
     def __init__(self):
         # Храним устройства как Pydantic объекты
@@ -29,6 +42,7 @@ class StateManager:
         self.groups: Dict[str, GroupModel] = {}
         # Сколько подряд циклов discovery устройство не видно
         self._missing_scan_counts: Dict[str, int] = {}
+        self.discovery_snapshot: DiscoverySnapshot | None = None
 
     def update_device(self, device_data: dict):
         """Обновляет или добавляет устройство в состояние."""
@@ -92,6 +106,18 @@ class StateManager:
             online=len(self.devices),
         )
 
+    def record_discovery(self, mode: str, result: DiscoveryApplyResult):
+        self.discovery_snapshot = DiscoverySnapshot(
+            last_scan_at=datetime.now().isoformat(),
+            last_scan_mode=mode,
+            summary=result,
+        )
+
+    def get_discovery_snapshot(self) -> dict | None:
+        if not self.discovery_snapshot:
+            return None
+        return self.discovery_snapshot.to_dict()
+
     def get_group_ips(self, group_id: str) -> List[str]:
         """Возвращает список IP всех ламп, входящих в группу."""
         group = self.groups.get(group_id)