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