"""This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license. """ import json import logging import os import re from datetime import datetime, timezone from pathlib import Path from urllib.parse import quote from django.conf import settings from django.core.exceptions import ValidationError from django.db import models from django.db.models.signals import post_save, pre_delete from django.dispatch import receiver from django.utils.translation import gettext_lazy as _ from io_storages.base_models import ( ExportStorage, ExportStorageLink, ImportStorage, ImportStorageLink, ProjectStorageMixin, ) from io_storages.localfiles.functions import normalize_storage_path from io_storages.utils import StorageObject, load_tasks_json from tasks.models import Annotation logger = logging.getLogger(__name__) class LocalFilesMixin(models.Model): path = models.TextField(_('path'), null=True, blank=True, help_text='Local path') regex_filter = models.TextField( _('regex_filter'), null=True, blank=True, help_text='Regex for filtering objects', ) use_blob_urls = models.BooleanField( _('use_blob_urls'), default=False, help_text='Interpret objects as BLOBs and generate URLs', ) def clean(self): super().clean() if self.path is not None: self.path = normalize_storage_path(self.path) def save(self, *args, **kwargs): if self.path is not None: self.path = normalize_storage_path(self.path) super().save(*args, **kwargs) def _get_storage_path_or_raise(self, exception_cls=ValueError) -> str: """Return a sanitized storage path or raise a caller-provided exception type. All downstream filesystem operations eventually need an absolute path without trailing slashes or mixed separators. Centralizing that logic here keeps the normalization rules consistent and allows callers to decide what exception class (ValidationError vs ValueError) should be raised when the path is missing or invalid. """ normalized = normalize_storage_path(self.path) if not normalized: raise exception_cls('Path must be set for Local Files storage') return normalized @staticmethod def community_auto_hint(): if settings.VERSION_EDITION == 'Community': return ( ' Community tip: create a "mydata" or "label-studio-data" directory next to the Label Studio ' 'command to auto-enable LOCAL_FILES_DOCUMENT_ROOT when the environment variables are unset.' ) return '' def validate_connection(self): normalized_path = self._get_storage_path_or_raise(ValidationError) self.path = normalized_path path = Path(normalized_path) document_root = Path(settings.LOCAL_FILES_DOCUMENT_ROOT) example_path = Path(settings.LOCAL_FILES_DOCUMENT_ROOT) / 'dataset1' if not path.exists(): raise ValidationError(f'Absolute local path "{self.path}" does not exist') if document_root == path: raise ValidationError( f'Absolute local path "{self.path}" cannot be the same as ' f'LOCAL_FILES_DOCUMENT_ROOT="{settings.LOCAL_FILES_DOCUMENT_ROOT}" by security reasons. Please add a subdirectory. ' f'For example: "{example_path}".' ) if document_root not in path.parents: raise ValidationError( f'Absolute local path "{self.path}" must be a subdirectory of ' f'LOCAL_FILES_DOCUMENT_ROOT="{settings.LOCAL_FILES_DOCUMENT_ROOT}" by security reasons. ' f'For example: "{example_path}".' ) if settings.LOCAL_FILES_SERVING_ENABLED is False: raise ValidationError( 'Serving local files from the host filesystem can be a security risk, so ' 'LOCAL_FILES_SERVING_ENABLED is disabled by default. ' 'To enable Local Files storage, set the LOCAL_FILES_SERVING_ENABLED environment ' 'variable to "true" and restart Label Studio. See ' 'https://labelstud.io/guide/storage.html#Local-storage for details.' '\n\n' f'{self.community_auto_hint()}' ) class LocalFilesImportStorageBase(LocalFilesMixin, ImportStorage): url_scheme = 'https' def can_resolve_url(self, url): return False recursive_scan = models.BooleanField( _('recursive scan'), default=False, db_default=False, null=True, help_text=_('Perform recursive scan over the directory content'), ) def iter_objects(self): path = Path(self._get_storage_path_or_raise()) regex = re.compile(str(self.regex_filter)) if self.regex_filter else None # For better control of imported tasks, file reading has been changed to ascending order of filenames. # In other words, the task IDs are sorted by filename order. iterator = path.rglob('*') if self.recursive_scan else path.glob('*') for file in sorted(iterator, key=os.path.basename): if file.is_file(): key = file.name if regex and not regex.match(key): logger.debug(key + ' is skipped by regex filter') continue yield file def iter_keys(self): for obj in self.iter_objects(): yield str(obj) def get_unified_metadata(self, obj): stat = obj.stat() return { 'key': str(obj), 'last_modified': datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc), 'size': stat.st_size, } def get_data(self, key) -> list[StorageObject]: path = Path(key) if self.use_blob_urls: # include self-hosted links pointed to local resources via # {settings.HOSTNAME}/data/local-files?d= document_root = Path(settings.LOCAL_FILES_DOCUMENT_ROOT) relative_path = str(path.relative_to(document_root)) task = { settings.DATA_UNDEFINED_NAME: f'{settings.HOSTNAME}/data/local-files/?d={quote(str(relative_path))}' } return [StorageObject(key=key, task_data=task)] try: with open(path, 'rb') as f: blob = f.read() return load_tasks_json(blob, key) except OSError as e: raise ValueError(f'Failed to read file {path}: {str(e)}') def scan_and_create_links(self): return self._scan_and_create_links(LocalFilesImportStorageLink) class Meta: abstract = True class LocalFilesImportStorage(ProjectStorageMixin, LocalFilesImportStorageBase): class Meta: abstract = False class LocalFilesExportStorage(LocalFilesMixin, ExportStorage): def save_annotation(self, annotation): logger.debug(f'Creating new object on {self.__class__.__name__} Storage {self} for annotation {annotation}') ser_annotation = self._get_serialized_data(annotation) # get key that identifies this object in storage key = LocalFilesExportStorageLink.get_key(annotation) storage_path = self._get_storage_path_or_raise() key = os.path.join(storage_path, f'{key}') # put object into storage with open(key, mode='w') as f: json.dump(ser_annotation, f, indent=2) # Create export storage link LocalFilesExportStorageLink.create(annotation, self) def delete_annotation(self, annotation): logger.debug(f'Deleting object on {self.__class__.__name__} Storage {self} for annotation {annotation}') key = LocalFilesExportStorageLink.get_key(annotation) storage_path = self._get_storage_path_or_raise() file_path = os.path.join(storage_path, f'{key}') try: os.remove(file_path) except FileNotFoundError: logger.warning(f'Export file {file_path} is already missing for annotation {annotation}') # nosec LocalFilesExportStorageLink.objects.filter(storage=self, annotation=annotation).delete() class LocalFilesImportStorageLink(ImportStorageLink): storage = models.ForeignKey(LocalFilesImportStorage, on_delete=models.CASCADE, related_name='links') class LocalFilesExportStorageLink(ExportStorageLink): storage = models.ForeignKey(LocalFilesExportStorage, on_delete=models.CASCADE, related_name='links') @receiver(post_save, sender=Annotation) def export_annotation_to_local_files(sender, instance, **kwargs): project = instance.project if hasattr(project, 'io_storages_localfilesexportstorages'): for storage in project.io_storages_localfilesexportstorages.all(): logger.debug(f'Export {instance} to Local Storage {storage}') storage.save_annotation(instance) @receiver(pre_delete, sender=Annotation) def delete_annotation_from_local_files(sender, instance, **kwargs): links = LocalFilesExportStorageLink.objects.filter(annotation=instance) for link in links: storage = link.storage if storage.can_delete_objects: logger.debug(f'Delete {instance} from Local Storage {storage}') # nosec storage.delete_annotation(instance)