123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
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_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:
|
|
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:
|
|
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]:
|
|
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:
|
|
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]]:
|
|
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:
|
|
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 |