from logging import getLogger
|
from typing import TYPE_CHECKING
|
|
from django.db.models import QuerySet
|
from tasks.models import AnnotationDraft, Task
|
|
logger = getLogger(__name__)
|
|
|
if TYPE_CHECKING:
|
from projects.models import Project, ProjectSummary
|
|
|
def get_unique_ids_list(tasks_queryset):
|
"""
|
Convert various input types to a list of unique IDs.
|
|
:param tasks_queryset: Can be:
|
- list of IDs (integers)
|
- list of objects with 'id' attribute
|
- Django QuerySet
|
- set of IDs or objects
|
:return: list of unique IDs
|
"""
|
if isinstance(tasks_queryset, (list, tuple)):
|
if not tasks_queryset:
|
return []
|
|
# Check if it's a list of IDs (integers)
|
if isinstance(tasks_queryset[0], int):
|
return list(set(tasks_queryset)) # Remove duplicates
|
|
# It's a list of objects with 'id' attribute
|
return list(set(obj.id for obj in tasks_queryset))
|
|
elif isinstance(tasks_queryset, set):
|
if not tasks_queryset:
|
return []
|
|
# Check if it's a set of IDs (integers)
|
first_item = next(iter(tasks_queryset))
|
if isinstance(first_item, int):
|
return list(tasks_queryset)
|
|
# It's a set of objects with 'id' attribute
|
return list(obj.id for obj in tasks_queryset)
|
|
elif isinstance(tasks_queryset, QuerySet):
|
# It's a Django QuerySet
|
return list(tasks_queryset.values_list('id', flat=True))
|
|
else:
|
raise ValueError(f'Unsupported type for tasks_queryset: {type(tasks_queryset)}')
|
|
|
def make_queryset_from_iterable(tasks_list):
|
"""
|
Make queryset from list/set of int/Tasks
|
:param tasks_list: Iterable of Tasks or IDs
|
:return: Tasks queryset
|
"""
|
if isinstance(tasks_list, set):
|
tasks_list = list(tasks_list)
|
# Make query set from list of IDs
|
if isinstance(tasks_list, list) and len(tasks_list) > 0:
|
# Extract task IDs from Tasks list
|
if isinstance(tasks_list[0], Task):
|
tasks_list = [task.id for task in tasks_list]
|
queryset = Task.objects.filter(id__in=tasks_list)
|
else:
|
ids = []
|
for task in tasks_list:
|
if isinstance(task, Task):
|
ids.append(task.id)
|
elif isinstance(task, int):
|
ids.append(task)
|
else:
|
raise ValueError(f'Unknown object type: {str(task)}')
|
queryset = Task.objects.filter(id__in=ids)
|
return queryset
|
|
|
def recalculate_created_annotations_and_labels_from_scratch(
|
project: 'Project', summary: 'ProjectSummary', organization_id: int
|
) -> None:
|
"""Recalculate from scratch:
|
task columns
|
created_labels
|
created_annotations
|
created_labels_drafts
|
|
:param project: Project
|
:param summary: ProjectSummary
|
:param organization_id: Organization.id, it is required for django-rq displaying on admin page
|
"""
|
logger.info(f'Reset cache started for project {project.id} and organization {organization_id}')
|
logger.info(f'recalculate_created_annotations_and_labels_from_scratch project_id={project.id}')
|
summary.all_data_columns = {}
|
summary.common_data_columns = []
|
summary.update_data_columns(project.tasks.only('data'))
|
|
summary.created_labels, summary.created_annotations = {}, {}
|
summary.update_created_annotations_and_labels(project.annotations.all())
|
|
summary.created_labels_drafts = {}
|
drafts = AnnotationDraft.objects.filter(task__project=project)
|
summary.update_created_labels_drafts(drafts)
|
|
logger.info(
|
f'Reset cache finished for project {project.id} and organization {organization_id}:\n'
|
f'created_annotations = {summary.created_annotations}\n'
|
f'created_labels = {summary.created_labels}\n'
|
f'created_labels_drafts = {summary.created_labels_drafts}'
|
)
|
