132 lines
5.2 KiB
Python
132 lines
5.2 KiB
Python
"""Сервисы для работы с камерами и storage: превью, автообнаружение, утилиты."""
|
||
|
||
import logging
|
||
from pathlib import Path
|
||
|
||
from django.conf import settings
|
||
from django.utils.text import slugify
|
||
|
||
from ..models import Camera
|
||
|
||
|
||
logger = logging.getLogger('camlaps')
|
||
|
||
|
||
def _storage_root() -> Path:
|
||
"""Возвращает корень storage, в который смонтированы фото камер."""
|
||
storage_path = getattr(settings, 'STORAGE_PATH', None)
|
||
if storage_path is None:
|
||
return Path('/app/storage')
|
||
return Path(storage_path)
|
||
|
||
|
||
def generate_unique_camera_slug(directory_name: str, storage_path: str) -> str:
|
||
"""Генерирует уникальный slug камеры для новых записей, избегая конфликтов."""
|
||
base_slug = (slugify(directory_name) or f'camera-{directory_name.lower()}')[:70]
|
||
slug = base_slug
|
||
suffix = 1
|
||
|
||
while Camera.objects.filter(slug=slug).exclude(storage_path=storage_path).exists():
|
||
suffix_part = f'-{suffix}'
|
||
slug = f"{base_slug[:80 - len(suffix_part)]}{suffix_part}"
|
||
suffix += 1
|
||
|
||
return slug
|
||
|
||
|
||
def is_storage_available() -> bool:
|
||
"""Проверяет, что storage примонтирован и доступен как директория."""
|
||
logger.info('cameras:storage_available:start')
|
||
try:
|
||
root = _storage_root()
|
||
ok = root.exists() and root.is_dir()
|
||
logger.info('cameras:storage_available:done ok=%s', ok)
|
||
return ok
|
||
except Exception:
|
||
logger.exception('cameras:storage_available:error')
|
||
raise
|
||
|
||
|
||
def list_active_cameras() -> list[Camera]:
|
||
"""Возвращает список активных камер для отображения в UI."""
|
||
logger.info('cameras:list_active:start')
|
||
try:
|
||
cameras = list(Camera.objects.filter(is_active=True).order_by('name'))
|
||
logger.info('cameras:list_active:done count=%s', len(cameras))
|
||
return cameras
|
||
except Exception:
|
||
logger.exception('cameras:list_active:error')
|
||
raise
|
||
|
||
|
||
def get_camera_lastsnap_path(camera: Camera) -> Path | None:
|
||
"""Возвращает путь к lastsnap.jpg для камеры или None, если файла нет/путь небезопасен."""
|
||
logger.info('cameras:lastsnap_path:start camera_id=%s', camera.id)
|
||
try:
|
||
root = _storage_root().resolve()
|
||
candidate = (root / camera.storage_path / 'lastsnap.jpg').resolve()
|
||
|
||
if candidate != root and root not in candidate.parents:
|
||
logger.info('cameras:lastsnap_path:done camera_id=%s found=false', camera.id)
|
||
return None
|
||
|
||
if not candidate.exists() or not candidate.is_file():
|
||
logger.info('cameras:lastsnap_path:done camera_id=%s found=false', camera.id)
|
||
return None
|
||
|
||
logger.info('cameras:lastsnap_path:done camera_id=%s found=true', camera.id)
|
||
return candidate
|
||
except Exception:
|
||
logger.exception('cameras:lastsnap_path:error camera_id=%s', camera.id)
|
||
raise
|
||
|
||
|
||
def discover_camera_candidates() -> list[dict[str, str]]:
|
||
"""Сканирует подпапки storage и возвращает кандидатов камер без записи в БД."""
|
||
logger.info('cameras:discover_candidates:start')
|
||
try:
|
||
root = _storage_root()
|
||
if not root.exists() or not root.is_dir():
|
||
logger.info('cameras:discover_candidates:done count=0')
|
||
return []
|
||
|
||
existing_paths = set(Camera.objects.values_list('storage_path', flat=True))
|
||
dirs = [p for p in root.iterdir() if p.is_dir() and not p.name.startswith('.') and p.name != 'timelapses']
|
||
|
||
candidates = []
|
||
for directory in sorted(dirs, key=lambda d: d.name.lower()):
|
||
storage_path = directory.name
|
||
if storage_path in existing_paths:
|
||
continue
|
||
candidates.append({'name': directory.name, 'storage_path': storage_path})
|
||
|
||
logger.info('cameras:discover_candidates:done count=%s', len(candidates))
|
||
return candidates
|
||
except Exception:
|
||
logger.exception('cameras:discover_candidates:error')
|
||
raise
|
||
|
||
|
||
def create_cameras_from_candidates(selected_storage_paths: list[str]) -> int:
|
||
"""Создает записи Camera в БД только для выбранных storage_path, возвращает количество созданных."""
|
||
logger.info('cameras:create_from_candidates:start count=%s', len(selected_storage_paths))
|
||
try:
|
||
created = 0
|
||
for storage_path in selected_storage_paths:
|
||
path = storage_path.strip()
|
||
if not path:
|
||
continue
|
||
|
||
slug = generate_unique_camera_slug(path, path)
|
||
_, was_created = Camera.objects.get_or_create(
|
||
storage_path=path,
|
||
defaults={'name': path, 'slug': slug, 'is_active': True},
|
||
)
|
||
if was_created:
|
||
created += 1
|
||
|
||
logger.info('cameras:create_from_candidates:done created=%s', created)
|
||
return created
|
||
except Exception:
|
||
logger.exception('cameras:create_from_candidates:error')
|
||
raise |