git_admin/CoRE.Vision
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

storage: multi-storage management with per-storage quota

Browse Source 
Replace the single fixed /records path with a list of storages and an
active selection (where new recordings go). Quota is now per-storage.

- config: StorageItem(id,name,type,path,quota_gb,nas-meta) + Storage keeps
  global retention_days/segment_seconds; active + items. Migration from the
  old single {path,quota_gb} schema → one local item.
- recorder: write to the active storage; restart_all() repoints on switch;
  NAS-not-mounted guard (never write into the container overlay).
- indexer: scan ALL storage roots so recordings survive an active switch.
- retention: enforce quota PER storage via db.total_size_under /
  oldest_recordings_under (path-prefix LIKE).
- api/storages: GET/POST/PUT/DELETE + activate + policy. NAS uses host-mount
  (backend stores SMB params, generates the mount/fstab command, reports
  ismount status). Retire old POST /api/storage.
- frontend: new "Хранилища" settings tab — table with inline per-row quota,
  add form (local folder / NAS SMB), global policy (segment + retention).
- compose: add shared ./storages:/storages root for extra local folders and
  host-mounted NAS shares.

VERSION 0.0.127.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
core
2026-06-01 19:31:15 +05:00
parent 24155a8492
commit a64749ccc8
16 changed files with 746 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/storages.py
+235
View File
@@ -0,0 +1,235 @@
"""REST: управление хранилищами записи — список, CRUD, выбор активного, политика.
Несколько хранилищ (local-папка или NAS/SMB) + одно активное (куда пишем новые записи).
Квота — на каждом хранилище. NAS монтируется на ХОСТЕ (backend не монтирует сам), здесь
лишь хранятся параметры и генерируется готовая команда mount/fstab."""
from __future__ import annotations
import os
import re
from dataclasses import replace
from fastapi import APIRouter, HTTPException, Request
from ..auth import require_role
from ..config import StorageItem, save_config
from ..runtime import runtime
from ..services.system import is_mounted, storage_disk
router = APIRouter(prefix="/api/storages", tags=["storages"])
ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,32}$")
STORAGES_ROOT = "/storages" # общий bind-mount корень для доп. хранилищ
HOST_STORAGES_ROOT = "/opt/nvr/storages" # он же на хосте (для команды монтирования NAS)
ALLOWED_SEGMENTS = (60, 120, 300, 600, 900) # 1 / 2 / 5 / 10 / 15 минут
# ── helpers ─────────────────────────────────────────────────────
async def _item_payload(item: StorageItem, active: str) -> dict:
nas = item.type == "nas"
return {
"id": item.id,
"name": item.name,
"type": item.type,
"path": item.path,
"quota_gb": item.quota_gb,
"server": item.server,
"share": item.share,
"username": item.username,
"password": item.password,
"subpath": item.subpath,
"enabled": item.enabled,
"active": item.id == active,
"mounted": is_mounted(item.path) if nas else True,
"used_bytes": await runtime.db.total_size_under(item.path),
"disk": storage_disk(item.path),
}
def _nas_mount_help(item: StorageItem) -> dict:
"""Готовая команда монтирования SMB на хосте + строка fstab (монтаж — на хосте)."""
host_path = f"{HOST_STORAGES_ROOT}/{item.id}"
src = f"//{item.server}/{item.share}"
if item.subpath:
src += "/" + item.subpath.strip("/")
cred = f"username={item.username},password={item.password}"
opts = f"{cred},uid=0,gid=0,vers=3.0,iocharset=utf8"
return {
"host_path": host_path,
"mount_cmd": f"sudo mkdir -p {host_path} && sudo mount -t cifs {src} {host_path} -o {opts}",
"fstab": f"{src} {host_path} cifs {opts},_netdev,nofail 0 0",
"note": "После монтирования на хосте выполните `docker compose up -d` (или включите "
"rshared-проброс), чтобы том стал виден в контейнере.",
}
# ── список ──────────────────────────────────────────────────────
@router.get("")
async def list_storages() -> dict:
cfg = runtime.config
active = cfg.storage.active
items = [await _item_payload(s, active) for s in cfg.storage.items]
return {
"storages": items,
"active": active,
"retention_days": cfg.storage.retention_days,
"segment_seconds": cfg.storage.segment_seconds,
}
# ── политика (глобально): срок хранения и длина сегмента ─────────
@router.post("/policy")
async def update_policy(request: Request) -> dict:
require_role(request, "admin")
body = await request.json()
st = runtime.config.storage
kw: dict = {}
seg_changed = False
if "segment_seconds" in body:
try:
seg = int(body["segment_seconds"])
except (TypeError, ValueError):
raise HTTPException(400, "segment_seconds должен быть числом")
if seg not in ALLOWED_SEGMENTS:
raise HTTPException(400, "допустимо 60 / 120 / 300 / 600 / 900 секунд")
kw["segment_seconds"] = seg
seg_changed = seg != st.segment_seconds
if "retention_days" in body:
try:
rd = int(body["retention_days"])
except (TypeError, ValueError):
raise HTTPException(400, "retention_days должен быть числом")
if rd < 0 or rd > 3650:
raise HTTPException(400, "retention_days в диапазоне 0..3650")
kw["retention_days"] = rd
if not kw:
raise HTTPException(400, "ожидается segment_seconds и/или retention_days")
runtime.config.storage = replace(st, **kw)
save_config(runtime.config)
if seg_changed:
await runtime.supervisor.restart_all()
return {
"ok": True,
"segment_seconds": runtime.config.storage.segment_seconds,
"retention_days": runtime.config.storage.retention_days,
}
# ── CRUD ────────────────────────────────────────────────────────
@router.post("")
async def create_storage(request: Request) -> dict:
require_role(request, "admin")
body = await request.json()
sid = (body.get("id") or "").strip()
if not ID_RE.match(sid):
raise HTTPException(400, "id: разрешены латиница, цифры, _ и - (до 32 символов)")
if runtime.config.storage_by_id(sid):
raise HTTPException(409, "хранилище с таким id уже существует")
stype = body.get("type", "local")
if stype not in ("local", "nas"):
raise HTTPException(400, "type: local | nas")
name = (body.get("name") or sid).strip()
quota = max(0, int(body.get("quota_gb") or 0))
helper = None
if stype == "local":
# путь = подпапка общего корня /storages/<id>; допускаем явный path (напр. /records)
path = (body.get("path") or "").strip() or f"{STORAGES_ROOT}/{sid}"
item = StorageItem(id=sid, name=name, type="local", path=path, quota_gb=quota)
try:
os.makedirs(path, exist_ok=True)
except OSError as exc:
raise HTTPException(400, f"не удалось создать папку {path}: {exc}")
else:
server = (body.get("server") or "").strip()
share = (body.get("share") or "").strip()
if not server or not share:
raise HTTPException(400, "для NAS обязательны server и share")
path = f"{STORAGES_ROOT}/{sid}"
item = StorageItem(
id=sid, name=name, type="nas", path=path, quota_gb=quota,
server=server, share=share,
username=(body.get("username") or "").strip(),
password=body.get("password") or "",
subpath=(body.get("subpath") or "").strip(),
)
# создаём точку монтирования на хосте (через bind-mount /storages) — писать туда
# рекордер начнёт только когда SMB реально смонтируют (guard по os.path.ismount)
try:
os.makedirs(path, exist_ok=True)
except OSError:
pass
helper = _nas_mount_help(item)
new_items = runtime.config.storage.items + (item,)
runtime.config.storage = replace(runtime.config.storage, items=new_items)
save_config(runtime.config)
out = await _item_payload(item, runtime.config.storage.active)
if helper:
out["mount_help"] = helper
return out
@router.put("/{sid}")
async def update_storage(sid: str, request: Request) -> dict:
require_role(request, "admin")
item = runtime.config.storage_by_id(sid)
if not item:
raise HTTPException(404, "storage not found")
body = await request.json()
kw: dict = {}
if "name" in body:
kw["name"] = (body.get("name") or item.name).strip()
if "quota_gb" in body:
kw["quota_gb"] = max(0, int(body.get("quota_gb") or 0))
if "enabled" in body:
kw["enabled"] = bool(body["enabled"])
for f in ("server", "share", "username", "subpath"): # метаданные NAS
if f in body:
kw[f] = (body.get(f) or "").strip()
if "password" in body:
kw["password"] = body.get("password") or ""
if not kw:
raise HTTPException(400, "нет полей для изменения")
# id/type/path неизменны (привязаны к записям на диске → не ломаем архив)
new = replace(item, **kw)
items = list(runtime.config.storage.items)
idx = next(i for i, s in enumerate(items) if s.id == sid)
items[idx] = new
runtime.config.storage = replace(runtime.config.storage, items=tuple(items))
save_config(runtime.config)
# quota применится при следующем проходе retention; path не менялся → рекордеры не трогаем
out = await _item_payload(new, runtime.config.storage.active)
if new.type == "nas":
out["mount_help"] = _nas_mount_help(new)
return out
@router.delete("/{sid}")
async def delete_storage(sid: str, request: Request) -> dict:
"""Убирает хранилище из списка. Файлы и записи в БД НЕ трогаются (архив сохраняется)."""
require_role(request, "admin")
if not runtime.config.storage_by_id(sid):
raise HTTPException(404, "storage not found")
if sid == runtime.config.storage.active:
raise HTTPException(409, "нельзя удалить активное хранилище — сначала выберите другое")
if len(runtime.config.storage.items) <= 1:
raise HTTPException(409, "нельзя удалить последнее хранилище")
items = tuple(s for s in runtime.config.storage.items if s.id != sid)
runtime.config.storage = replace(runtime.config.storage, items=items)
save_config(runtime.config)
return {"ok": True}
@router.post("/{sid}/activate")
async def activate_storage(sid: str, request: Request) -> dict:
"""Делает хранилище активным (новые записи идут туда). Перезапускает рекордеры."""
require_role(request, "admin")
item = runtime.config.storage_by_id(sid)
if not item:
raise HTTPException(404, "storage not found")
runtime.config.storage = replace(runtime.config.storage, active=sid)
save_config(runtime.config)
await runtime.supervisor.restart_all() # репойнт рекордеров на новый путь
mounted = is_mounted(item.path) if item.type == "nas" else True
return {"ok": True, "active": sid, "mounted": mounted}
backend/app/api/system.py
+1 -26
View File
@@ -3,16 +3,13 @@ from __future__ import annotations
import os
from fastapi import APIRouter, HTTPException, Request
from fastapi import APIRouter
from ..auth import require_role
from ..config import Storage, save_config
from ..runtime import runtime
from ..services.system import system_stats
router = APIRouter(prefix="/api", tags=["system"])
ALLOWED_SEGMENTS = (60, 120, 300, 600, 900) # 1 / 2 / 5 / 10 / 15 минут
VERSION_FILE = os.environ.get("NVR_VERSION_FILE", "/app/VERSION")
@@ -36,25 +33,3 @@ async def get_system() -> dict:
@router.get("/health")
async def health() -> dict:
return {"status": "ok"}
@router.post("/storage")
async def update_storage(request: Request) -> dict:
"""Смена размера сегмента записи (1/5/10/15 мин). Перезапускает рекордеры."""
require_role(request, "admin")
body = await request.json()
try:
seg = int(body.get("segment_seconds"))
except (TypeError, ValueError):
raise HTTPException(400, "segment_seconds должен быть числом")
if seg not in ALLOWED_SEGMENTS:
raise HTTPException(400, "допустимо 60 / 300 / 600 / 900 секунд")
st = runtime.config.storage
runtime.config.storage = Storage(
path=st.path, quota_gb=st.quota_gb,
retention_days=st.retention_days, segment_seconds=seg,
)
save_config(runtime.config)
await runtime.supervisor.restart_all()
return {"segment_seconds": seg}
backend/app/config.py
+111 -5
View File
@@ -29,12 +29,45 @@ class Camera:
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:
path: str = "/records"
quota_gb: int = 50
"""Глобальная политика записи + список хранилищ и 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)
@@ -55,6 +88,21 @@ class Config:
def enabled_cameras(self) -> list[Camera]:
return [c for c in self.cameras if c.enabled]
# ── хранилища ───────────────────────────────────────────────
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")
@@ -69,7 +117,7 @@ def load_config(path: str | None = None) -> Config:
except (OSError, yaml.YAMLError):
raw = {}
storage = Storage(**(raw.get("storage") or {}))
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)
@@ -96,6 +144,49 @@ def _camera_from_raw(c: dict) -> Camera:
)
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,
@@ -110,15 +201,30 @@ def camera_to_yaml(c: Camera) -> dict:
}
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": {
"path": config.storage.path,
"quota_gb": config.storage.quota_gb,
"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],
backend/app/db.py
+26
View File
@@ -8,6 +8,14 @@ import aiosqlite
from .models import Recording
def _like_prefix(prefix: str) -> str:
"""LIKE-шаблон для путей внутри хранилища: '<path>/%' с экранированием %, _, \\."""
p = prefix.rstrip("/") + "/"
p = p.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
return p + "%"
SCHEMA = """
CREATE TABLE IF NOT EXISTS recordings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -105,6 +113,24 @@ class Database:
rows = await cur.fetchall()
return [Recording(**dict(r)) for r in rows]
# ── per-storage (по префиксу пути) — для квоты на каждое хранилище ──
async def total_size_under(self, prefix: str) -> int:
cur = await self.conn.execute(
"SELECT COALESCE(SUM(size_bytes), 0) FROM recordings WHERE path LIKE ? ESCAPE '\\'",
(_like_prefix(prefix),),
)
row = await cur.fetchone()
return int(row[0]) if row else 0
async def oldest_recordings_under(self, prefix: str, limit: int = 50) -> list[Recording]:
cur = await self.conn.execute(
"SELECT * FROM recordings WHERE path LIKE ? ESCAPE '\\' "
"ORDER BY started_at ASC LIMIT ?",
(_like_prefix(prefix), limit),
)
rows = await cur.fetchall()
return [Recording(**dict(r)) for r in rows]
async def recordings_before(self, ts: int) -> list[Recording]:
cur = await self.conn.execute(
"SELECT * FROM recordings WHERE started_at < ? ORDER BY started_at ASC", (ts,)
backend/app/main.py
+8 -4
View File
@@ -17,7 +17,7 @@ from .recorder.supervisor import Supervisor
from .services.retention import Retention
from .runtime import runtime
from . import auth, prefs
from .api import cameras, recordings, system, ws
from .api import cameras, recordings, storages, system, ws
from .api.ws import ConnectionManager
from .services import go2rtc
@@ -46,15 +46,18 @@ async def lifespan(app: FastAPI):
runtime.indexer = Indexer(runtime.config, runtime.db)
runtime.retention = Retention(runtime.config, runtime.db)
os.makedirs(runtime.config.storage.path, exist_ok=True)
# каталоги локальных хранилищ; NAS не создаём — ждём монтаж на хосте (guard в recorder_task)
for st in runtime.config.storage.items:
if st.type == "local":
os.makedirs(st.path, exist_ok=True)
await go2rtc.sync_all(runtime.config) # потоки (вкл. <cam>_main) в go2rtc для рестрима
runtime.reconciler = go2rtc.Reconciler(runtime.config) # самовосстановление потоков go2rtc
runtime.supervisor.start_all()
runtime.indexer.start()
runtime.retention.start()
runtime.reconciler.start()
log.info("NVR запущен: камер=%d, хранилище=%s",
len(runtime.config.cameras), runtime.config.storage.path)
log.info("NVR запущен: камер=%d, активное хранилище=%s",
len(runtime.config.cameras), runtime.config.active_storage().path)
try:
yield
@@ -99,6 +102,7 @@ app.include_router(auth.router)
app.include_router(prefs.router)
app.include_router(cameras.router)
app.include_router(recordings.router)
app.include_router(storages.router)
app.include_router(system.router)
app.include_router(ws.router)
backend/app/recorder/ffmpeg.py
+6 -4
View File
@@ -5,19 +5,21 @@ import asyncio
import json
import os
from ..config import Camera, Storage
from ..config import Camera
def record_args(cam: Camera, storage: Storage, input_url: str | None = None) -> list[str]:
def record_args(cam: Camera, out_root: str, segment_seconds: int,
input_url: str | None = None) -> list[str]:
"""ffmpeg для записи основного потока сегментами mp4 копированием.
-c copy => CPU почти не используется. -strftime даёт имена сегментов по времени.
out_root — корень активного хранилища (внутри пишем подкаталог камеры).
input_url — источник (по умолчанию рестрим go2rtc: камера держит одну сессию).
"""
# Плоско в каталоге камеры: дата в имени файла. Каталог создаётся один раз
# (recorder_task), новых подкаталогов не нужно — корректно переживает полночь.
# Подкаталог даты не используем: этот ffmpeg-билд не знает -strftime_mkdir.
out_template = os.path.join(storage.path, cam.id, "%Y-%m-%d_%H-%M-%S.mp4")
out_template = os.path.join(out_root, cam.id, "%Y-%m-%d_%H-%M-%S.mp4")
return [
"ffmpeg",
"-nostdin",
@@ -29,7 +31,7 @@ def record_args(cam: Camera, storage: Storage, input_url: str | None = None) ->
"-an", # звук не пишем (камеры обычно без него)
"-c:v", "copy",
"-f", "segment",
"-segment_time", str(storage.segment_seconds),
"-segment_time", str(segment_seconds),
"-segment_atclocktime", "1", # резать по часам: кратно segment_time от 00:00
"-segment_clocktime_offset", "0",
"-segment_format", "mp4",
backend/app/recorder/indexer.py
+19 -15
View File
@@ -80,24 +80,28 @@ class Indexer:
await asyncio.sleep(SCAN_INTERVAL)
async def scan_once(self) -> int:
root = self.config.storage.path
by_cam = _scan_segments(root)
active_path = self.config.active_storage().path
known = await self.db.known_paths()
added = 0
for cam_id, files in by_cam.items():
# последний файл может писаться прямо сейчас — не индексируем
complete = files[:-1] if len(files) > 1 else []
for path in complete:
if path in known:
continue
duration, size = await probe_duration(path)
if size is None or size == 0:
continue
await self.db.add_recording(
cam_id, path, _parse_started_at(path), duration, size
)
added += 1
# сканируем ВСЕ хранилища: после смены активного старые записи не теряются
for root in self.config.storage_paths():
by_cam = _scan_segments(root)
is_active = (root == active_path)
for cam_id, files in by_cam.items():
# в активном хранилище последний файл ещё пишется — пропускаем;
# в неактивных все сегменты завершены, индексируем целиком.
complete = files[:-1] if is_active else files
for path in complete:
if path in known:
continue
duration, size = await probe_duration(path)
if size is None or size == 0:
continue
await self.db.add_recording(
cam_id, path, _parse_started_at(path), duration, size
)
added += 1
# уборка пропавших файлов
for path in known:
backend/app/recorder/recorder_task.py
+23 -5
View File
@@ -7,7 +7,7 @@ import os
import time
from collections import deque
from ..config import Camera, Storage
from ..config import Camera, StorageItem
from ..models import CameraState, CameraStatus
from .ffmpeg import record_args
@@ -21,9 +21,11 @@ MIN_HEALTHY_UPTIME = 15.0 # процесс прожил столько => сч
class RecorderTask:
"""Держит один процесс ffmpeg на камеру, перезапускает при падении."""
def __init__(self, cam: Camera, storage: Storage, on_status=None, input_url: str | None = None):
def __init__(self, cam: Camera, store: StorageItem, segment_seconds: int,
on_status=None, input_url: str | None = None):
self.cam = cam
self.storage = storage
self.store = store # активное хранилище (куда пишем)
self.segment_seconds = segment_seconds
self.input_url = input_url # источник записи (рестрим go2rtc)
self.on_status = on_status # callback(CameraStatus) для WS-уведомлений
self.status = CameraStatus(
@@ -69,11 +71,27 @@ class RecorderTask:
# ── основной цикл ───────────────────────────────────────────
async def _run(self) -> None:
backoff = BACKOFF_START
os.makedirs(os.path.join(self.storage.path, self.cam.id), exist_ok=True)
while not self._stopping:
# NAS: пишем только если том реально смонтирован на хосте. Иначе запись
# ушла бы в overlay контейнера (потеря при рестарте) — ждём монтаж, каталог
# тоже не создаём.
if self.store.type == "nas" and not os.path.ismount(self.store.path):
self._set_state(CameraState.RETRYING, "NAS не смонтирован")
backoff = min(backoff * 2, BACKOFF_MAX)
await asyncio.sleep(backoff)
continue
try:
os.makedirs(os.path.join(self.store.path, self.cam.id), exist_ok=True)
except OSError as exc:
self._set_state(CameraState.RETRYING, f"mkdir: {exc}")
backoff = min(backoff * 2, BACKOFF_MAX)
await asyncio.sleep(backoff)
continue
self._set_state(CameraState.STARTING)
args = record_args(self.cam, self.storage, self.input_url)
args = record_args(self.cam, self.store.path, self.segment_seconds, self.input_url)
log.info("[%s] start ffmpeg: %s", self.cam.id, _redact(args, self.cam))
try:
self._proc = await asyncio.create_subprocess_exec(
backend/app/recorder/supervisor.py
+8 -4
View File
@@ -21,10 +21,15 @@ class Supervisor:
# запись основного потока идёт через рестрим go2rtc (одна сессия к камере)
return go2rtc.restream_main_url(self.config, cam.id)
def _new_task(self, cam: Camera) -> RecorderTask:
# пишем в активное хранилище; смена активного → restart_all репойнтит все задачи
active = self.config.active_storage()
return RecorderTask(cam, active, self.config.storage.segment_seconds,
on_status=self.on_status, input_url=self._input_url(cam))
def start_all(self) -> None:
for cam in self.config.cameras:
task = RecorderTask(cam, self.config.storage, on_status=self.on_status,
input_url=self._input_url(cam))
task = self._new_task(cam)
self.tasks[cam.id] = task
if cam.enabled and cam.record:
task.start()
@@ -57,8 +62,7 @@ class Supervisor:
def add_camera(self, cam: Camera) -> None:
"""Создаёт задачу записи для новой камеры и запускает (если включена)."""
task = RecorderTask(cam, self.config.storage, on_status=self.on_status,
input_url=self._input_url(cam))
task = self._new_task(cam)
self.tasks[cam.id] = task
if cam.enabled and cam.record:
task.start()
backend/app/services/retention.py
+14 -12
View File
@@ -61,20 +61,22 @@ class Retention:
return len(old)
async def _enforce_quota(self) -> int:
quota = self.config.storage.quota_gb * 1024 ** 3
if not quota:
return 0
# квота — на КАЖДОЕ хранилище отдельно (0 = без лимита). Для каждого удаляем
# самые старые записи ВНУТРИ него, пока его размер не уложится в его квоту.
removed = 0
# удаляем самые старые, пока суммарный размер не уложится в квоту
while await self.db.total_size() > quota:
batch = await self.db.oldest_recordings(limit=20)
if not batch:
break
for rec in batch:
await self._remove(rec.id, rec.path)
removed += 1
if await self.db.total_size() <= quota:
for item in self.config.storage.items:
if not item.quota_gb:
continue
quota = item.quota_gb * 1024 ** 3
while await self.db.total_size_under(item.path) > quota:
batch = await self.db.oldest_recordings_under(item.path, limit=20)
if not batch:
break
for rec in batch:
await self._remove(rec.id, rec.path)
removed += 1
if await self.db.total_size_under(item.path) <= quota:
break
return removed
async def _remove(self, rec_id: int, path: str) -> None:
backend/app/services/system.py
+19 -6
View File
@@ -1,6 +1,7 @@
"""Системная статистика: диск хранилища, CPU, RAM, аптайм."""
from __future__ import annotations
import os
import shutil
import time
@@ -11,19 +12,31 @@ from ..config import Config
_START = time.time()
def system_stats(config: Config) -> dict:
path = config.storage.path
def storage_disk(path: str) -> dict:
"""Использование диска, на котором лежит path (для одного хранилища)."""
try:
usage = shutil.disk_usage(path)
disk = {
return {
"total_gb": round(usage.total / 1024 ** 3, 1),
"used_gb": round(usage.used / 1024 ** 3, 1),
"free_gb": round(usage.free / 1024 ** 3, 1),
"percent": round(usage.used / usage.total * 100, 1) if usage.total else 0,
}
except OSError:
disk = {"total_gb": 0, "used_gb": 0, "free_gb": 0, "percent": 0}
return {"total_gb": 0, "used_gb": 0, "free_gb": 0, "percent": 0}
def is_mounted(path: str) -> bool:
"""Точка монтирования? (для NAS — признак, что том реально подключён)."""
try:
return os.path.ismount(path)
except OSError:
return False
def system_stats(config: Config) -> dict:
active = config.active_storage()
path = active.path
vm = psutil.virtual_memory()
return {
"uptime_s": int(time.time() - _START),
@@ -32,8 +45,8 @@ def system_stats(config: Config) -> dict:
"ram_used_gb": round(vm.used / 1024 ** 3, 1),
"ram_total_gb": round(vm.total / 1024 ** 3, 1),
"storage_path": path,
"disk": disk,
"quota_gb": config.storage.quota_gb,
"disk": storage_disk(path),
"quota_gb": active.quota_gb,
"retention_days": config.storage.retention_days,
"segment_seconds": config.storage.segment_seconds,
}