# Copyright (C) 2021-2022 Intel Corporation # Copyright (C) CVAT.ai Corporation # # SPDX-License-Identifier: MIT import io import mimetypes import os import re import shutil import tempfile from abc import ABCMeta, abstractmethod from collections import defaultdict, deque from collections.abc import Collection, Iterable from contextlib import closing from copy import deepcopy from datetime import timedelta from enum import Enum from logging import Logger from pathlib import Path, PurePath from typing import Any, ClassVar from zipfile import ZipFile, ZipInfo import rapidjson from django.conf import settings from django.core.exceptions import ObjectDoesNotExist from django.core.files.uploadedfile import SimpleUploadedFile from django.db import transaction from django.db.models import Min, Prefetch from django.utils import timezone from rest_framework.exceptions import ValidationError from rest_framework.parsers import JSONParser from rest_framework.renderers import JSONRenderer import cvat.apps.dataset_manager as dm from cvat.apps.dataset_manager.util import ( ExportCacheManager, TmpDirManager, extend_export_file_lifetime, get_export_cache_lock, ) from cvat.apps.dataset_manager.views import ( EXPORT_CACHE_LOCK_ACQUISITION_TIMEOUT, EXPORT_CACHE_LOCK_TTL, EXPORT_LOCKED_RETRY_INTERVAL, LockNotAvailableError, retry_current_rq_job, ) from cvat.apps.engine import models from cvat.apps.engine.cache import MediaCache from cvat.apps.engine.log import ServerLogManager from cvat.apps.engine.models import DataChoice, StorageChoice, TaskMode from cvat.apps.engine.serializers import ( AnnotationGuideWriteSerializer, AssetWriteSerializer, AttributeSerializer, DataSerializer, JobWriteSerializer, LabeledDataSerializer, LabeledShapeSerializer, LabelSerializer, ProjectReadSerializer, SegmentSerializer, SimpleJobSerializer, TaskReadSerializer, ValidationParamsSerializer, ) from cvat.apps.engine.task import JobFileMapping, initialize_task from cvat.apps.engine.utils import av_scan_paths from cvat.utils import django_database as db_utils from cvat.utils.paths import join_untrusted_path, problem_with_untrusted_path from utils.dataset_manifest import ImageManifestManager slogger = ServerLogManager(__name__) class Version(Enum): V1 = "1.0" def _get_label_mapping(db_labels): label_mapping = {db_label.id: db_label.name for db_label in db_labels} for db_label in db_labels: label_mapping[db_label.id] = { "value": db_label.name, "attributes": {}, } for db_attribute in db_label.attributespec_set.all(): label_mapping[db_label.id]["attributes"][db_attribute.id] = db_attribute.name return label_mapping def _write_annotation_guide( zip_object: ZipFile, annotation_guide: models.AnnotationGuide | None, guide_filename: str, assets_dirname: str, target_dir: str, ) -> None: if annotation_guide is not None: md = annotation_guide.markdown assets = annotation_guide.assets.all() assets_dirname = os.path.join(target_dir, assets_dirname) guide_filename = os.path.join(target_dir, guide_filename) for db_asset in assets: md = md.replace( f"/api/assets/{str(db_asset.pk)}", os.path.join(assets_dirname, db_asset.filename) ) file = os.path.join(settings.ASSETS_ROOT, str(db_asset.pk), db_asset.filename) with open(file, "rb") as asset_file: zip_object.writestr( os.path.join(assets_dirname, db_asset.filename), asset_file.read() ) zip_object.writestr(guide_filename, data=md) def _read_annotation_guide(zip_object, guide_filename, assets_dirname): files = zip_object.namelist() if guide_filename in files: annotation_guide = io.BytesIO(zip_object.read(guide_filename)) assets = filter(lambda x: x.startswith(f"{assets_dirname}/"), files) assets = [(x, zip_object.read(x)) for x in assets] if len(assets) > settings.ASSET_MAX_COUNT_PER_GUIDE: raise ValidationError(f"Maximum number of assets per guide reached") for asset in assets: if len(asset[1]) / (1024 * 1024) > settings.ASSET_MAX_SIZE_MB: raise ValidationError(f"Maximum size of asset is {settings.ASSET_MAX_SIZE_MB} MB") if mimetypes.guess_type(asset[0])[0] not in settings.ASSET_SUPPORTED_TYPES: raise ValidationError( f"File is not supported as an asset. Supported are {settings.ASSET_SUPPORTED_TYPES}" ) return annotation_guide.getvalue(), assets return None, [] def _import_annotation_guide(owner, guide_data, assets): guide_serializer = AnnotationGuideWriteSerializer(data=guide_data) markdown = guide_data["markdown"] if guide_serializer.is_valid(raise_exception=True): guide_serializer.save() for asset in assets: name, data = asset asset_serializer = AssetWriteSerializer( data={ "file": SimpleUploadedFile( os.path.basename(name), data, mimetypes.guess_type(name)[0] ), "guide_id": guide_serializer.instance.id, } ) asset_serializer.is_valid(raise_exception=True) asset_serializer.save(owner=owner) markdown = markdown.replace(f"{name}", f"/api/assets/{asset_serializer.instance.pk}") guide_serializer.instance.markdown = markdown guide_serializer.instance.save() class _BackupBase: ANNOTATION_GUIDE_FILENAME = "annotation_guide.md" ASSETS_DIRNAME = "assets" def __init__(self, *args, logger=None, **kwargs): super().__init__(*args, **kwargs) self._logger = logger def _prepare_meta(self, allowed_keys, meta): keys_to_drop = set(meta.keys()) - allowed_keys if keys_to_drop: if self._logger: self._logger.warning("the following keys are dropped {}".format(keys_to_drop)) for key in keys_to_drop: del meta[key] return meta def _prepare_label_meta(self, label): allowed_fields = { "name", "color", "attributes", "type", "svg", "sublabels", } self._prepare_meta(allowed_fields, label) for sublabel in label["sublabels"]: sublabel_id = sublabel["id"] sublabel_name = sublabel["name"] label["svg"] = label["svg"].replace( f'data-label-id="{sublabel_id}"', f'data-label-name="{sublabel_name}"' ) self._prepare_meta(allowed_fields, sublabel) sublabel["attributes"] = [ self._prepare_attribute_meta(a) for a in sublabel["attributes"] ] return label def _prepare_attribute_meta(self, attribute): allowed_fields = { "name", "mutable", "input_type", "default_value", "values", } return self._prepare_meta(allowed_fields, attribute) class _TaskBackupBase(_BackupBase): MANIFEST_FILENAME = "task.json" MEDIA_MANIFEST_FILENAME = "manifest.jsonl" MEDIA_MANIFEST_INDEX_FILENAME = "index.json" ANNOTATIONS_FILENAME = "annotations.json" DATA_DIRNAME = "data" def _prepare_task_meta(self, task): allowed_fields = { "name", "bug_tracker", "status", "subset", "labels", } return self._prepare_meta(allowed_fields, task) def _prepare_data_meta(self, data): allowed_fields = { "chunk_size", "image_quality", "start_frame", "stop_frame", "frame_filter", "chunk_type", "storage_method", "storage", "sorting_method", "deleted_frames", "custom_segments", "job_file_mapping", "validation_layout", } self._prepare_meta(allowed_fields, data) if "validation_layout" in data: self._prepare_meta( allowed_keys={"mode", "frames", "frames_per_job_count"}, meta=data["validation_layout"], ) if "frame_filter" in data and not data["frame_filter"]: data.pop("frame_filter") return data def _prepare_job_meta(self, job): allowed_fields = { "status", "type", } return self._prepare_meta(allowed_fields, job) def _prepare_annotations(self, annotations, label_mapping): allowed_fields = { "label", "label_id", "type", "occluded", "outside", "z_order", "points", "rotation", "frame", "group", "source", "attributes", "shapes", "elements", "score", "start", "stop", } def _update_attribute(attribute, label): if "name" in attribute: source, dest = attribute.pop("name"), "spec_id" else: source, dest = attribute.pop("spec_id"), "name" attribute[dest] = label_mapping[label]["attributes"][source] def _update_label(shape, parent_label=""): if "label_id" in shape: source = shape.pop("label_id") shape["label"] = label_mapping[source]["value"] elif "label" in shape: source = parent_label + shape.pop("label") shape["label_id"] = label_mapping[source]["value"] return source def _prepare_shapes(shapes, parent_label=""): for shape in shapes: label = _update_label(shape, parent_label) for attr in shape["attributes"]: _update_attribute(attr, label) deque(_prepare_shapes(shape.get("elements", []), label), maxlen=0) self._prepare_meta(allowed_fields, shape) yield shape def _prepare_tracks(tracks, parent_label=""): for track in tracks: label = _update_label(track, parent_label) for shape in track["shapes"]: for attr in shape["attributes"]: _update_attribute(attr, label) self._prepare_meta(allowed_fields, shape) deque(_prepare_tracks(track.get("elements", []), label), maxlen=0) for attr in track["attributes"]: _update_attribute(attr, label) self._prepare_meta(allowed_fields, track) yield track def _prepare_intervals(intervals, parent_label=""): for interval in intervals: label = _update_label(interval, parent_label) for attr in interval["attributes"]: _update_attribute(attr, label) self._prepare_meta(allowed_fields, interval) yield interval def _prepare_tags(tags): for tag in tags: label = _update_label(tag) for attr in tag["attributes"]: _update_attribute(attr, label) self._prepare_meta(allowed_fields, tag) yield tag annotations["tags"] = _prepare_tags(annotations["tags"]) annotations["shapes"] = _prepare_shapes(annotations["shapes"]) annotations["tracks"] = _prepare_tracks(annotations["tracks"]) annotations["intervals"] = _prepare_intervals(annotations["intervals"]) return annotations def _get_db_jobs(self): if not self._db_task: return db_segments = ( self._db_task.segment_set.annotate(min_job_id=Min("job__id")) .order_by("min_job_id") .prefetch_related(Prefetch("job_set", queryset=models.Job.objects.order_by("id"))) ) for db_segment in db_segments: yield from db_segment.job_set.all() class _ExporterBase(metaclass=ABCMeta): ModelClass: ClassVar[models.Project | models.Task] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @staticmethod def _write_files(source_dir, zip_object, files, target_dir): for filename in files: arcname = os.path.normpath( os.path.join( target_dir, os.path.relpath(filename, source_dir), ) ) zip_object.write(filename=filename, arcname=arcname) def _write_directory( self, source_dir, zip_object, target_dir, recursive=True, exclude_files=None ): for root, dirs, files in os.walk(source_dir, topdown=True): if not recursive: dirs.clear() if files: self._write_files( source_dir=source_dir, zip_object=zip_object, files=( os.path.join(root, f) for f in files if not exclude_files or f not in exclude_files ), target_dir=target_dir, ) @abstractmethod def export_to(self, file: str) -> None: ... @classmethod def get_object(cls, pk: int) -> models.Project | models.Task: # FUTURE-FIXME: need to check permissions one more time when background task is called try: return cls.ModelClass.objects.get(pk=pk) except ObjectDoesNotExist: raise ValidationError(f"Such a {cls.ModelClass.__name__.lower()} does not exist") class TaskExporter(_ExporterBase, _TaskBackupBase): ModelClass: ClassVar[models.Task] = models.Task def __init__(self, pk, version=Version.V1, *, lightweight: bool): super().__init__(logger=slogger.task[pk]) self._db_task: models.Task = ( models.Task.objects.prefetch_related("data__images", "annotation_guide__assets") .select_related( "data__video", "data__audio", "data__validation_layout", "annotation_guide" ) .get(pk=pk) ) self._db_data: models.Data = self._db_task.data self._version = version db_labels = ( (self._db_task.project if self._db_task.project_id else self._db_task) .label_set.all() .prefetch_related("attributespec_set") ) self._label_mapping = _get_label_mapping(db_labels) self._lightweight = lightweight self._manifest_was_filtered = False def _write_annotation_guide(self, zip_object: ZipFile, target_dir: str) -> None: annotation_guide = ( self._db_task.annotation_guide if hasattr(self._db_task, "annotation_guide") else None ) _write_annotation_guide( zip_object, annotation_guide, self.ANNOTATION_GUIDE_FILENAME, self.ASSETS_DIRNAME, target_dir=target_dir, ) def _write_filtered_media_manifest(self, zip_object: ZipFile, target_dir: str) -> None: # When making a heavyweight backup of a task with images, we only include those frames # that match the task's frame range. This function filters the manifest so that it also # includes only those frames. That way, we don't have a manifest referencing nonexistent # images in the backup. target_data_dir = os.path.join(target_dir, self.DATA_DIRNAME) match (self._db_task.media_type, self._db_task.mode): case (models.MediaType.AUDIO, models.TaskMode.INTERPOLATION): return # there are no audio manifests case (models.MediaType.IMAGE, models.TaskMode.INTERPOLATION): # No filtering necessary; just use the original manifest. self._write_files( source_dir=self._db_data.get_upload_dirname(), zip_object=zip_object, files=[self._db_data.get_manifest_path()], target_dir=target_data_dir, ) return case ( models.MediaType.IMAGE | models.MediaType.POINT_CLOUD, models.TaskMode.ANNOTATION, ): imm_original = ImageManifestManager( self._db_data.get_manifest_path(), create_index=False ) # The task may have been created before we started generating manifests in every task. # If there is no manifest, don't add one to the backup but still set the filtered flag, # so that start_frame and other fields are adjusted properly later. if imm_original.exists: present_frame_nums = {im.frame for im in self._db_data.images.all()} with tempfile.TemporaryDirectory() as tmp_dir: filtered_manifest_path = Path(tmp_dir, self.MEDIA_MANIFEST_FILENAME) imm_filtered = ImageManifestManager( filtered_manifest_path, create_index=False ) imm_filtered.create( entry for frame_num, entry in imm_original if frame_num in present_frame_nums ) self._write_files( source_dir=tmp_dir, zip_object=zip_object, files=[filtered_manifest_path], target_dir=target_data_dir, ) self._manifest_was_filtered = True case (media_type, mode): assert False, f"Unknown media type '{media_type}' with mode '{mode}'" def _write_data_from_cloud_storage(self, zip_object: ZipFile, target_dir: str) -> None: assert not hasattr(self._db_data, "video"), "Only images can be stored in cloud storage" target_data_dir = os.path.join(target_dir, self.DATA_DIRNAME) data_dir = self._db_data.get_upload_dirname() self._write_filtered_media_manifest(zip_object=zip_object, target_dir=target_dir) files_for_local_copy = [] media_files_to_download: list[PurePath] = [] for media_file in self._db_data.related_files.all(): media_path = PurePath(media_file.path) local_path = os.path.join(data_dir, media_path) if os.path.exists(local_path): files_for_local_copy.append(local_path) else: media_files_to_download.append(media_path) frame_ids_to_download = [] frame_names_to_download = [] for media_file in self._db_data.images.order_by("frame").all(): media_path = media_file.path local_path = os.path.join(data_dir, media_path) if os.path.exists(local_path): files_for_local_copy.append(local_path) else: frame_ids_to_download.append(media_file.frame) frame_names_to_download.append(media_file.path) if media_files_to_download: storage_client = self._db_data.get_cloud_storage_instance() with tempfile.TemporaryDirectory() as tmp_dir: storage_client.bulk_download_to_dir( files=media_files_to_download, upload_dir=Path(tmp_dir) ) self._write_files( source_dir=tmp_dir, zip_object=zip_object, files=[os.path.join(tmp_dir, file) for file in media_files_to_download], target_dir=target_data_dir, ) if frame_ids_to_download: media_cache = MediaCache() with closing( media_cache.read_raw_images( self._db_task, frame_ids=frame_ids_to_download, decode=False ) ) as frame_iter: # Avoid closing the frame iter before the files are copied downloaded_paths = [] for _ in frame_ids_to_download: downloaded_paths.append(next(frame_iter)[1]) tmp_dir = downloaded_paths[0].removesuffix(frame_names_to_download[0]) self._write_files( source_dir=tmp_dir, zip_object=zip_object, files=downloaded_paths, target_dir=target_data_dir, ) self._write_files( source_dir=data_dir, zip_object=zip_object, files=files_for_local_copy, target_dir=target_data_dir, ) def _write_data(self, zip_object: ZipFile, target_dir: str) -> None: target_data_dir = os.path.join(target_dir, self.DATA_DIRNAME) if self._db_data.storage == StorageChoice.LOCAL: data_dir = self._db_data.get_upload_dirname() if self._db_data.local_storage_backing_cs: self._write_data_from_cloud_storage(zip_object, target_dir) else: self._write_directory( source_dir=data_dir, zip_object=zip_object, target_dir=target_data_dir, exclude_files=[self.MEDIA_MANIFEST_INDEX_FILENAME], ) elif self._db_data.storage == StorageChoice.SHARE: data_dir = settings.SHARE_ROOT match (self._db_task.media_type, self._db_task.mode): case (models.MediaType.IMAGE, models.TaskMode.INTERPOLATION): media_files = (os.path.join(data_dir, self._db_data.video.path),) case (models.MediaType.AUDIO, models.TaskMode.INTERPOLATION): media_files = (os.path.join(data_dir, self._db_data.audio.path),) case ( models.MediaType.IMAGE | models.MediaType.POINT_CLOUD, models.TaskMode.ANNOTATION, ): media_files = ( os.path.join(data_dir, im.path) for im in self._db_data.images.all() ) case (media_type, mode): assert False, f"Unknown media type '{media_type}' with '{mode}' mode" self._write_files( source_dir=data_dir, zip_object=zip_object, files=media_files, target_dir=target_data_dir, ) self._write_filtered_media_manifest(zip_object=zip_object, target_dir=target_dir) elif self._db_data.storage == StorageChoice.CLOUD_STORAGE: if (self._db_task.media_type, self._db_task.mode) not in ( (models.MediaType.IMAGE, models.TaskMode.ANNOTATION), (models.MediaType.POINT_CLOUD, models.TaskMode.ANNOTATION), ): raise AssertionError("Only images can be stored in cloud storage") data_dir = self._db_data.get_upload_dirname() if self._lightweight: self._write_files( source_dir=data_dir, zip_object=zip_object, files=[self._db_data.get_manifest_path()], target_dir=target_data_dir, ) else: self._write_data_from_cloud_storage(zip_object, target_dir) else: raise NotImplementedError def _write_manifest(self, zip_object: ZipFile, target_dir: str) -> None: def serialize_task(): task_serializer = TaskReadSerializer(self._db_task) for field in ( "url", "owner", "assignee", "jobs", "labels", "source_storage", "target_storage", ): task_serializer.fields.pop(field) task_labels = LabelSerializer(self._db_task.get_labels(prefetch=True), many=True) serialized_task = task_serializer.data task = self._prepare_task_meta(serialized_task) task["labels"] = [ self._prepare_label_meta(l) for l in task_labels.data if not l["has_parent"] ] for label in task["labels"]: label["attributes"] = [self._prepare_attribute_meta(a) for a in label["attributes"]] return task def serialize_segment(db_segment): segment_serializer = SegmentSerializer(db_segment) segment_serializer.fields.pop("jobs") serialized_segment = segment_serializer.data segment_type = serialized_segment.pop("type") if ( self._db_task.segment_size == 0 and segment_type == models.SegmentType.RANGE or self._db_data.validation_mode == models.ValidationMode.GT_POOL ): assert self._db_task.media_type != models.MediaType.AUDIO serialized_segment.update(serialize_segment_file_names(db_segment)) return serialized_segment def serialize_jobs(): db_segments = list(self._db_task.segment_set.all()) db_segments.sort(key=lambda i: i.job_set.first().id) serialized_jobs = [] for db_segment in db_segments: serialized_segment = serialize_segment(db_segment) db_jobs = list(db_segment.job_set.all()) db_jobs.sort(key=lambda v: v.id) for db_job in db_jobs: job_serializer = SimpleJobSerializer(db_job) for field in ("url", "assignee"): job_serializer.fields.pop(field) serialized_job = self._prepare_job_meta(job_serializer.data) serialized_job.update(deepcopy(serialized_segment)) serialized_jobs.append(serialized_job) return serialized_jobs def serialize_segment_file_names(db_segment: models.Segment): if self._db_task.mode == TaskMode.ANNOTATION: files: Iterable[models.Image] = self._db_data.images.order_by("frame").all() return {"files": [files[f].path for f in sorted(db_segment.frame_set)]} else: assert False, ( "Backups with custom file mapping are not supported" " in the 'interpolation' task mode" ) def serialize_data(): data_serializer = DataSerializer(self._db_data) data = data_serializer.data data["chunk_type"] = self._db_data.compressed_chunk_type # There are no deleted frames in DataSerializer so we need to pick it data["deleted_frames"] = self._db_data.deleted_frames if self._db_task.segment_size == 0: data["custom_segments"] = True if ( validation_layout := getattr(self._db_data, "validation_layout", None) ) and validation_layout.mode == models.ValidationMode.GT_POOL: validation_params_serializer = ValidationParamsSerializer( { "mode": validation_layout.mode, "frame_selection_method": models.JobFrameSelectionMethod.MANUAL, "frames_per_job_count": validation_layout.frames_per_job_count, } ) validation_params = validation_params_serializer.data media_filenames = dict( self._db_data.images.order_by("frame") .filter( frame__gte=min(validation_layout.frames), frame__lte=max(validation_layout.frames), ) .values_list("frame", "path") .all() ) validation_params["frames"] = [ media_filenames[frame] for frame in validation_layout.frames ] data["validation_layout"] = validation_params if ( self._db_data.storage == StorageChoice.SHARE or self._db_data.storage == StorageChoice.CLOUD_STORAGE and not self._lightweight ): data["storage"] = StorageChoice.LOCAL else: data["storage"] = self._db_data.storage if self._manifest_was_filtered: del data["start_frame"] del data["stop_frame"] del data["frame_filter"] return self._prepare_data_meta(data) task = serialize_task() task["version"] = self._version.value task["data"] = serialize_data() task["jobs"] = serialize_jobs() target_manifest_file = os.path.join(target_dir, self.MANIFEST_FILENAME) zip_object.writestr(target_manifest_file, data=JSONRenderer().render(task)) def _write_annotations(self, zip_object: ZipFile, target_dir: str) -> None: def serialize_annotations(): db_jobs = self._get_db_jobs() db_job_ids = (j.id for j in db_jobs) for db_job_id in db_job_ids: with db_utils.transaction_with_repeatable_read(): annotations = dm.task.get_job_data(db_job_id, streaming=True) assert not isinstance(annotations["shapes"], list) # Django many=True fields can only handle the list type # we're using a generator here, so it's processed separately annotations_serializer = LabeledDataSerializer( data=dict(annotations, shapes=[]) ) annotations_serializer.is_valid(raise_exception=True) annotation_data = annotations_serializer.data def serialize_shapes(): for shape in annotations["shapes"]: shape_serializer = LabeledShapeSerializer(data=shape) shape_serializer.is_valid(raise_exception=True) yield shape_serializer.data annotation_data["shapes"] = serialize_shapes() yield self._prepare_annotations(annotation_data, self._label_mapping) annotations = serialize_annotations() target_annotations_file = os.path.join(target_dir, self.ANNOTATIONS_FILENAME) with zip_object.open( # without this, the file will have the default timestamp (1980-01-01) ZipInfo(target_annotations_file, date_time=timezone.now().timetuple()), "w", ) as f: rapidjson.dump(annotations, f) def _export_task(self, zip_obj: ZipFile, target_dir: str) -> None: self._write_data(zip_obj, target_dir) self._write_manifest(zip_obj, target_dir) self._write_annotations(zip_obj, target_dir) self._write_annotation_guide(zip_obj, target_dir) def export_to(self, file: str | ZipFile, target_dir: str = "") -> None: if isinstance(file, str): with ZipFile(file, "w") as zf: self._export_task(zip_obj=zf, target_dir=target_dir) elif isinstance(file, ZipFile): self._export_task(zip_obj=file, target_dir=target_dir) else: raise ValueError("Unsupported type of file argument") class _ImporterBase: @staticmethod def _read_version(manifest): version = manifest.pop("version") try: return Version(version) except ValueError: raise ValueError("{} version is not supported".format(version)) @staticmethod def _prepare_dirs(filepath): target_dir = os.path.dirname(filepath) if not os.path.exists(target_dir): os.makedirs(target_dir) def _create_labels(self, labels, db_task=None, db_project=None, parent_label=None): label_mapping = {} if db_task: label_relation = {"task": db_task} else: label_relation = {"project": db_project} for label in labels: label_name = label["name"] attributes = label.pop("attributes", []) svg = label.pop("svg", "") sublabels = label.pop("sublabels", []) db_label = models.Label.objects.create(**label_relation, parent=parent_label, **label) label_mapping[(parent_label.name if parent_label else "") + label_name] = { "value": db_label.id, "attributes": {}, } label_mapping.update(self._create_labels(sublabels, db_task, db_project, db_label)) if db_label.type == str(models.LabelType.SKELETON): for db_sublabel in list(db_label.sublabels.all()): svg = svg.replace( f'data-label-name="{db_sublabel.name}"', f'data-label-id="{db_sublabel.id}"' ) models.Skeleton.objects.create(root=db_label, svg=svg) for attribute in attributes: attribute_name = attribute["name"] attribute_serializer = AttributeSerializer(data=attribute) attribute_serializer.is_valid(raise_exception=True) db_attribute = attribute_serializer.save(label=db_label) label_mapping[(parent_label.name if parent_label else "") + label_name][ "attributes" ][attribute_name] = db_attribute.id return label_mapping class TaskImporter(_ImporterBase, _TaskBackupBase): def __init__( self, file, user_id, org_id=None, project_id=None, subdir=None, label_mapping=None ): super().__init__(logger=slogger.glob) self._file = file self._subdir = subdir "Task subdirectory with the separator included, e.g. task_0/" self._user_id = user_id self._org_id = org_id self._manifest, self._annotations, self._annotation_guide, self._assets = self._read_meta() self._version = self._read_version(self._manifest) self._labels_mapping = label_mapping self._db_task = None self._project_id = project_id def _read_annotation_guide(self, zip_object): annotation_guide_filename = os.path.join(self._subdir or "", self.ANNOTATION_GUIDE_FILENAME) assets_dirname = os.path.join(self._subdir or "", self.ASSETS_DIRNAME) return _read_annotation_guide(zip_object, annotation_guide_filename, assets_dirname) def _read_meta(self): def read(zip_object): manifest_filename = os.path.join(self._subdir or "", self.MANIFEST_FILENAME) annotations_filename = os.path.join(self._subdir or "", self.ANNOTATIONS_FILENAME) manifest = JSONParser().parse(io.BytesIO(zip_object.read(manifest_filename))) annotations = JSONParser().parse(io.BytesIO(zip_object.read(annotations_filename))) annotation_guide, assets = self._read_annotation_guide(zip_object) return manifest, annotations, annotation_guide, assets if isinstance(self._file, str): with ZipFile(self._file, "r") as input_file: return read(input_file) elif isinstance(self._file, ZipFile): return read(self._file) raise ValueError("Unsupported type of file argument") @staticmethod def _fix_annotation_source(annotation: dict[str, Any]) -> None: # Workaround for the DB records that could have been introduced by the UI before # https://github.com/cvat-ai/cvat/issues/8874 was fixed. Backups can contain # invalid "source" field values. This fix only covers the known "Ground truth" value # errors that we know about, so the value validation keeps working for invalid inputs. # We silently replace them with the default value here, as the id-based workaround # in the serializer will miss the 'id' field in annotations from backups. if annotation.get("source") == "Ground truth": annotation["source"] = str(models.SourceType.MANUAL) for shape in annotation.get("shapes", []): TaskImporter._fix_annotation_source(shape) for element in annotation.get("elements", []): TaskImporter._fix_annotation_source(element) def _create_annotations(self, db_job, annotations): for annotation_type in ("tags", "shapes", "tracks", "intervals"): annotations.setdefault(annotation_type, []) self._prepare_annotations(annotations, self._labels_mapping) for annotation_type in ("tags", "shapes", "tracks", "intervals"): assert not isinstance(annotations[annotation_type], list) annotations[annotation_type] = list(annotations[annotation_type]) # backward compatibility for annotation in annotations[annotation_type]: self._fix_annotation_source(annotation) serializer = LabeledDataSerializer(data=annotations) serializer.is_valid(raise_exception=True) dm.task.put_job_data(db_job.id, serializer.data) @staticmethod def _calculate_segment_size(jobs: list[dict[str, Any]]) -> tuple[int, int]: # The type field will be missing in backups create before the GT jobs were introduced jobs = [ j for j in jobs if j.get("type", models.JobType.ANNOTATION) == models.JobType.ANNOTATION ] segment_size = jobs[0]["stop_frame"] - jobs[0]["start_frame"] + 1 overlap = 0 if len(jobs) == 1 else jobs[0]["stop_frame"] - jobs[1]["start_frame"] + 1 return segment_size, overlap @staticmethod def _parse_segment_frames(*, jobs: list[dict[str, Any]]) -> JobFileMapping: segments = [] for i, segment in enumerate(jobs): segment_size = segment["stop_frame"] - segment["start_frame"] + 1 if segment_frames := segment.get("frames"): segment_frames = set(segment_frames) segment_range = range(segment["start_frame"], segment["stop_frame"] + 1) if not segment_frames.issubset(segment_range): raise ValidationError( "Segment frames must be inside the range [start_frame; stop_frame]" ) segment_size = len(segment_frames) segment_files = segment["files"] if len(segment_files) != segment_size: raise ValidationError(f"segment {i}: segment files do not match segment size") segments.append(segment_files) return segments def _copy_input_files( self, input_archive: ZipFile | str, *, excluded_filenames: Collection[str] | None = None, ) -> list[str]: if isinstance(input_archive, str): with ZipFile(input_archive, "r") as zf: return self._copy_input_files( input_archive=zf, excluded_filenames=excluded_filenames, ) input_data_dirname = self.DATA_DIRNAME output_data_path = self._db_task.data.get_upload_dirname() uploaded_files = [] for file_path in input_archive.namelist(): if file_path.endswith("/") or self._subdir and not file_path.startswith(self._subdir): continue file_name = os.path.relpath(file_path, self._subdir) if excluded_filenames and file_name in excluded_filenames: continue if file_name.startswith(input_data_dirname + "/"): # It should be impossible for file_name to enable a path traversal attack # because it's the result of relpath(), which puts any ".." components in the front, # and the if condition will be false for any path that starts with "..". # But in case the surrounding logic changes, let's treat it as untrusted anyway. target_file = join_untrusted_path( output_data_path, os.path.relpath(file_name, input_data_dirname) ) self._prepare_dirs(target_file) with open(target_file, "wb") as out, input_archive.open(file_path) as source: shutil.copyfileobj(source, out) uploaded_files.append(os.path.relpath(file_name, input_data_dirname)) return uploaded_files def _import_task(self): data = self._manifest.pop("data") labels = self._manifest.pop("labels") jobs = self._manifest.pop("jobs") self._prepare_task_meta(self._manifest) self._manifest["owner_id"] = self._user_id self._manifest["project_id"] = self._project_id self._prepare_data_meta(data) excluded_input_files = [os.path.join(self.DATA_DIRNAME, self.MEDIA_MANIFEST_INDEX_FILENAME)] job_file_mapping = None if data.pop("custom_segments", False): job_file_mapping = self._parse_segment_frames( jobs=[v for v in jobs if v.get("type") == models.JobType.ANNOTATION] ) for d in [self._manifest, data]: for k in [ "segment_size", "overlap", "start_frame", "stop_frame", "sorting_method", "frame_filter", "filename_pattern", ]: d.pop(k, None) elif len(jobs) > 1: self._manifest["segment_size"], self._manifest["overlap"] = ( self._calculate_segment_size(jobs) ) validation_params = data.pop("validation_layout", None) if validation_params: validation_params["frame_selection_method"] = models.JobFrameSelectionMethod.MANUAL validation_params_serializer = ValidationParamsSerializer(data=validation_params) validation_params_serializer.is_valid(raise_exception=True) validation_params = validation_params_serializer.data gt_jobs = [v for v in jobs if v.get("type") == models.JobType.GROUND_TRUTH] if not gt_jobs: raise ValidationError("Can't find any GT jobs info in the backup files") elif len(gt_jobs) != 1: raise ValidationError("A task can have only one GT job info in the backup files") validation_params["frames"] = validation_params_serializer.initial_data["frames"] if validation_params["mode"] == models.ValidationMode.GT_POOL: gt_job_frames = self._parse_segment_frames(jobs=gt_jobs)[0] if set(gt_job_frames) != set(validation_params_serializer.initial_data["frames"]): raise ValidationError("GT job frames do not match validation frames") # Validation frames can have a different order, we must use the GT job order if not job_file_mapping: raise ValidationError("Expected segment info in the backup files") job_file_mapping.append(gt_job_frames) data["validation_params"] = validation_params if job_file_mapping and ( not validation_params or validation_params["mode"] != models.ValidationMode.GT_POOL ): # It's currently prohibited to have repeated file names in jobs. # DataSerializer checks for this, but we don't need it for tasks with a GT pool data["job_file_mapping"] = job_file_mapping self._manifest["consensus_replicas"] = self._determine_replica_counts(jobs) self._db_task = models.Task.objects.create(**self._manifest, organization_id=self._org_id) task_data_path = self._db_task.get_dirname() if os.path.isdir(task_data_path): shutil.rmtree(task_data_path) os.makedirs(task_data_path) if not self._labels_mapping: self._labels_mapping = self._create_labels(db_task=self._db_task, labels=labels) data_serializer = DataSerializer(data=data) data_serializer.is_valid(raise_exception=True) db_data = data_serializer.save() self._db_task.data = db_data self._db_task.save() uploaded_files = self._copy_input_files(self._file, excluded_filenames=excluded_input_files) data["use_zip_chunks"] = data.pop("chunk_type") == DataChoice.IMAGESET data = data_serializer.data data["client_files"] = uploaded_files if job_file_mapping or ( validation_params and validation_params["mode"] == models.ValidationMode.GT_POOL ): data["job_file_mapping"] = job_file_mapping if validation_params: data["validation_params"] = validation_params if data_serializer.initial_data["storage"] == StorageChoice.CLOUD_STORAGE: db_data.storage = StorageChoice.CLOUD_STORAGE if data["client_files"] != [self.MEDIA_MANIFEST_FILENAME]: raise ValidationError(f"Expected {self.MEDIA_MANIFEST_FILENAME} in backup files") manifest = ImageManifestManager( self._db_task.data.get_upload_dirname() / self.MEDIA_MANIFEST_FILENAME ) data["server_files"] = [] for _, manifest_entry in manifest: data["server_files"].append(manifest_entry.full_name) data["server_files"].extend( manifest_entry.get("meta", {}).get("related_images", []) ) for server_file in data["server_files"]: if problem := problem_with_untrusted_path(server_file): raise ValidationError(f"Unsafe file path in manifest: {problem}") else: if data_serializer.initial_data["storage"] != StorageChoice.LOCAL: raise ValidationError(f"Unexpected storage type in the backup files") db_data.storage = StorageChoice.LOCAL db_data.save(update_fields=["storage"]) initialize_task(db_task=self._db_task.pk, data=data.copy(), is_backup_restore=True) self._db_task.refresh_from_db() db_data.refresh_from_db() db_data.deleted_frames = data_serializer.initial_data.get("deleted_frames", []) db_data.save(update_fields=["deleted_frames"]) if not validation_params: # In backups created before addition of GT pools there was no validation_layout field # Recreate Ground Truth jobs self._import_gt_jobs(jobs) for db_job, job in zip(self._get_db_jobs(), jobs): db_job.status = job["status"] db_job.save() @staticmethod def _get_job_type(job: dict) -> models.JobType: try: # The type field will be missing in backups created before the GT jobs were introduced raw_job_type = job.get("type", models.JobType.ANNOTATION.value) job_type = models.JobType(raw_job_type) except ValueError: raise ValidationError(f"Unexpected job type {raw_job_type}") return job_type def _collect_replica_counts(self, jobs: dict) -> dict[models.JobType, list[int]]: replica_counts = [] for job in jobs: job_type = self._get_job_type(job) if job_type != models.JobType.CONSENSUS_REPLICA: replica_counts.append([job_type, 0]) else: if not replica_counts: raise ValidationError( f"Invalid job order, jobs of type '{job_type}' " "must follow their parent jobs." ) replica_counts[-1][1] += 1 replica_counts_map = defaultdict(list) for primary_type, count in replica_counts: replica_counts_map[primary_type].append(count) return replica_counts_map def _determine_replica_counts(self, jobs: dict) -> int: replica_counts = self._collect_replica_counts(jobs) return min(replica_counts.get(models.JobType.ANNOTATION, [0])) def _import_gt_jobs(self, jobs): for job in jobs: job_type = self._get_job_type(job) if job_type == models.JobType.GROUND_TRUTH: job_serializer = JobWriteSerializer( data={ "task_id": self._db_task.id, "type": job_type.value, **( { "frame_selection_method": ( models.JobFrameSelectionMethod.MANUAL.value ), "frames": job["frames"], } if self._db_task.media_type != models.MediaType.AUDIO else {} ), } ) job_serializer.is_valid(raise_exception=True) job_serializer.save() elif job_type in [models.JobType.ANNOTATION, models.JobType.CONSENSUS_REPLICA]: continue else: assert False def _import_annotations(self): db_jobs = self._get_db_jobs() for db_job, annotations in zip(db_jobs, self._annotations): self._create_annotations(db_job, annotations) def _import_annotation_guide(self): if self._annotation_guide: markdown = self._annotation_guide.decode() _import_annotation_guide( self._db_task.owner, {"markdown": markdown, "task_id": self._db_task.id}, self._assets, ) def import_task(self): self._import_task() self._import_annotations() self._import_annotation_guide() return self._db_task @transaction.atomic def import_task(filename, user, org_id): av_scan_paths(filename) task_importer = TaskImporter(filename, user, org_id) db_task = task_importer.import_task() return db_task.id class _ProjectBackupBase(_BackupBase): MANIFEST_FILENAME = "project.json" TASKNAME_TEMPLATE = "task_{}" def _prepare_project_meta(self, project): allowed_fields = { "bug_tracker", "labels", "name", "status", } return self._prepare_meta(allowed_fields, project) class ProjectExporter(_ExporterBase, _ProjectBackupBase): ModelClass: ClassVar[models.Project] = models.Project def __init__(self, pk, *, lightweight: bool, version: Version = Version.V1): super().__init__(logger=slogger.project[pk]) self._db_project = ( self.ModelClass.objects.prefetch_related("tasks", "annotation_guide__assets") .select_related("annotation_guide") .get(pk=pk) ) self._version = version db_labels = self._db_project.label_set.all().prefetch_related("attributespec_set") self._label_mapping = _get_label_mapping(db_labels) self._lightweight = lightweight def _write_annotation_guide(self, zip_object: ZipFile) -> None: annotation_guide = ( self._db_project.annotation_guide if hasattr(self._db_project, "annotation_guide") else None ) _write_annotation_guide( zip_object, annotation_guide, self.ANNOTATION_GUIDE_FILENAME, self.ASSETS_DIRNAME, target_dir="", ) def _write_tasks(self, zip_object): for idx, db_task in enumerate(self._db_project.tasks.all().order_by("id")): if db_task.data is not None: TaskExporter( db_task.id, self._version, lightweight=self._lightweight, ).export_to(zip_object, self.TASKNAME_TEMPLATE.format(idx)) def _write_manifest(self, zip_object): def serialize_project(): project_serializer = ProjectReadSerializer(self._db_project) for field in ("assignee", "owner", "url"): project_serializer.fields.pop(field) project_labels = LabelSerializer( self._db_project.get_labels(prefetch=True), many=True ).data project = self._prepare_project_meta(project_serializer.data) project["labels"] = [ self._prepare_label_meta(l) for l in project_labels if not l["has_parent"] ] for label in project["labels"]: label["attributes"] = [self._prepare_attribute_meta(a) for a in label["attributes"]] return project project = serialize_project() project["version"] = self._version.value zip_object.writestr(self.MANIFEST_FILENAME, data=JSONRenderer().render(project)) def export_to(self, file: str) -> None: with ZipFile(file, "w") as output_file: self._write_annotation_guide(output_file) self._write_manifest(output_file) self._write_tasks(output_file) class ProjectImporter(_ImporterBase, _ProjectBackupBase): TASKNAME_RE = r"task_(\d+)/" def __init__(self, filename, user_id, org_id=None): super().__init__(logger=slogger.glob) self._filename = filename self._user_id = user_id self._org_id = org_id self._manifest, self._annotation_guide, self._assets = self._read_meta() self._version = self._read_version(self._manifest) self._db_project = None self._labels_mapping = {} def _read_annotation_guide(self, zip_object): return _read_annotation_guide( zip_object, self.ANNOTATION_GUIDE_FILENAME, self.ASSETS_DIRNAME ) def _read_meta(self): with ZipFile(self._filename, "r") as input_file: manifest = JSONParser().parse(io.BytesIO(input_file.read(self.MANIFEST_FILENAME))) annotation_guide, assets = self._read_annotation_guide(input_file) return manifest, annotation_guide, assets def _import_project(self): labels = self._manifest.pop("labels") self._prepare_project_meta(self._manifest) self._manifest["owner_id"] = self._user_id self._db_project = models.Project.objects.create( **self._manifest, organization_id=self._org_id ) project_path = self._db_project.get_dirname() if os.path.isdir(project_path): shutil.rmtree(project_path) os.makedirs(project_path) self._labels_mapping = self._create_labels(db_project=self._db_project, labels=labels) def _import_tasks(self): def get_tasks(zip_object): tasks = {} for fname in zip_object.namelist(): m = re.match(self.TASKNAME_RE, fname) if m: tasks[int(m.group(1))] = m.group(0) return [v for _, v in sorted(tasks.items())] with ZipFile(self._filename, "r") as zf: task_dirs = get_tasks(zf) for task_dir in task_dirs: TaskImporter( file=zf, user_id=self._user_id, org_id=self._org_id, project_id=self._db_project.id, subdir=task_dir, label_mapping=self._labels_mapping, ).import_task() def _import_annotation_guide(self): if self._annotation_guide: markdown = self._annotation_guide.decode() _import_annotation_guide( self._db_project.owner, {"markdown": markdown, "project_id": self._db_project.id}, self._assets, ) def import_project(self): self._import_project() self._import_annotation_guide() self._import_tasks() return self._db_project @transaction.atomic def import_project(filename, user, org_id): av_scan_paths(filename) project_importer = ProjectImporter(filename, user, org_id) db_project = project_importer.import_project() return db_project.id def create_backup( instance_id: int, Exporter: type[ProjectExporter | TaskExporter], logger: Logger, cache_ttl: timedelta, *, lightweight: bool = None, ): db_instance = Exporter.get_object(instance_id) try: instance_type = db_instance.__class__.__name__ instance_timestamp = timezone.localtime(db_instance.updated_date).timestamp() output_path = ExportCacheManager.make_backup_file_path( instance_id=db_instance.id, instance_type=instance_type, instance_timestamp=instance_timestamp, lightweight=lightweight, ) with get_export_cache_lock( output_path, block=True, acquire_timeout=EXPORT_CACHE_LOCK_ACQUISITION_TIMEOUT, ttl=EXPORT_CACHE_LOCK_TTL, ): # output_path includes timestamp of the last update if os.path.exists(output_path): extend_export_file_lifetime(output_path) return output_path with TmpDirManager.get_tmp_directory_for_export(instance_type=instance_type) as tmp_dir: temp_file = os.path.join(tmp_dir, "dump") exporter = Exporter(db_instance.id, lightweight=lightweight) exporter.export_to(temp_file) with get_export_cache_lock( output_path, block=True, acquire_timeout=EXPORT_CACHE_LOCK_ACQUISITION_TIMEOUT, ttl=EXPORT_CACHE_LOCK_TTL, ): shutil.move(temp_file, output_path) logger.info( f"The {db_instance.__class__.__name__.lower()} '{db_instance.id}' is backed up at {output_path!r} " f"and available for downloading for the next {cache_ttl}." ) except LockNotAvailableError: # Need to retry later if the lock was not available retry_current_rq_job(EXPORT_LOCKED_RETRY_INTERVAL) logger.info( "Failed to acquire export cache lock. Retrying in {}".format( EXPORT_LOCKED_RETRY_INTERVAL ) ) raise return output_path def get_backup_dirname(): return TmpDirManager.TMP_ROOT