267 lines
11 KiB
Python
267 lines
11 KiB
Python
"""Загрузка и валидация конфигурации NVR из cameras.yaml."""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from functools import lru_cache
|
||
|
||
import yaml
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Camera:
|
||
id: str
|
||
name: str
|
||
enabled: bool
|
||
ip: str
|
||
user: str
|
||
password: str
|
||
main_path: str
|
||
sub_path: str | None = None
|
||
record: bool = True # REC: писать на диск. enabled — камера вообще включена (live+запись)
|
||
storage_id: str = "" # в какое хранилище писать ("" = активное/по умолчанию)
|
||
|
||
def rtsp_url(self, path: str) -> str:
|
||
return f"rtsp://{self.user}:{self.password}@{self.ip}:554{path}"
|
||
|
||
@property
|
||
def record_url(self) -> str:
|
||
"""Поток для записи (основной, H.265, копированием)."""
|
||
return self.rtsp_url(self.main_path)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StorageItem:
|
||
"""Одно хранилище записи: локальная папка или сетевой диск (SMB/NAS).
|
||
|
||
`path` — путь ВНУТРИ контейнера, куда пишутся сегменты. `quota_gb` — лимит
|
||
размера ЭТОГО хранилища (0 = без лимита). Для NAS поля server/share/username/
|
||
password/subpath — метаданные (монтаж делается на хосте, backend ими не
|
||
монтирует, только показывает статус и генерирует команду mount).
|
||
"""
|
||
id: str
|
||
name: str
|
||
type: str # "local" | "nas"
|
||
path: str
|
||
quota_gb: int = 0
|
||
server: str = ""
|
||
share: str = ""
|
||
username: str = ""
|
||
password: str = ""
|
||
subpath: str = ""
|
||
enabled: bool = True
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Storage:
|
||
"""Глобальная политика записи + список хранилищ и id активного.
|
||
|
||
Квота — на каждом StorageItem (не здесь). retention_days/segment_seconds —
|
||
глобальные: срок хранения по возрасту и длина сегмента для всех хранилищ.
|
||
"""
|
||
active: str = "local"
|
||
retention_days: int = 7
|
||
segment_seconds: int = 600
|
||
items: tuple[StorageItem, ...] = ()
|
||
|
||
|
||
# Дефолтное хранилище (когда конфиг пуст / битый) — совпадает с историческим /records.
|
||
_DEFAULT_STORAGE = StorageItem(
|
||
id="local", name="Локальное хранилище", type="local", path="/records", quota_gb=30,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Live:
|
||
go2rtc_url: str = "http://localhost:1984"
|
||
|
||
|
||
@dataclass
|
||
class Config:
|
||
storage: Storage
|
||
live: Live
|
||
cameras: list[Camera] = field(default_factory=list)
|
||
|
||
def camera(self, cam_id: str) -> Camera | None:
|
||
return next((c for c in self.cameras if c.id == cam_id), None)
|
||
|
||
@property
|
||
def enabled_cameras(self) -> list[Camera]:
|
||
return [c for c in self.cameras if c.enabled]
|
||
|
||
# ── слоты записи ────────────────────────────────────────────
|
||
# Папка камеры в хранилище = номер её слота (позиция в списке, 1-based) с
|
||
# ведущим нулём: слот 5 → «05», слот 12 → «12». Так совпадает со столбцом ID в UI.
|
||
def slot_of(self, cam_id: str) -> int | None:
|
||
for i, c in enumerate(self.cameras, start=1):
|
||
if c.id == cam_id:
|
||
return i
|
||
return None
|
||
|
||
def archive_dir(self, cam_id: str) -> str:
|
||
"""Имя подпапки записи камеры в хранилище. Откат на id, если камеры нет в списке."""
|
||
slot = self.slot_of(cam_id)
|
||
return f"{slot:02d}" if slot is not None else cam_id
|
||
|
||
def camera_by_archive_dir(self, name: str) -> Camera | None:
|
||
"""Обратное: имя папки-слота («05») → камера в этом слоте. None для нечисловых
|
||
(легаси-папки по старому id индексируются под своим именем как раньше)."""
|
||
if name.isdigit():
|
||
idx = int(name)
|
||
if 1 <= idx <= len(self.cameras):
|
||
return self.cameras[idx - 1]
|
||
return None
|
||
|
||
# ── хранилища ───────────────────────────────────────────────
|
||
def storage_by_id(self, sid: str) -> StorageItem | None:
|
||
return next((s for s in self.storage.items if s.id == sid), None)
|
||
|
||
def active_storage(self) -> StorageItem:
|
||
"""Хранилище, куда пишутся новые записи (fallback — первый/дефолтный)."""
|
||
return (
|
||
self.storage_by_id(self.storage.active)
|
||
or (self.storage.items[0] if self.storage.items else _DEFAULT_STORAGE)
|
||
)
|
||
|
||
def storage_paths(self) -> list[str]:
|
||
"""Пути всех хранилищ — индексатор сканирует их все (старые записи не теряются)."""
|
||
return [s.path for s in self.storage.items] or [_DEFAULT_STORAGE.path]
|
||
|
||
|
||
def load_config(path: str | None = None) -> Config:
|
||
path = path or os.environ.get("NVR_CONFIG", "cameras.yaml")
|
||
# Толерантно к отсутствию/повреждению: если файла нет, на его месте каталог
|
||
# (битый bind-mount) или YAML невалиден — стартуем без камер (добавляются из UI),
|
||
# а не падаем с ошибкой на старте.
|
||
raw: dict = {}
|
||
if os.path.isfile(path):
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
raw = yaml.safe_load(fh) or {}
|
||
except (OSError, yaml.YAMLError):
|
||
raw = {}
|
||
|
||
storage = _storage_from_raw(raw.get("storage") or {})
|
||
live = Live(**(raw.get("live") or {}))
|
||
cameras = [_camera_from_raw(c) for c in (raw.get("cameras") or [])]
|
||
return Config(storage=storage, live=live, cameras=cameras)
|
||
|
||
|
||
def _camera_from_raw(c: dict) -> Camera:
|
||
# Миграция со старой схемы: раньше `enabled` означал «писать на диск» (live был всегда),
|
||
# отдельного «камера выключена» не существовало. Старый файл (без ключа record) трактуем
|
||
# так: запись = старый enabled, а сама камера включена. Новый файл (с record) — как есть.
|
||
if "record" in c:
|
||
enabled, record = bool(c.get("enabled", True)), bool(c["record"])
|
||
else:
|
||
enabled, record = True, bool(c.get("enabled", True))
|
||
return Camera(
|
||
id=c["id"],
|
||
name=c.get("name", c["id"]),
|
||
enabled=enabled,
|
||
record=record,
|
||
ip=c["ip"],
|
||
user=c["user"],
|
||
password=c["password"],
|
||
main_path=c["main_path"],
|
||
sub_path=c.get("sub_path"),
|
||
storage_id=c.get("storage_id", "") or "",
|
||
)
|
||
|
||
|
||
def _storage_item_from_raw(s: dict) -> StorageItem:
|
||
return StorageItem(
|
||
id=s["id"],
|
||
name=s.get("name", s["id"]),
|
||
type=s.get("type", "local"),
|
||
path=s["path"],
|
||
quota_gb=int(s.get("quota_gb", 0) or 0),
|
||
server=s.get("server", "") or "",
|
||
share=s.get("share", "") or "",
|
||
username=s.get("username", "") or "",
|
||
password=s.get("password", "") or "",
|
||
subpath=s.get("subpath", "") or "",
|
||
enabled=bool(s.get("enabled", True)),
|
||
)
|
||
|
||
|
||
def _storage_from_raw(raw: dict) -> Storage:
|
||
# Новая схема: есть список items.
|
||
if raw.get("items"):
|
||
items = tuple(_storage_item_from_raw(s) for s in raw["items"])
|
||
active = raw.get("active") or (items[0].id if items else "local")
|
||
return Storage(
|
||
active=active,
|
||
retention_days=int(raw.get("retention_days", 7)),
|
||
segment_seconds=int(raw.get("segment_seconds", 600)),
|
||
items=items,
|
||
)
|
||
# Миграция со старой одиночной схемы {path, quota_gb, retention_days, segment_seconds}:
|
||
# один локальный item, старый глобальный quota → квота этого хранилища.
|
||
legacy_path = raw.get("path") or "/records"
|
||
legacy_quota = int(raw.get("quota_gb", 30) or 30)
|
||
local = StorageItem(
|
||
id="local", name="Локальное хранилище", type="local",
|
||
path=legacy_path, quota_gb=legacy_quota,
|
||
)
|
||
return Storage(
|
||
active="local",
|
||
retention_days=int(raw.get("retention_days", 7)),
|
||
segment_seconds=int(raw.get("segment_seconds", 600)),
|
||
items=(local,),
|
||
)
|
||
|
||
|
||
def camera_to_yaml(c: Camera) -> dict:
|
||
return {
|
||
"id": c.id,
|
||
"name": c.name,
|
||
"enabled": c.enabled,
|
||
"record": c.record,
|
||
"ip": c.ip,
|
||
"user": c.user,
|
||
"password": c.password,
|
||
"main_path": c.main_path,
|
||
"sub_path": c.sub_path,
|
||
"storage_id": c.storage_id,
|
||
}
|
||
|
||
|
||
def storage_item_to_yaml(s: StorageItem) -> dict:
|
||
d = {
|
||
"id": s.id, "name": s.name, "type": s.type,
|
||
"path": s.path, "quota_gb": s.quota_gb,
|
||
}
|
||
if s.type == "nas":
|
||
d.update({
|
||
"server": s.server, "share": s.share, "username": s.username,
|
||
"password": s.password, "subpath": s.subpath,
|
||
})
|
||
if not s.enabled:
|
||
d["enabled"] = False
|
||
return d
|
||
|
||
|
||
def save_config(config: Config, path: str | None = None) -> None:
|
||
"""Сохраняет конфиг обратно в cameras.yaml (комментарии теряются)."""
|
||
path = path or os.environ.get("NVR_CONFIG", "cameras.yaml")
|
||
data = {
|
||
"storage": {
|
||
"active": config.storage.active,
|
||
"retention_days": config.storage.retention_days,
|
||
"segment_seconds": config.storage.segment_seconds,
|
||
"items": [storage_item_to_yaml(s) for s in config.storage.items],
|
||
},
|
||
"live": {"go2rtc_url": config.live.go2rtc_url},
|
||
"cameras": [camera_to_yaml(c) for c in config.cameras],
|
||
}
|
||
# пишем на месте: cameras.yaml примонтирован как отдельный файл (bind mount),
|
||
# переименование поверх точки монтирования даёт EBUSY.
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
yaml.safe_dump(data, fh, allow_unicode=True, sort_keys=False)
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def get_config() -> Config:
|
||
return load_config()
|