from __future__ import annotations import hashlib import json import os import re from collections.abc import Callable from pathlib import Path from threading import Lock from typing import Annotated, Literal from urllib.parse import unquote, urlsplit from fastapi import APIRouter, Header, HTTPException, Query, Request from fastapi import Path as ApiPath from fastapi.responses import FileResponse from pydantic import BaseModel, ConfigDict, Field, model_validator EnvironmentSurfaceId = Literal[ "home", "fleet", "observation", "missions", "data", "system", "polygon", ] ENVIRONMENT_SURFACE_IDS: tuple[EnvironmentSurfaceId, ...] = ( "home", "fleet", "observation", "missions", "data", "system", "polygon", ) EnvironmentMediaKind = Literal["image", "video"] EnvironmentMediaSource = Literal["file", "url"] ENVIRONMENT_SCHEMA_VERSION: Literal["missioncore.operator-environment/v3"] = ( "missioncore.operator-environment/v3" ) ENVIRONMENT_MEDIA_SCHEMA_VERSION: Literal[ "missioncore.operator-environment-media/v1" ] = "missioncore.operator-environment-media/v1" ENVIRONMENT_PLAYLIST_MEDIA_SCHEMA_VERSION: Literal[ "missioncore.operator-environment-media/v2" ] = "missioncore.operator-environment-media/v2" MAX_ENVIRONMENT_MEDIA_BYTES = 256 * 1024 * 1024 MAX_ENVIRONMENT_MEDIA_ITEMS = 24 DEFAULT_IMAGE_DURATION_SECONDS = 10 SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9А-Яа-яЁё._ -]+") SAFE_MEDIA_ITEM_ID = re.compile(r"^media-[a-z0-9-]{1,58}$") SUPPORTED_MEDIA_TYPES: dict[str, tuple[EnvironmentMediaKind, str]] = { "image/avif": ("image", ".avif"), "image/gif": ("image", ".gif"), "image/jpeg": ("image", ".jpg"), "image/png": ("image", ".png"), "image/webp": ("image", ".webp"), "video/mp4": ("video", ".mp4"), "video/quicktime": ("video", ".mov"), "video/webm": ("video", ".webm"), } class StrictApiModel(BaseModel): model_config = ConfigDict(extra="forbid") class EnvironmentMediaItem(StrictApiModel): id: str = Field(pattern=r"^media-[a-z0-9-]{1,58}$") source: EnvironmentMediaSource = "file" url: str = Field(max_length=2048) media_kind: EnvironmentMediaKind file_name: str | None = Field(default=None, max_length=255) @model_validator(mode="after") def validate_source(self) -> EnvironmentMediaItem: if self.source == "url": parsed = urlsplit(self.url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError("external media URL must use HTTP or HTTPS") elif not self.url.startswith("/api/v1/environment/media/"): raise ValueError("file media must use the Mission Core media endpoint") return self class EnvironmentBackground(StrictApiModel): enabled: bool = False image_duration_seconds: int = Field( default=DEFAULT_IMAGE_DURATION_SECONDS, ge=1, le=300, ) items: list[EnvironmentMediaItem] = Field( default_factory=list, max_length=MAX_ENVIRONMENT_MEDIA_ITEMS, ) @model_validator(mode="after") def validate_playlist(self) -> EnvironmentBackground: ids = [item.id for item in self.items] if len(ids) != len(set(ids)): raise ValueError("background media ids must be unique") if self.enabled and not self.items: raise ValueError("enabled background requires at least one media item") return self class EnvironmentPage(StrictApiModel): header_label: str = Field(min_length=1, max_length=40) eyebrow: str = Field(min_length=1, max_length=80) title: str = Field(min_length=1, max_length=120) description: str = Field(min_length=1, max_length=500) primary_workspace_id: str | None = Field( default=None, max_length=80, pattern=r"^[a-z0-9-]+$", ) secondary_workspace_id: str | None = Field( default=None, max_length=80, pattern=r"^[a-z0-9-]+$", ) background: EnvironmentBackground @model_validator(mode="after") def validate_quick_actions(self) -> EnvironmentPage: if ( self.primary_workspace_id is not None and self.primary_workspace_id == self.secondary_workspace_id ): raise ValueError("quick actions must target different workspaces") return self class EnvironmentPages(StrictApiModel): home: EnvironmentPage fleet: EnvironmentPage observation: EnvironmentPage missions: EnvironmentPage data: EnvironmentPage system: EnvironmentPage polygon: EnvironmentPage class EnvironmentSettingsPut(StrictApiModel): revision: int = Field(ge=0) pages: EnvironmentPages class EnvironmentSettingsDocument(EnvironmentSettingsPut): schema_version: Literal["missioncore.operator-environment/v3"] = ENVIRONMENT_SCHEMA_VERSION class EnvironmentMediaDocument(StrictApiModel): schema_version: Literal["missioncore.operator-environment-media/v1"] = ( ENVIRONMENT_MEDIA_SCHEMA_VERSION ) surface_id: EnvironmentSurfaceId url: str file_name: str media_kind: EnvironmentMediaKind media_type: str byte_length: int = Field(ge=1, le=MAX_ENVIRONMENT_MEDIA_BYTES) sha256: str = Field(pattern=r"^[a-f0-9]{64}$") class EnvironmentPlaylistMediaDocument(StrictApiModel): schema_version: Literal["missioncore.operator-environment-media/v2"] = ( ENVIRONMENT_PLAYLIST_MEDIA_SCHEMA_VERSION ) surface_id: EnvironmentSurfaceId item_id: str = Field(pattern=r"^media-[a-z0-9-]{1,58}$") url: str file_name: str media_kind: EnvironmentMediaKind media_type: str byte_length: int = Field(ge=1, le=MAX_ENVIRONMENT_MEDIA_BYTES) sha256: str = Field(pattern=r"^[a-f0-9]{64}$") def default_environment_settings() -> EnvironmentSettingsDocument: background = EnvironmentBackground() return EnvironmentSettingsDocument( revision=0, pages=EnvironmentPages( home=EnvironmentPage( header_label="Mission Core", eyebrow="NODEDC / MISSION CORE", title="Mission Core", description=( "Наблюдение, планирование и корректировка миссий " "в одной модульной рабочей области." ), primary_workspace_id="spatial-scene", secondary_workspace_id="local-device", background=background.model_copy(), ), fleet=EnvironmentPage( header_label="Парк", eyebrow="ПАРК / УСТРОЙСТВА", title="Аппараты и устройства", description=( "Одинаково подключать одиночный стенд, наземную платформу " "и будущий рой." ), primary_workspace_id="contour-health", secondary_workspace_id="local-device", background=background.model_copy(), ), observation=EnvironmentPage( header_label="Наблюдение", eyebrow="СИТУАЦИОННАЯ ОСВЕДОМЛЁННОСТЬ", title="Наблюдение", description=( "Свести все сенсоры в синхронную и управляемую операторскую картину." ), primary_workspace_id="cameras", secondary_workspace_id="world-map", background=background.model_copy(), ), missions=EnvironmentPage( header_label="Миссии", eyebrow="УПРАВЛЕНИЕ МИССИЯМИ", title="Миссии", description=( "Собрать задачу из точек, ограничений и действий до передачи на борт." ), primary_workspace_id="mission-planner", secondary_workspace_id=None, background=background.model_copy(), ), data=EnvironmentPage( header_label="Данные", eyebrow="РАБОЧАЯ ОБЛАСТЬ ДАННЫХ", title="Данные и записи", description=( "Хранить живой контур и воспроизводимый эксперимент " "как одну модель данных." ), primary_workspace_id="recordings", secondary_workspace_id="datasets", background=background.model_copy(), ), system=EnvironmentPage( header_label="Система", eyebrow="УПРАВЛЕНИЕ ПЛАТФОРМОЙ", title="Система", description=( "Подключать новые возможности модульно, " "не связывая интерфейс с одним устройством." ), primary_workspace_id="modules", secondary_workspace_id="integrations", background=background.model_copy(), ), polygon=EnvironmentPage( header_label="Тестировочный контур", eyebrow="ЛАБОРАТОРНЫЕ ИССЛЕДОВАНИЯ", title="Тестировочный контур", description=( "Фиксировать каждый эксперимент как проверяемую лабораторную работу." ), primary_workspace_id="lab-archive", secondary_workspace_id=None, background=background.model_copy(), ), ), ) def _upgrade_legacy_background( surface_id: str, value: object, ) -> EnvironmentBackground: if not isinstance(value, dict): raise ValueError(f"legacy background {surface_id} must be an object") enabled = value.get("enabled") source = value.get("source") url = value.get("url") media_kind = value.get("media_kind") file_name = value.get("file_name") if not isinstance(enabled, bool): raise ValueError(f"legacy background {surface_id} is incomplete") if url is None and media_kind is None: return EnvironmentBackground(enabled=False) if source == "file": source_value: EnvironmentMediaSource = "file" elif source == "url": source_value = "url" else: raise ValueError(f"legacy background {surface_id} has invalid source") if not isinstance(url, str): raise ValueError(f"legacy background {surface_id} has invalid URL") if media_kind == "image": media_kind_value: EnvironmentMediaKind = "image" elif media_kind == "video": media_kind_value = "video" else: raise ValueError(f"legacy background {surface_id} has invalid media kind") if file_name is not None and not isinstance(file_name, str): raise ValueError(f"legacy background {surface_id} has invalid file name") item = EnvironmentMediaItem( id=f"media-legacy-{surface_id}", source=source_value, url=url, media_kind=media_kind_value, file_name=file_name, ) return EnvironmentBackground(enabled=enabled, items=[item]) def _upgrade_v1_environment(payload: dict[str, object]) -> EnvironmentSettingsDocument: revision = payload.get("revision") labels = payload.get("header_labels") backgrounds = payload.get("backgrounds") if not isinstance(revision, int) or not isinstance(labels, dict) or not isinstance( backgrounds, dict, ): raise ValueError("legacy environment document is incomplete") defaults = default_environment_settings() pages: dict[str, EnvironmentPage] = {} for surface_id, default_page in defaults.pages: legacy_label = labels.get(surface_id) pages[surface_id] = default_page.model_copy( update={ "header_label": ( legacy_label if isinstance(legacy_label, str) and legacy_label.strip() else default_page.header_label ), "background": _upgrade_legacy_background( surface_id, backgrounds.get(surface_id), ), }, ) return EnvironmentSettingsDocument( revision=revision, pages=EnvironmentPages.model_validate(pages), ) def _upgrade_v2_environment(payload: dict[str, object]) -> EnvironmentSettingsDocument: revision = payload.get("revision") raw_pages = payload.get("pages") if not isinstance(revision, int) or not isinstance(raw_pages, dict): raise ValueError("v2 environment document is incomplete") pages: dict[str, EnvironmentPage] = {} for surface_id in ENVIRONMENT_SURFACE_IDS: raw_page = raw_pages.get(surface_id) if not isinstance(raw_page, dict): raise ValueError(f"v2 environment page {surface_id} is incomplete") pages[surface_id] = EnvironmentPage.model_validate( { **raw_page, "background": _upgrade_legacy_background( surface_id, raw_page.get("background"), ), } ) return EnvironmentSettingsDocument( revision=revision, pages=EnvironmentPages.model_validate(pages), ) def _upgrade_environment(payload: object) -> EnvironmentSettingsDocument: if not isinstance(payload, dict): raise ValueError("environment document must be an object") schema_version = payload.get("schema_version") if schema_version == "missioncore.operator-environment/v1": document = _upgrade_v1_environment(payload) elif schema_version == "missioncore.operator-environment/v2": document = _upgrade_v2_environment(payload) else: document = EnvironmentSettingsDocument.model_validate(payload) # The owner moved the local live scene to LAB. Retire only its old # Control shortcuts, preserving custom copy, backgrounds and other links. page = document.pages.observation if page.primary_workspace_id == "spatial-scene": page.primary_workspace_id = ( page.secondary_workspace_id if page.secondary_workspace_id not in (None, "spatial-scene") else "cameras" ) page.secondary_workspace_id = None elif page.secondary_workspace_id == "spatial-scene": page.secondary_workspace_id = None return document class EnvironmentSettingsStore: def __init__(self, root: Path) -> None: self.root = root.expanduser().resolve() self.settings_path = self.root / "settings.json" self.media_root = self.root / "media" self._lock = Lock() def read(self) -> EnvironmentSettingsDocument: with self._lock: if not self.settings_path.is_file(): return default_environment_settings() try: payload = json.loads(self.settings_path.read_text(encoding="utf-8")) return _upgrade_environment(payload) except (OSError, ValueError) as exc: raise RuntimeError("operator environment settings are corrupt") from exc def save(self, request: EnvironmentSettingsPut) -> EnvironmentSettingsDocument: with self._lock: current = self._read_unlocked() if request.revision != current.revision: raise RuntimeError("operator environment settings revision changed") document = EnvironmentSettingsDocument( revision=current.revision + 1, pages=request.pages, ) self.root.mkdir(mode=0o700, parents=True, exist_ok=True) temporary = self.settings_path.with_suffix(".json.tmp") temporary.write_text( json.dumps( document.model_dump(mode="json"), ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) os.replace(temporary, self.settings_path) return document def _read_unlocked(self) -> EnvironmentSettingsDocument: if not self.settings_path.is_file(): return default_environment_settings() try: payload = json.loads(self.settings_path.read_text(encoding="utf-8")) return _upgrade_environment(payload) except (OSError, ValueError) as exc: raise RuntimeError("operator environment settings are corrupt") from exc def media_metadata_path(self, surface_id: EnvironmentSurfaceId) -> Path: return self.media_root / f"{surface_id}.json" def read_media( self, surface_id: EnvironmentSurfaceId, generation: str, ) -> tuple[Path, EnvironmentMediaDocument]: with self._lock: metadata_path = self.media_metadata_path(surface_id) try: metadata = EnvironmentMediaDocument.model_validate_json( metadata_path.read_text(encoding="utf-8") ) except (OSError, ValueError) as exc: raise FileNotFoundError(surface_id) from exc if generation != metadata.sha256: raise PermissionError("media generation changed") path = self.media_root / f"{surface_id}{SUPPORTED_MEDIA_TYPES[metadata.media_type][1]}" if not path.is_file() or path.stat().st_size != metadata.byte_length: raise FileNotFoundError(surface_id) return path, metadata def finalize_media( self, *, surface_id: EnvironmentSurfaceId, temporary_path: Path, file_name: str, media_type: str, byte_length: int, sha256: str, ) -> EnvironmentMediaDocument: media_kind, extension = SUPPORTED_MEDIA_TYPES[media_type] with self._lock: self.media_root.mkdir(mode=0o700, parents=True, exist_ok=True) destination = self.media_root / f"{surface_id}{extension}" for stale_path in self.media_root.glob(f"{surface_id}.*"): if stale_path == self.media_metadata_path(surface_id): continue if stale_path != destination and stale_path.is_file(): stale_path.unlink() os.replace(temporary_path, destination) document = EnvironmentMediaDocument( surface_id=surface_id, url=f"/api/v1/environment/media/{surface_id}?generation={sha256}", file_name=file_name, media_kind=media_kind, media_type=media_type, byte_length=byte_length, sha256=sha256, ) metadata_path = self.media_metadata_path(surface_id) metadata_temporary = metadata_path.with_suffix(".json.tmp") metadata_temporary.write_text( json.dumps(document.model_dump(mode="json"), ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) os.replace(metadata_temporary, metadata_path) return document def playlist_media_root(self, surface_id: EnvironmentSurfaceId) -> Path: return self.media_root / surface_id def playlist_media_metadata_path( self, surface_id: EnvironmentSurfaceId, item_id: str, ) -> Path: if not SAFE_MEDIA_ITEM_ID.fullmatch(item_id): raise ValueError("invalid media item id") return self.playlist_media_root(surface_id) / f"{item_id}.json" def read_playlist_media( self, surface_id: EnvironmentSurfaceId, item_id: str, generation: str, ) -> tuple[Path, EnvironmentPlaylistMediaDocument]: with self._lock: metadata_path = self.playlist_media_metadata_path(surface_id, item_id) try: metadata = EnvironmentPlaylistMediaDocument.model_validate_json( metadata_path.read_text(encoding="utf-8") ) except (OSError, ValueError) as exc: raise FileNotFoundError(item_id) from exc if generation != metadata.sha256: raise PermissionError("media generation changed") extension = SUPPORTED_MEDIA_TYPES[metadata.media_type][1] path = self.playlist_media_root(surface_id) / f"{item_id}{extension}" if not path.is_file() or path.stat().st_size != metadata.byte_length: raise FileNotFoundError(item_id) return path, metadata def finalize_playlist_media( self, *, surface_id: EnvironmentSurfaceId, item_id: str, temporary_path: Path, file_name: str, media_type: str, byte_length: int, sha256: str, ) -> EnvironmentPlaylistMediaDocument: if not SAFE_MEDIA_ITEM_ID.fullmatch(item_id): raise ValueError("invalid media item id") media_kind, extension = SUPPORTED_MEDIA_TYPES[media_type] with self._lock: item_root = self.playlist_media_root(surface_id) item_root.mkdir(mode=0o700, parents=True, exist_ok=True) destination = item_root / f"{item_id}{extension}" metadata_path = self.playlist_media_metadata_path(surface_id, item_id) for stale_path in item_root.glob(f"{item_id}.*"): if stale_path == metadata_path: continue if stale_path != destination and stale_path.is_file(): stale_path.unlink() os.replace(temporary_path, destination) document = EnvironmentPlaylistMediaDocument( surface_id=surface_id, item_id=item_id, url=( f"/api/v1/environment/media/{surface_id}/{item_id}" f"?generation={sha256}" ), file_name=file_name, media_kind=media_kind, media_type=media_type, byte_length=byte_length, sha256=sha256, ) metadata_temporary = metadata_path.with_suffix(".json.tmp") metadata_temporary.write_text( json.dumps(document.model_dump(mode="json"), ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) os.replace(metadata_temporary, metadata_path) return document async def _write_upload( request: Request, temporary_path: Path, ) -> tuple[int, str]: digest = hashlib.sha256() byte_length = 0 try: with temporary_path.open("xb") as output: async for chunk in request.stream(): if not chunk: continue byte_length += len(chunk) if byte_length > MAX_ENVIRONMENT_MEDIA_BYTES: raise HTTPException(status_code=413, detail="Файл превышает лимит 256 МБ.") output.write(chunk) digest.update(chunk) if byte_length == 0: raise HTTPException(status_code=422, detail="Пустой файл не поддерживается.") return byte_length, digest.hexdigest() except BaseException: temporary_path.unlink(missing_ok=True) raise def _safe_file_name(value: str | None, extension: str) -> str: decoded = unquote(value) if value else f"background{extension}" candidate = SAFE_FILE_NAME.sub("_", decoded.strip()) if not candidate: candidate = f"background{extension}" return candidate[:255] def build_environment_router( root_provider: Callable[[], Path], ) -> APIRouter: router = APIRouter(prefix="/api/v1/environment", tags=["environment"]) def store() -> EnvironmentSettingsStore: return EnvironmentSettingsStore(root_provider()) @router.get("/settings") def get_environment_settings() -> EnvironmentSettingsDocument: try: return store().read() except RuntimeError as exc: raise HTTPException( status_code=500, detail="Настройки окружения повреждены.", ) from exc @router.put("/settings") def put_environment_settings( request: EnvironmentSettingsPut, ) -> EnvironmentSettingsDocument: try: return store().save(request) except RuntimeError as exc: if "revision changed" in str(exc): raise HTTPException( status_code=412, detail="Настройки окружения были изменены в другом окне.", ) from exc raise HTTPException( status_code=500, detail="Настройки окружения повреждены.", ) from exc @router.put("/media/{surface_id}") async def upload_environment_media( surface_id: EnvironmentSurfaceId, request: Request, file_name: Annotated[str | None, Header(alias="X-NODEDC-File-Name")] = None, ) -> EnvironmentMediaDocument: media_type = request.headers.get("content-type", "").split(";", 1)[0].strip().lower() if media_type not in SUPPORTED_MEDIA_TYPES: raise HTTPException( status_code=415, detail="Поддерживаются PNG, JPEG, WebP, GIF, AVIF, MP4, WebM и MOV.", ) content_length = request.headers.get("content-length") if content_length is not None: try: if int(content_length) > MAX_ENVIRONMENT_MEDIA_BYTES: raise HTTPException(status_code=413, detail="Файл превышает лимит 256 МБ.") except ValueError as exc: raise HTTPException(status_code=422, detail="Некорректный размер файла.") from exc environment_store = store() environment_store.media_root.mkdir(mode=0o700, parents=True, exist_ok=True) temporary_path = environment_store.media_root / f".{surface_id}.{os.getpid()}.upload" temporary_path.unlink(missing_ok=True) byte_length, sha256 = await _write_upload(request, temporary_path) extension = SUPPORTED_MEDIA_TYPES[media_type][1] return environment_store.finalize_media( surface_id=surface_id, temporary_path=temporary_path, file_name=_safe_file_name(file_name, extension), media_type=media_type, byte_length=byte_length, sha256=sha256, ) @router.get("/media/{surface_id}") def get_environment_media( surface_id: EnvironmentSurfaceId, generation: Annotated[str, Query(pattern=r"^[a-f0-9]{64}$")], ) -> FileResponse: try: path, metadata = store().read_media(surface_id, generation) except FileNotFoundError as exc: raise HTTPException(status_code=404, detail="Фон окружения не найден.") from exc except PermissionError as exc: raise HTTPException(status_code=412, detail="Фон окружения был заменён.") from exc return FileResponse( path, media_type=metadata.media_type, headers={ "Cache-Control": "private, max-age=31536000, immutable, no-transform", "ETag": f'"{metadata.sha256}"', "X-Content-Type-Options": "nosniff", }, ) @router.put("/media/{surface_id}/{item_id}") async def upload_environment_playlist_media( surface_id: EnvironmentSurfaceId, item_id: Annotated[str, ApiPath(pattern=r"^media-[a-z0-9-]{1,58}$")], request: Request, file_name: Annotated[str | None, Header(alias="X-NODEDC-File-Name")] = None, ) -> EnvironmentPlaylistMediaDocument: media_type = request.headers.get("content-type", "").split(";", 1)[0].strip().lower() if media_type not in SUPPORTED_MEDIA_TYPES: raise HTTPException( status_code=415, detail="Поддерживаются PNG, JPEG, WebP, GIF, AVIF, MP4, WebM и MOV.", ) content_length = request.headers.get("content-length") if content_length is not None: try: if int(content_length) > MAX_ENVIRONMENT_MEDIA_BYTES: raise HTTPException(status_code=413, detail="Файл превышает лимит 256 МБ.") except ValueError as exc: raise HTTPException(status_code=422, detail="Некорректный размер файла.") from exc environment_store = store() item_root = environment_store.playlist_media_root(surface_id) item_root.mkdir(mode=0o700, parents=True, exist_ok=True) temporary_path = item_root / f".{item_id}.{os.getpid()}.upload" temporary_path.unlink(missing_ok=True) byte_length, sha256 = await _write_upload(request, temporary_path) extension = SUPPORTED_MEDIA_TYPES[media_type][1] return environment_store.finalize_playlist_media( surface_id=surface_id, item_id=item_id, temporary_path=temporary_path, file_name=_safe_file_name(file_name, extension), media_type=media_type, byte_length=byte_length, sha256=sha256, ) @router.get("/media/{surface_id}/{item_id}") def get_environment_playlist_media( surface_id: EnvironmentSurfaceId, item_id: Annotated[str, ApiPath(pattern=r"^media-[a-z0-9-]{1,58}$")], generation: Annotated[str, Query(pattern=r"^[a-f0-9]{64}$")], ) -> FileResponse: try: path, metadata = store().read_playlist_media( surface_id, item_id, generation, ) except FileNotFoundError as exc: raise HTTPException(status_code=404, detail="Медиафайл окружения не найден.") from exc except PermissionError as exc: raise HTTPException(status_code=412, detail="Медиафайл окружения был заменён.") from exc return FileResponse( path, media_type=metadata.media_type, headers={ "Cache-Control": "private, max-age=31536000, immutable, no-transform", "ETag": f'"{metadata.sha256}"', "X-Content-Type-Options": "nosniff", }, ) return router