import json
import logging
import re
import uuid
import requests
from django.conf import settings
from django.contrib.auth.models import User
from django.contrib.contenttypes.fields import GenericRelation
from django.db import models, transaction
from django.utils.functional import cached_property
from django.utils.text import slugify
from django_comments_xtd.models import XtdComment
import iconolab.signals.handlers as iconolab_signals
logger = logging.getLogger(__name__)
# https://docs.djangoproject.com/fr/1.11/topics/db/sql/#executing-custom-sql-directly
def dictfetchall(cursor):
"Return all rows from a cursor as a dict"
columns = [col[0] for col in cursor.description]
return [
dict(zip(columns, row))
for row in cursor.fetchall()
]
class Collection(models.Model):
"""
Collection objects are the thematic item repositories in Iconolab
name: the name displayed in the url and also used to identify the collection
verbose_name: the name displayed in the text of the pages
description: the short description of the collection that will be
displayed by default in pages
complete_description: the complete description that will be shown
with a "view more" button/link
image/height/width: the collection image that will be shown in the
collection description
show_image_on_home: if True, the collection will appear by default
on the homepage as one of the bigger images
"""
name = models.SlugField(max_length=50, unique=True)
verbose_name = models.CharField(max_length=50, null=True, blank=True)
description = models.TextField(null=True, blank=True, default="")
complete_description = models.TextField(null=True, blank=True, default="")
image = models.ImageField(
upload_to='uploads/', height_field='height', width_field='width', null=True, blank=True)
height = models.IntegerField(null=True, blank=True)
width = models.IntegerField(null=True, blank=True)
show_image_on_home = models.BooleanField(default=False)
link_text = models.CharField(max_length=1024, default="", null=True, blank=True)
link_url = models.CharField(max_length=1024, default="", null=True, blank=True)
@cached_property
def items_count(self):
return self.items.count()
@cached_property
def completed_percent(self):
items_with_annotation = \
ImageStats.objects.filter(image__item__collection=self, annotations_count__gt=0)\
.values('image__item').distinct().count()
total_items = self.items_count
return int(round((items_with_annotation * 100) / total_items)) if total_items > 0 else 0
def __str__(self):
return self.name
class Folder(models.Model):
"""
Some items may belong to a "folder". This is actually a physical folder
"""
folder_guid = models.UUIDField(default=uuid.uuid4, editable=False)
collection = models.ForeignKey(Collection, related_name="folders", on_delete=models.PROTECT)
name = models.TextField(null=False, blank=False)
description = models.TextField(null=True, blank=True)
original_id = models.CharField(max_length=256, null=True, blank=True)
display_image = models.ImageField(blank=True, null=True, upload_to='uploads/')
@cached_property
def items(self):
return Item.objects.filter(folders=self)
@cached_property
def items_count(self):
return self.items.count()
@property
def image(self):
if self.display_image:
return self.display_image
first_image = Image.objects.filter(item__folders=self).order_by('item__id', '-name').first()
# first_item = self.items.first()
# if not first_item:
# return None
# images = Image.objects.filter(item=first_item)
# first_image = images.first()
return first_image.media if first_image else None
def __str__(self):
return 'Folder ' + self.name
class Item(models.Model):
"""
Item objects belong to a collection, are linked to a metadata item, and
to one or more images
"""
collection = models.ForeignKey(Collection, related_name="items", on_delete=models.PROTECT)
item_guid = models.UUIDField(default=uuid.uuid4, editable=False)
folders = models.ManyToManyField('Folder')
def __str__(self):
return str(self.item_guid) + ":from:" + self.collection.name
@cached_property
def images_sorted_by_name(self):
res = list(self.images.all())
res.sort(key=lambda img: img.name, reverse=True)
return res
@cached_property
def get_item_link_text(self):
link_text = self.collection.link_text
if not link_text:
return ''
else:
try:
return link_text.format(**self.metadatas.metadata_obj)
except (ValueError, SyntaxError, TypeError, NameError):
return''
logger.info('the text link constitution is not corresponding.')
@cached_property
def get_item_link_url(self):
link_url = self.collection.link_url
if not link_url:
return ''
else:
try:
return link_url.format(**self.metadatas.metadata_obj)
except (ValueError, SyntaxError, TypeError, NameError):
return ''
logger.info('the url link constitution is not corresponding.')
class ItemMetadata(models.Model):
"""
Metadata object for the item class.
"""
item = models.OneToOneField('Item', related_name='metadatas', on_delete=models.CASCADE)
natural_key = models.CharField(max_length=1024, default="", unique=True)
"""
JSON field integration
"""
metadata = models.TextField(default="", null=False)
@staticmethod
def get_natural_key(collection, raw_natural_key):
return '%s|%s' % (collection.name, raw_natural_key)
def __init__(self, *args, **kwargs):
self.__metadata_obj = None
self.__raw_natural_key = None
super().__init__(*args, **kwargs)
def __setattr__(self, name, value):
if name == 'metadata':
self.__metadata_obj = None
elif name == 'natural_key':
self.__raw_natural_key = None
return super().__setattr__(name, value)
@property
def metadata_obj(self):
if self.__metadata_obj is None:
self.__metadata_obj = json.loads(self.metadata)
return self.__metadata_obj
@property
def raw_natural_key(self):
if self.__raw_natural_key is None:
self.__raw_natural_key = (self.natural_key or "").split("|")[-1]
return self.__raw_natural_key
def __str__(self):
return "metadatas:for:" + str(self.item.item_guid)
class Image(models.Model):
"""
Each image object is linked to one item, users can create annotations on images
"""
image_guid = models.UUIDField(default=uuid.uuid4, editable=False)
name = models.CharField(max_length=200)
media = models.ImageField(upload_to='uploads/',
height_field='height', width_field='width')
item = models.ForeignKey(
'Item', related_name='images', null=True, blank=True, on_delete=models.CASCADE)
height = models.IntegerField(null=False, blank=False)
width = models.IntegerField(null=False, blank=False)
created = models.DateTimeField(auto_now_add=True, null=True)
def __str__(self):
return str(self.image_guid) + ":" + self.name
@property
def wh_ratio(self):
return self.width / self.height
@property
def latest_annotations(self):
return self.annotations.all().order_by('-created')
@property
def tag_labels(self):
tag_list = []
for annotation in self.annotations.all():
revision_tags = json.loads(
annotation.current_revision.get_tags_json())
tag_list += [tag_infos['tag_label']
for tag_infos in revision_tags if tag_infos.get('tag_label') is not None] # deal with
return tag_list
def is_bookmarked_by(self, user):
return Bookmark.objects.filter(image=self, category__user=user).count() > 0
class ImageStats(models.Model):
"""
Stats objects for a given image, keep count of several values to be
displayed in image and item pages
"""
image = models.OneToOneField(
'Image', related_name='stats', blank=False, null=False, on_delete=models.CASCADE)
views_count = models.IntegerField(blank=True, null=True, default=0)
annotations_count = models.IntegerField(blank=True, null=True, default=0)
submitted_revisions_count = models.IntegerField(
blank=True, null=True, default=0)
comments_count = models.IntegerField(blank=True, null=True, default=0)
folders_inclusion_count = models.IntegerField(
blank=True, null=True, default=0)
tag_count = models.IntegerField(blank=True, null=True, default=0)
def __str__(self):
return "stats:for:" + str(self.image.image_guid)
def set_tags_stats(self):
self.tag_count = Tag.objects.filter(
tagginginfo__revision__annotation__image=self.image).distinct().count()
@transaction.atomic
def update_stats(self):
self.annotations_count = 0
self.submitted_revisions_count = 0
self.comments_count = 0
image_annotations = Annotation.objects.filter(image=self.image)
# views_count - Can't do much about views count
# annotations_count
self.annotations_count = image_annotations.count()
# submitted_revisions_count & comment_count
for annotation in image_annotations.all():
annotation_revisions = annotation.revisions
self.submitted_revisions_count += annotation_revisions.count()
self.comments_count += XtdComment.objects.for_app_models("iconolab.annotation").filter(
object_pk=annotation.pk,
).count()
# tag_count
self.tag_count = Tag.objects.filter(
tagginginfo__revision__annotation__image=self.image).distinct().count()
self.save()
class AnnotationManager(models.Manager):
"""
Manager class for annotation, it handles annotation creation (with initial revision creation, and
has methods to get a list of annotation commented for a given user, and a list of annotations contributed for a
given user
"""
@transaction.atomic
def create_annotation(self, author, image, title='', description='', fragment='', tags_json='[]'):
"""
Creates a new Annotation with its associated AnnotationStats and initial AnnotationRevision
"""
# Create annotation object
new_annotation = Annotation(
image=image,
author=author
)
new_annotation.save()
# Create initial revision
initial_revision = AnnotationRevision(
annotation=new_annotation,
author=author,
title=title,
description=description,
fragment=fragment,
state=AnnotationRevision.ACCEPTED
)
initial_revision.save()
initial_revision.set_tags(tags_json)
# Create stats object
new_annotation_stats = AnnotationStats(annotation=new_annotation)
new_annotation_stats.set_tags_stats()
new_annotation_stats.save()
# Link everything to parent
new_annotation.current_revision = initial_revision
new_annotation.stats = new_annotation_stats
new_annotation.save()
iconolab_signals.revision_created.send(
sender=AnnotationRevision, instance=initial_revision)
return new_annotation
@transaction.atomic
def get_annotations_contributed_for_user(self, user):
"""
user is the user whom we want to get the contributed annotations
Returns the list of all the annotations on which the user submitted
a revision but did not create the annotation
List of dict in the format:
{
"annotation_obj": annotation object,
"revisions_count": revisions count for user
"awaiting_count": awaiting revisions for user on this annotation
"accepted_count": accepted revisions for user
"latest_submitted_revision": date of the latest submitted revision
from user on annotation
}
"""
latest_revision_on_annotations = []
user_contributed_annotations = Annotation.objects.filter(revisions__author=user).exclude(author=user).prefetch_related(
'current_revision',
'revisions',
'image',
'image__item',
'image__item__collection').distinct()
for annotation in user_contributed_annotations.all():
latest_revision_on_annotations.append(
annotation.revisions.filter(author=user).latest(field_name="created"))
contributed_list = []
if latest_revision_on_annotations:
latest_revision_on_annotations.sort(
key=lambda item: item.created, reverse=True)
contributed_list = [
{
"annotation_obj": revision.annotation,
"revisions_count": revision.annotation.revisions.filter(author=user).count(),
"awaiting_count": revision.annotation.revisions.filter(author=user, state=AnnotationRevision.AWAITING).count(),
"accepted_count": revision.annotation.revisions.filter(author=user, state=AnnotationRevision.ACCEPTED).count(),
"latest_submitted_revision": revision.created
}
for revision in latest_revision_on_annotations
]
return contributed_list
@transaction.atomic
def get_annotations_commented_for_user(self, user, ignore_revisions_comments=True):
"""
user is the user for which we want to get the commented annotations
ignore_revisions_comment allows to filter comments that are associated with a revision
Returns a list of all annotations on which a given user commented
with user-comments-related data
List of dict in the format:
{
"annotation_obj": annotation object,
"comment_count": comment count for user
"latest_comment_date": date of the latest comment from user on annotation
}
"""
user_comments = IconolabComment.objects.filter(
user=user, content_type__app_label='iconolab', content_type__model='annotation').order_by('-submit_date')
if ignore_revisions_comments:
user_comments = user_comments.filter(revision__isnull=True)
all_user_comments_data = [
(comment.object_pk, comment.submit_date) for comment in user_comments]
unique_ordered_comments_data = []
for (id, submit_date) in all_user_comments_data:
if id not in [item["annotation_id"] for item in unique_ordered_comments_data]:
unique_ordered_comments_data.append(
{"annotation_id": id, "latest_comment_date": submit_date})
commented_annotations = Annotation.objects.filter(id__in=[item["annotation_id"] for item in unique_ordered_comments_data]).prefetch_related(
'current_revision',
'revisions',
'image',
'image__item',
'image__item__collection'
).distinct()
sorted_annotations_list = []
for comment_data in unique_ordered_comments_data:
annotation_obj = commented_annotations.get(
id=comment_data["annotation_id"])
sorted_annotations_list.append(
{
"annotation_obj": annotation_obj,
"comment_count_for_user": user_comments.filter(object_pk=annotation_obj.id).count(),
"latest_comment_date": comment_data["latest_comment_date"]
}
)
return sorted_annotations_list
class Annotation(models.Model):
"""
Annotation objects are created on a given image, each annotation have a
list of revisions to keep track of its history, the latest revision is
the 'current revision' that will be displayed by default in most pages.
Annotation data (title, description, fragment) is thus stored in the revision.
Annotations can be considered validated or not depending on the metacategories
posted in their comments through the attribute validation_state. Their
validation state can also be overriden and in such case we can use
validation_state_overriden attribute to remember it in the model (so for
instance if an admin un-validates an annotation we could block it from
being validated again)
"""
UNVALIDATED = 0
VALIDATED = 1
VALIDATION_STATES = (
(UNVALIDATED, 'unvalidated'),
(VALIDATED, 'validated'),
)
annotation_guid = models.UUIDField(default=uuid.uuid4, editable=False)
image = models.ForeignKey(
'Image', related_name='annotations', on_delete=models.CASCADE)
source_revision = models.ForeignKey(
'AnnotationRevision', related_name='source_related_annotation', blank=True, null=True, on_delete=models.SET_NULL)
current_revision = models.OneToOneField(
'AnnotationRevision', related_name='current_for_annotation', blank=True, null=True, on_delete=models.SET_NULL)
author = models.ForeignKey(User, null=True, on_delete=models.PROTECT)
created = models.DateTimeField(auto_now_add=True, null=True)
comments = GenericRelation(
'IconolabComment', content_type_field='content_type_id', object_id_field='object_pk')
validation_state = models.IntegerField(
choices=VALIDATION_STATES, default=UNVALIDATED)
validation_state_overriden = models.BooleanField(default=False)
objects = AnnotationManager()
def __str__(self):
return str(self.annotation_guid) + ":" + self.current_revision.title
@property
def awaiting_revisions_count(self):
return self.revisions.filter(state=AnnotationRevision.AWAITING).distinct().count()
@property
def accepted_revisions_count(self):
return self.revisions.filter(state=AnnotationRevision.ACCEPTED).distinct().count()
@property
def rejected_revisions_count(self):
return self.revisions.filter(state=AnnotationRevision.REJECTED).distinct().count()
@property
def studied_revisions_count(self):
return self.revisions.filter(state=AnnotationRevision.STUDIED).distinct().count()
@property
def total_revisions_count(self):
return self.revisions.distinct().count()
@property
def collection(self):
return self.image.collection
@property
def tag_labels(self):
if self.current_revision is None:
return []
current_revision_tags = json.loads(
self.current_revision.get_tags_json())
return [tag_infos['tag_label'] for tag_infos in current_revision_tags if tag_infos.get('tag_label') is not None]
def latest_revision_for_user(self, user):
user_revisions = self.revisions.filter(creator=user)
if user_revisions.exists():
return user_revisions.order_by("-created").first()
return None
@transaction.atomic
def make_new_revision(self, author, title, description, fragment, tags_json):
"""
Called to create a new revision, potentially from a merge
"""
if author == self.author:
# We're creating an automatically accepted revision
new_revision_state = AnnotationRevision.ACCEPTED
else:
# Revision will require validation
new_revision_state = AnnotationRevision.AWAITING
new_revision = AnnotationRevision(
annotation=self,
parent_revision=self.current_revision,
title=title,
description=description,
author=author,
fragment=fragment,
state=new_revision_state
)
new_revision.save()
new_revision.set_tags(tags_json)
if new_revision.state == AnnotationRevision.ACCEPTED:
self.current_revision = new_revision
self.save()
iconolab_signals.revision_created.send(
sender=AnnotationRevision, instance=new_revision)
return new_revision
@transaction.atomic
def validate_existing_revision(self, revision_to_validate):
"""
Called when we're validating an awaiting revision whose parent is
the current revision AS IT WAS CREATED
"""
if revision_to_validate.parent_revision == self.current_revision and revision_to_validate.state == AnnotationRevision.AWAITING:
self.current_revision = revision_to_validate
revision_to_validate.state = AnnotationRevision.ACCEPTED
revision_to_validate.save()
self.save()
iconolab_signals.revision_accepted.send(
sender=AnnotationRevision, instance=revision_to_validate)
@transaction.atomic
def reject_existing_revision(self, revision_to_reject):
"""
Called when we reject a revision
"""
if revision_to_reject.state == AnnotationRevision.AWAITING:
revision_to_reject.state = AnnotationRevision.REJECTED
revision_to_reject.save()
iconolab_signals.revision_rejected.send(
sender=AnnotationRevision, instance=revision_to_reject)
@transaction.atomic
def merge_existing_revision(self, title, description, fragment, tags, revision_to_merge):
"""
Called when we're validating an awaiting revision whose parent isn't
the current revision or if the awaiting revision was modified by the annotation author
"""
merged_revision = self.make_new_revision(
author=self.author, title=title, description=description, fragment=fragment, tags_json=tags)
merged_revision.merge_parent_revision = revision_to_merge
merged_revision.save()
revision_to_merge.state = AnnotationRevision.STUDIED
revision_to_merge.save()
iconolab_signals.revision_accepted.send(
sender=AnnotationRevision, instance=revision_to_merge)
self.current_revision = merged_revision
self.save()
return merged_revision
class AnnotationStats(models.Model):
"""
Stats objects for a given annotation, keep count of several values to be
displayed in annotation pages
"""
annotation = models.OneToOneField(
'Annotation', related_name='stats', blank=False, null=False, on_delete=models.CASCADE)
submitted_revisions_count = models.IntegerField(
blank=True, null=True, default=1)
awaiting_revisions_count = models.IntegerField(
blank=True, null=True, default=0)
accepted_revisions_count = models.IntegerField(
blank=True, null=True, default=1)
contributors_count = models.IntegerField(blank=True, null=True, default=1)
contributors = models.ManyToManyField(User, related_name='+')
views_count = models.IntegerField(blank=True, null=True, default=0)
comments_count = models.IntegerField(blank=True, null=True, default=0)
tag_count = models.IntegerField(blank=True, null=True, default=0)
relevant_tags_count = models.IntegerField(blank=True, null=True, default=0)
accurate_tags_count = models.IntegerField(blank=True, null=True, default=0)
metacategories = models.ManyToManyField(
'MetaCategory',
through='MetaCategoriesCountInfo',
through_fields=('annotation_stats_obj', 'metacategory')
)
def __str__(self):
return "stats:for:" + str(self.annotation.annotation_guid)
@cached_property
def contributors_list(self):
user_ids_list = self.annotation.revisions.filter(
state__in=[AnnotationRevision.ACCEPTED, AnnotationRevision.STUDIED]
).values_list("author__id", flat=True)
return User.objects.filter(id__in=user_ids_list).distinct()
@cached_property
def commenters(self):
user_ids_list = IconolabComment.objects.filter(
content_type__app_label="iconolab",
content_type__model="annotation",
object_pk=self.annotation.id
).values_list("user__id", flat=True)
return User.objects.filter(id__in=user_ids_list).distinct()
def set_tags_stats(self):
self.tag_count = Tag.objects.filter(
tagginginfo__revision=self.annotation.current_revision).distinct().count()
self.relevant_tags_count = self.relevant_tags_count_calc()
self.accurate_tags_count = self.accurate_tags_count_calc()
def relevant_tags_count_calc(self, score=settings.RELEVANT_TAGS_MIN_SCORE):
return TaggingInfo.objects.filter(revision=self.annotation.current_revision, relevancy__gte=score).distinct().count()
def accurate_tags_count_calc(self, score=settings.ACCURATE_TAGS_MIN_SCORE):
return TaggingInfo.objects.filter(revision=self.annotation.current_revision, accuracy__gte=score).distinct().count()
@transaction.atomic
def update_stats(self):
# views_count - Can't do much about views count
# submitted_revisions_count
annotation_revisions = self.annotation.revisions
self.submitted_revisions_count = annotation_revisions.count()
# aawaiting_revisions_count
self.awaiting_revisions_count = annotation_revisions.filter(
state=AnnotationRevision.AWAITING).count()
# accepted_revisions_count
self.accepted_revisions_count = annotation_revisions.filter(state=AnnotationRevision.ACCEPTED).count(
) + annotation_revisions.filter(state=AnnotationRevision.STUDIED).count()
# comment_count
self.comments_count = XtdComment.objects.for_app_models("iconolab.annotation").filter(
object_pk=self.annotation.pk,
).count()
# contributors
contrib_list = self.contributors_list
self.contributors.set(contrib_list)
self.contributors_count = len(contrib_list)
# tag_count
annotation_comments_with_metacategories = IconolabComment.objects.filter(
content_type__app_label="iconolab",
content_type__model="annotation",
object_pk=self.annotation.id,
metacategories__collection=self.annotation.image.item.collection
)
m2m_objects = MetaCategoriesCountInfo.objects.filter(
annotation_stats_obj=self)
for obj in m2m_objects.all():
obj.count = 0
obj.save()
for comment in annotation_comments_with_metacategories.all():
for metacategory in comment.metacategories.all():
if metacategory not in self.metacategories.all():
MetaCategoriesCountInfo.objects.create(
annotation_stats_obj=self, metacategory=metacategory, count=1)
else:
m2m_object = MetaCategoriesCountInfo.objects.filter(
annotation_stats_obj=self, metacategory=metacategory).first()
m2m_object.count += 1
m2m_object.save()
self.set_tags_stats()
self.save()
class MetaCategoriesCountInfo(models.Model):
"""
M2M class to keep a count of a given metacategory on a given annotation.
Metacategories are linked to comments, themselve linked to an annotation
"""
annotation_stats_obj = models.ForeignKey(
'AnnotationStats', on_delete=models.CASCADE, related_name='metacategoriescountinfos')
metacategory = models.ForeignKey('MetaCategory', on_delete=models.CASCADE)
count = models.IntegerField(default=1, blank=False, null=False)
def __str__(self):
return "metacategory_count_for:" + self.metacategory.label + ":on:" + str(self.annotation_stats_obj.annotation.annotation_guid)
class AnnotationRevision(models.Model):
"""
AnnotationRevisions objects are linked to an annotation and store the data of the annotation at a given time
A revision is always in one out of multiple states:
- Awaiting: the revision has been submitted but must be validated by the
original author of the related annotation
- Accepted: the revision has been accepted *as-is* by the author of the
related annotation (this happens automatically if the revision is created
by the author of the annotation)
- Rejected: the revision has been rejected by the author of the related annotation
- Studied: the revision has been studied by the author of the related
annotation and was either modified or at the very least compared with the current state
through the merge interface, thus creating a new revision merging the
current state with the proposal. At this point the proposal is flagged
as "studied" to show that the author of the original annotation has considered it
"""
AWAITING = 0
ACCEPTED = 1
REJECTED = 2
STUDIED = 3
REVISION_STATES = (
(AWAITING, 'awaiting'),
(ACCEPTED, 'accepted'),
(REJECTED, 'rejected'),
(STUDIED, 'studied'),
)
revision_guid = models.UUIDField(default=uuid.uuid4)
annotation = models.ForeignKey(
'Annotation', related_name='revisions', null=False, blank=False, on_delete=models.CASCADE)
parent_revision = models.ForeignKey(
'AnnotationRevision', related_name='child_revisions', blank=True, null=True, on_delete=models.SET_NULL)
merge_parent_revision = models.ForeignKey(
'AnnotationRevision', related_name='child_revisions_merge', blank=True, null=True, on_delete=models.SET_NULL)
author = models.ForeignKey(User, null=True, on_delete=models.PROTECT)
title = models.CharField(max_length=255)
description = models.TextField(null=True)
fragment = models.TextField()
tags = models.ManyToManyField(
'Tag', through='TaggingInfo', through_fields=('revision', 'tag'))
state = models.IntegerField(choices=REVISION_STATES, default=AWAITING)
created = models.DateTimeField(auto_now_add=True, null=True)
def __str__(self):
return str(self.revision_guid) + ":" + self.title
def set_tags(self, tags_json_string):
"""
This method creates tags object and links them to the revision, from a given json that has the following format:
[
{
"tag_input": the tag string that has been provided. If it is
an http(s?):// pattern, it means the tag is external, else
it means it is a custom tag
"accuracy": the accuracy value provided by the user
"relevancy": the relevancy value provided by the user
},
{
...
}
]
"""
try:
tags_dict = json.loads(tags_json_string)
except ValueError:
pass
for tag_data in tags_dict:
tag_string = tag_data.get("tag_input")
tag_label = tag_data.get("tag_label")
tag_accuracy = tag_data.get("accuracy", 0)
tag_relevancy = tag_data.get("relevancy", 0)
# check if url
if tag_string.startswith("http://") or tag_string.startswith("https://"):
# check if tag already exists
if Tag.objects.filter(link=tag_string).exists():
tag_obj = Tag.objects.get(link=tag_string)
else:
tag_obj = Tag.objects.create(
link=tag_string,
label=tag_label
)
else:
new_tag_link = settings.BASE_URL + '/' + slugify(tag_string)
if Tag.objects.filter(link=new_tag_link).exists():
# Somehow we received a label for an existing tag
tag_obj = Tag.objects.get(link=new_tag_link)
else:
tag_obj = Tag.objects.create(
label=tag_string,
label_slug=slugify(tag_string),
description="",
link=settings.INTERNAL_TAGS_URL +
'/' + slugify(tag_string),
collection=self.annotation.image.item.collection
)
tag_info = TaggingInfo.objects.create(
tag=tag_obj,
revision=self,
accuracy=tag_accuracy,
relevancy=tag_relevancy
)
# FIXME Avoid calling DBPedia all the time
def get_tags_json(self):
"""
This method returns the json data that will be sent to the js to display
tags for the revision.
The json data returned will be of the following format:
[
{
"tag_label": the tag label for display purposes,
"tag_link": the link of the tag, for instance for dbpedia links,
"accuracy": the accuracy value of the tag,
"relevancy": the relevancy value of the tag,
"is_internal": will be True if the tag is 'internal',
meaning specific to Iconolab and
not an external tag like a dbpedia reference for instance
},
{
...
}
]
"""
def fetch_from_dbpedia(uri, lang, source):
sparql_template = 'select distinct * where { <<%uri%>> rdfs:label ?l FILTER( langMatches( lang(?l), "<%lang%>" ) ) }'
sparql_query = re.sub("<%uri%>", uri, re.sub(
"<%lang%>", lang, sparql_template))
sparql_query_url = source + 'sparql'
try:
dbpedia_resp = requests.get(
sparql_query_url,
params={
"query": sparql_query,
"format": "json"
}
)
except:
# dbpedia is down, will be handled with database label
pass
try:
results = json.loads(dbpedia_resp.text).get("results", {})
except:
# if error with json, results is empty
results = {}
variable_bindings = results.get("bindings", None)
label_data = {}
if variable_bindings:
label_data = variable_bindings.pop()
return label_data.get("l", {"value": False}).get("value")
final_list = []
for tagging_info in self.tagginginfo_set.select_related("tag").all():
if tagging_info.tag.is_internal():
final_list.append({
"tag_label": tagging_info.tag.label,
"tag_link": tagging_info.tag.link,
"accuracy": tagging_info.accuracy,
"relevancy": tagging_info.relevancy,
"is_internal": tagging_info.tag.is_internal()
})
else:
tag_link = tagging_info.tag.link
# import label from external
externaL_repos_fetch_dict = {
"http://dbpedia.org/": fetch_from_dbpedia,
"http://fr.dbpedia.org/": fetch_from_dbpedia
}
try:
(source, fetch_label) = next(
item for item in externaL_repos_fetch_dict.items() if tag_link.startswith(item[0]))
tag_label = fetch_label(tag_link, "fr", source)
if not tag_label: # Error happened and we got False as a fetch return
tag_label = tagging_info.tag.label
else:
tagging_info.tag.label = tag_label
tagging_info.tag.save()
final_list.append({
"tag_label": tag_label,
"tag_link": tag_link,
"accuracy": tagging_info.accuracy,
"relevancy": tagging_info.relevancy,
"is_internal": tagging_info.tag.is_internal()
})
except StopIteration:
pass
return json.dumps(final_list)
class Tag(models.Model):
"""
Tag objects that are linked to revisions.
Each tag is linked to a specific collection, this is important for internal tags
so each collection can build its own vocabulary
"""
label = models.CharField(max_length=255, blank=True, null=True)
label_slug = models.SlugField(blank=True, null=True)
link = models.URLField(unique=True)
description = models.CharField(max_length=255, blank=True, null=True)
collection = models.ForeignKey('Collection', blank=True, null=True, on_delete=models.PROTECT)
def is_internal(self):
return self.link.startswith(settings.INTERNAL_TAGS_URL)
def __str__(self):
return "Tag:" + str(self.label)
class TaggingInfo(models.Model):
"""
M2M object for managing tag relation to a revision with its associated
relevancy and accuracy
"""
revision = models.ForeignKey(
'AnnotationRevision', on_delete=models.CASCADE)
tag = models.ForeignKey('Tag', on_delete=models.CASCADE)
accuracy = models.IntegerField(db_index=True)
relevancy = models.IntegerField()
def __str__(self):
return str(str(self.tag.label) + ":to:" + str(self.revision.revision_guid))
class IconolabComment(XtdComment):
"""
Comment objects that extends XtdComment model, itself extending the
django-contrib-comments model.
Each comment can have 0 or 1 revision, if it is a comment created alongside a revision
Each comment can have a set of metacategories
"""
revision = models.ForeignKey(
'AnnotationRevision', related_name='creation_comment', null=True, blank=True, on_delete=models.PROTECT)
metacategories = models.ManyToManyField(
'MetaCategory', through='MetaCategoryInfo', through_fields=('comment', 'metacategory'))
objects = XtdComment.objects
def __str__(self):
return str(self.id)
class Meta:
ordering = ["thread_id", "id"]
@property
def annotation(self):
if self.content_type.app_label == "iconolab" and self.content_type.model == "annotation":
return Annotation.objects.get(pk=self.object_pk)
return None
def get_comment_page(self):
"""
Shortcut function to get page for considered comment, with
COMMENTS_PER_PAGE_DEFAULT comments per page, used for notifications links generation
"""
return (IconolabComment.objects.for_app_models("iconolab.annotation").filter(
object_pk=self.object_pk,
).filter(thread_id__gte=self.thread_id).filter(order__lte=self.order).count() + 1) // settings.COMMENTS_PER_PAGE_DEFAULT + 1
class MetaCategory(models.Model):
"""
Metacategories are objects that can be linked to a comment to augment it
with meaning (depending on the metacategories defined for a given collection)
Metacategories can trigger notifications when they are linked to a given
coment depending on their trigger_notifications property:
- NONE : Notifies nobody
- CONTRIBUTORS : Notifies contributors (revision owners) on target annotation
- COMMENTERS : Notifies commenters (contributors + comment owners) on target annotation
- COLLECTION_ADMINS : Notifies collection admins
Metacategories can be used to consider an annotation as "validated" if a
certain agreement threshold is reached using their validation_value property
- NEUTRAL : The metacategory doesn't affect the validation state
- AGREEMENT : The metacategory can be used to validate the annotation
when linked to a comment on said annotation
- DISAGREEMENT : The metacategory can be used to unvalidate the annotation
when linked to a comment on said annotation
"""
NONE = 0
CONTRIBUTORS = 1
COMMENTERS = 2
COLLECTION_ADMINS = 3
NOTIFIED_USERS = (
(NONE, 'none'),
(CONTRIBUTORS, 'contributors'),
(COMMENTERS, 'commenters'),
(COLLECTION_ADMINS, 'collection admins'),
)
NEUTRAL = 0
AGREEMENT = 1
DISAGREEMENT = 2
VALIDATION_VALUES = (
(NEUTRAL, 'neutral'),
(AGREEMENT, 'agreement'),
(DISAGREEMENT, 'disagreement'),
)
class Meta:
unique_together = (("collection", "label"),)
collection = models.ForeignKey(Collection, related_name="metacategories", on_delete=models.PROTECT)
label = models.CharField(max_length=255)
triggers_notifications = models.IntegerField(
choices=NOTIFIED_USERS, default=NONE)
validation_value = models.IntegerField(
choices=VALIDATION_VALUES, default=NEUTRAL)
def __str__(self):
return self.label + ":" + self.collection.name
class MetaCategoryInfo(models.Model):
"""
M2M class linking comments and metacategories
"""
comment = models.ForeignKey('IconolabComment', on_delete=models.CASCADE)
metacategory = models.ForeignKey('MetaCategory', on_delete=models.CASCADE)
def __str__(self):
return "metacategory:" + self.metacategory.label + ":on:" + self.comment.id
class CommentAttachement(models.Model):
"""
This class is supposed to represent added resources to a given comment
Not implemented as of v0.0.19
"""
LINK = 0
IMAGE = 1
PDF = 2
COMMENT_CHOICES = (
(LINK, 'link'),
(IMAGE, 'image'),
(PDF, 'pdf')
)
comment = models.ForeignKey(
'IconolabComment', related_name='attachments', on_delete=models.CASCADE)
attachment_type = models.IntegerField(choices=COMMENT_CHOICES, default=0)
data = models.TextField(blank=False)
class UserProfile(models.Model):
"""
UserProfile objects are extensions of user model
As of v0.0.19 they are used to define collection admins. Each user can
thus managed 0-N collections.
"""
user = models.OneToOneField(
User, related_name='profile', on_delete=models.CASCADE)
managed_collections = models.ManyToManyField(
'Collection', related_name='admins', blank=True)
def __str__(self):
return "profile:" + self.user.username
class BookmarkCategory(models.Model):
user = models.ForeignKey(User, on_delete=models.PROTECT)
name = models.CharField(max_length=255)
created = models.DateTimeField(auto_now_add=True)
class Bookmark(models.Model):
category = models.ForeignKey(BookmarkCategory, on_delete=models.PROTECT)
image = models.ForeignKey(Image, on_delete=models.PROTECT)
created = models.DateTimeField(auto_now_add=True)